As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

Começando com a avaliação sob demanda

Siga estas etapas para configurar e executar sua primeira avaliação sob demanda.

Pré-requisitos

Para usar os recursos de AgentCore OnDemand avaliação, você precisa:

Frameworks compatíveis

Crie seu agente com uma biblioteca de estrutura e instrumentação compatível com o AgentCore Evaluations. Para obter mais informações sobre estruturas compatíveis e bibliotecas de instrumentação, consulte Estruturas de agentes suportadas.

Etapa 1: criar e implantar seu agente

Crie e implante seu agente seguindo o guia de introdução ao AgentCore Runtime. Você pode encontrar exemplos adicionais nas amostras de AgentCore avaliações.

Etapa 2: invocar seu agente

Chame seu agente usando o comando a seguir e visualize os rastreamentos, sessões e métricas no painel GenAI Observability em. CloudWatch

Exemplo de invoke_agent.py

import boto3
import json
import uuid

region = "region-code"
ace_demo_agent_arn = "agent-arn from step-2"

agent_core_client = boto3.client('bedrock-agentcore', region_name=region)

text_to_analyze = "Sample text to test agent for agentcore evaluations demo"

payload = json.dumps({
    "prompt": f"Can you analyze this text and tell me about its statistics: {text_to_analyze}"
})

# random session-id, you can set your own here
session_id = "test-ace-demo-session-18a1dba0-62a0-462g"

response = agent_core_client.invoke_agent_runtime(
    agentRuntimeArn=ace_demo_agent_arn,
    runtimeSessionId=session_id,
    payload=payload,
    qualifier="DEFAULT"
)

response_body = response['response'].read()
response_data = json.loads(response_body)
print("Agent Response:", response_data)
print("SessionId:", session_id)

Etapa 3: avaliar o agente

Depois de fazer algumas invocações para seu agente, você estará pronto para avaliá-las. Para avaliações, exigimos:

Amostras de código para AgentCore CLI e SDK AgentCore

Os exemplos de código a seguir demonstram como executar avaliações sob demanda usando diferentes abordagens de desenvolvimento. Escolha o método mais adequado ao seu ambiente de desenvolvimento e às suas preferências.

AWS SDK

Baixe span-logs de CloudWatch

Antes de chamar a Evaluate API, você precisa baixar os logs de span do CloudWatch. Você pode usar o código Python abaixo para fazer isso e, opcionalmente, salvá-los em um arquivo JSON. Isso facilita a solicitação da mesma sessão com avaliadores diferentes.

import boto3
import time
import json
from datetime import datetime, timedelta

region = "region-code"
agent_id = "agent-id-from-step-2"
session_id = "session-id-from-step-3"

def query_logs(log_group_name, query_string):
    client = boto3.client('logs', region_name=region)
    start_time = datetime.now() - timedelta(minutes=60) # past 1 hour
    end_time = datetime.now()

    query_id = client.start_query(
        logGroupName=log_group_name,
        startTime=int(start_time.timestamp()),
        endTime=int(end_time.timestamp()),
        queryString=query_string
    )['queryId']

    while (result := client.get_query_results(queryId=query_id))['status'] not in ['Complete', 'Failed']:
        time.sleep(1)

    if result['status'] == 'Failed':
        raise Exception("Query failed")
    return result['results']

def query_session_logs(log_group_name, session_id, **kwargs):
    query = f"""fields @timestamp, @message
    | filter ispresent(scope.name) and ispresent(attributes.session.id)
    | filter attributes.session.id = "{session_id}"
    | sort @timestamp asc"""
    return query_logs(log_group_name, query, **kwargs)

def query_agent_runtime_logs(agent_id, endpoint, session_id, **kwargs):
    return query_session_logs(
        f"/aws/bedrock-agentcore/runtimes/{agent_id}-{endpoint}",
        session_id, **kwargs)

def query_aws_spans_logs(session_id, **kwargs):
    return query_session_logs("aws/spans", session_id, **kwargs)

def extract_messages_as_json(query_results):
    return [json.loads(f['value']) for row in query_results
            for f in row if f['field'] == '@message'
            and f['value'].strip().startswith('{')]

def get_session_span_logs():
    agent_runtime_logs = query_agent_runtime_logs(
        agent_id=agent_id, endpoint="DEFAULT", session_id=session_id
    )
    print(f"Downloaded {len(agent_runtime_logs)} runtime-log entries")

    aws_span_logs = query_aws_spans_logs(session_id=session_id)
    print(f"Downloaded {len(aws_span_logs)} aws/span entries")

    session_span_logs = extract_messages_as_json(aws_span_logs) + extract_messages_as_json(agent_runtime_logs)
    print(f"Returning {len(aws_span_logs) + len(agent_runtime_logs)} total records")
    return session_span_logs

# get the spans from cloudwatch
session_span_logs = get_session_span_logs()

# optional (dump in a json file for reuse)
session_span_logs_file_name = "ace-demo-session.json"
with open(session_span_logs_file_name, "w") as f:
    json.dump(session_span_logs, f, indent=2)

Avaliação de chamadas

Depois de ter os intervalos de entrada, você pode invocar a Evaluate API. Observe que as respostas podem demorar alguns instantes, pois um grande modelo de linguagem está pontuando seus rastros.

