Introdução à avaliação sob demanda

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

Pré-requisitos

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

Frameworks compatíveis

AgentCore Atualmente, as avaliações oferecem suporte às seguintes estruturas de agentes e bibliotecas de instrumentação:

Etapa 1: criar e implantar seu agente

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

Etapa 2: invocar seu agente

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

Exemplo 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 ao seu agente, você estará pronto para avaliá-las. Para avaliações, exigimos:

Exemplos 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 registros 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 levar alguns minutos, 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 a opção acima e despejar os períodos de sessão em um arquivo json, também poderá executar subseqüentemente a avaliação 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 ferramenta específica em uma sessão, você pode especificar o destino 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 da 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 em nível de intervalo (comoBuiltin.ToolSelectionAccuracy), defina os IDs de intervalo 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 traços usando um avaliador de 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 normalmente contêm o contexto mais relevante para a qualidade da avaliação.

Falhas parciais

Uma chamada de API pode processar uma avaliação enquanto minha falha. 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 ajudá-lo 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 resultado 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 ferramenta (comoBuiltin.ToolSelectionAccuracy) for fornecido, haverá um resultado por extensão 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 resultado

{
    "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...."
}