Commencer à utiliser l'évaluation à la demande

Suivez ces étapes pour configurer et exécuter votre première évaluation à la demande.

Conditions préalables

Pour utiliser les fonctionnalités OnDemand d'évaluation des AgentCore évaluations, vous devez :

Cadres pris en charge

AgentCore Les évaluations prennent actuellement en charge les cadres agentiques et les bibliothèques d'instruments suivants :

Étape 1 : créer et déployer votre agent

Créez et déployez votre agent en suivant le guide de démarrage pour AgentCore Runtime. Vous trouverez des exemples supplémentaires dans les exemples d'AgentCore évaluations.

Étape 2 : Invoquez votre agent

Appelez votre agent à l'aide de la commande suivante et consultez les traces, les sessions et les métriques sur le tableau de bord GenAI Observability sur. CloudWatch

Exemple 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)

Étape 3 : Evaluer l'agent

Une fois que vous avez fait quelques invocations à votre agent, vous êtes prêt à l'évaluer. Pour les évaluations, nous avons besoin de :

Exemples de code pour la AgentCore CLI et le AgentCore SDK

Les exemples de code suivants montrent comment exécuter des évaluations à la demande en utilisant différentes approches de développement. Choisissez la méthode qui correspond le mieux à votre environnement de développement et à vos préférences.

AWS Kit SDK

Téléchargez les span-logs depuis CloudWatch

Avant d'appeler l'EvaluateAPI, vous devez télécharger les journaux d'extension depuis CloudWatch. Pour ce faire, vous pouvez utiliser le code Python ci-dessous et éventuellement les enregistrer dans un fichier JSON. Il est ainsi plus facile de faire la demande pour la même session auprès de différents évaluateurs.

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)

Appelez Evaluate

Une fois que vous avez les plages d'entrée, vous pouvez appeler l'EvaluateAPI. Veuillez noter que les réponses peuvent prendre quelques instants, car un grand modèle linguistique marque vos traces.

# 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"])

Si vous utilisez ci-dessus et que vous enregistrez les durées de session dans un fichier json, vous pouvez également exécuter ensuite evaluate comme ci-dessous

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"])

Utilisation des cibles d'évaluation

Pour évaluer une trace ou un outil spécifique au cours d'une session, vous pouvez spécifier la cible à l'aide du evaluationTarget paramètre de votre demande.

Session-level évaluateur

Comme le service ne prend en charge qu'une seule session par évaluation, il n'est pas nécessaire de définir explicitement l'objectif d'évaluation.

Trace-level évaluateur

Pour les évaluateurs au niveau de la trace (tels que Builtin.Helpfulness ouBuiltin.Correctness), définissez les ID de trace dans le paramètre : evaluationTarget

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

Outil : évaluateur du niveau d'appel

Pour les évaluateurs au niveau de la plage (tels queBuiltin.ToolSelectionAccuracy), définissez les ID d'étendue dans le paramètre : evaluationTarget

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

Étape 4 : Résultats de l'évaluation

Chaque appel Evaluate d'API renvoie une réponse contenant une liste des résultats de l'évaluateur. Comme une seule session peut inclure plusieurs traces et appels d'outils, ces éléments sont évalués en tant qu'entités distinctes. Par conséquent, un seul appel d'API peut renvoyer plusieurs résultats d'évaluation.

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

Limite de résultats

Le nombre d'évaluations renvoyées par appel d'API est limité à 10 résultats. Par exemple, si vous évaluez une session contenant 15 traces à l'aide d'un évaluateur au niveau des traces, la réponse inclut un maximum de 10 résultats. Par défaut, l'API renvoie les 10 dernières évaluations, car celles-ci contiennent généralement le contexte le plus pertinent pour la qualité des évaluations.

Défaillances partielles

Un appel d'API peut traiter n évaluations alors que m d'entre elles échouent. Les défaillances peuvent survenir pour diverses raisons, notamment :

En cas d'échec partiel, la réponse inclut à la fois des évaluations réussies et des évaluations infructueuses. Les résultats d'échec incluent un code d'erreur et un message d'erreur pour vous aider à diagnostiquer le problème.

Contexte de Span

Chaque résultat de l'évaluateur comporte un spanContext champ qui identifie l'entité évaluée :

Exemple de saisie de résultats réussie

Il ne s'agit que d'une entrée. Si une session comporte plusieurs traces, vous verrez plusieurs entrées de ce type, une pour chaque trace. De même, pour les évaluateurs au niveau de l'outil, s'il y a plusieurs appels d'outils et qu'un évaluateur d'outil (tel queBuiltin.ToolSelectionAccuracy) est fourni, il y aura un résultat par plage d'outils.

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

Exemple d'échec de la saisie des résultats

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