# initialise client
ace_dp_client = boto3.client('bedrock-agentcore', region_name = region)

# call evaluate
response = ace_dp_client.evaluate(
    evaluatorId = "Builtin.Helpfulness", # can be a custom evaluator id as well
    evaluationInput = {"sessionSpans": session_span_logs})

print(response["evaluationResults"])

Se você usar acima e despejar os períodos de sessão em um arquivo json, você também pode executar o evaluate posteriormente conforme abaixo

with open(session_span_logs_file_name, "r") as f:
    session_span_logs = json.load(f)

# initialise client
ace_dp_client = boto3.client('bedrock-agentcore', region_name = region)

# call evaluate
response = ace_dp_client.evaluate(
    evaluatorId = "Builtin.ToolSelectionAccuracy", # can be a custom evaluator id as well
    evaluationInput = {"sessionSpans": session_span_logs})

print(response["evaluationResults"])

Usando metas de avaliação

Para avaliar um rastreamento ou uma ferramenta específica em uma sessão, você pode especificar o alvo usando o evaluationTarget parâmetro em sua solicitação.

Session-level avaliador

Como o serviço oferece suporte a apenas uma sessão por avaliação, você não precisa definir explicitamente a meta de avaliação.

Trace-level avaliador

Para avaliadores em nível de rastreamento (como Builtin.Helpfulness ouBuiltin.Correctness), defina os IDs de rastreamento no parâmetro: evaluationTarget

response = ace_dp_client.evaluate(
    evaluatorId = "Builtin.Helpfulness",
    evaluationInput = {"sessionSpans": session_span_logs},
    evaluationTarget = {"traceIds": ["trace-id-1", "trace-id-2"]}
)

Avaliador do nível de chamada da ferramenta

Para avaliadores de nível de extensão (comoBuiltin.ToolSelectionAccuracy), defina os IDs de extensão no parâmetro: evaluationTarget

response = ace_dp_client.evaluate(
    evaluatorId = "Builtin.ToolSelectionAccuracy",
    evaluationInput = {"sessionSpans": session_span_logs},
    evaluationTarget = {"spanIds": ["span-id-1", "span-id-2"]}
)

Etapa 4: resultados da avaliação

Cada chamada de Evaluate API retorna uma resposta contendo uma lista dos resultados do avaliador. Como uma única sessão pode incluir vários rastreamentos e chamadas de ferramentas, esses elementos são avaliados como entidades separadas. Consequentemente, uma única chamada de API pode retornar vários resultados de avaliação.

{
    "evaluationResults": [ {evaluation-result-1}, {evaluation-result_2},.... ]
}

Limite de resultados

O número de avaliações retornadas por chamada de API é limitado a 10 resultados. Por exemplo, se você avaliar uma sessão contendo 15 rastreamentos usando um avaliador em nível de rastreamento, a resposta incluirá no máximo 10 resultados. Por padrão, a API retorna as últimas 10 avaliações, pois elas geralmente contêm o contexto mais relevante para a qualidade da avaliação.

Falhas parciais

Uma chamada de API pode processar n avaliações enquanto muitas delas falham. As falhas podem ocorrer devido a vários motivos, incluindo:

Em casos de falha parcial, a resposta inclui avaliações bem-sucedidas e fracassadas. Os resultados falhados incluem um código de erro e uma mensagem de erro para ajudar você a diagnosticar o problema.

Contexto da extensão

Cada resultado do avaliador tem um spanContext campo que identifica a entidade avaliada:

Exemplo de entrada de resultados bem-sucedida

Essa é apenas uma entrada. Se uma sessão tiver vários rastreamentos, você verá várias dessas entradas, uma para cada rastreamento. Da mesma forma, para avaliadores em nível de ferramenta, se houver várias chamadas de ferramentas e um avaliador de ferramentas (comoBuiltin.ToolSelectionAccuracy) for fornecido, haverá um resultado por intervalo de ferramenta.

{
  "evaluatorArn": "arn:aws:bedrock-agentcore:::evaluator/Builtin.Helpfulness",
  "evaluatorId": "Builtin.Helpfulness",
  "evaluatorName": "Builtin.Helpfulness",
  "explanation": ".... evaluation explanation will be added here ...",
  "context": {
    "spanContext": {
      "sessionId": "test-ace-demo-session-18a1dba0-62a0-462e",
      "traceId": "....trace_id......."
    }
  },
  "value": 0.83,
  "label": "Very Helpful",
  "tokenUsage": {
    "inputTokens": 958,
    "outputTokens": 211,
    "totalTokens": 1169
  }
}

Exemplo de falha na entrada de resultados

{
    "evaluatorArn": "arn:aws:bedrock-agentcore:::evaluator/Builtin.Helpfulness",
    "evaluatorId": "Builtin.Helpfulness",
    "evaluatorName": "Builtin.Helpfulness",
    "context": {
        "spanContext": {
            "sessionId": "test-ace-demo-session-18a1dba0-62a0-462e",
            "traceId": "....trace_id......."
        }
    },
    "errorMessage": ".... details of the error....",
    "errorCode": ".... name/code of the error...."
}