Erste Schritte mit der On-Demand-Evaluierung

Gehen Sie wie folgt vor, um Ihre erste On-Demand-Evaluierung einzurichten und durchzuführen.

Voraussetzungen

Um die OnDemand Bewertungsfunktionen von AgentCore Evaluationen nutzen zu können, benötigen Sie:

Unterstützte Frameworks

AgentCore Evaluations unterstützt derzeit die folgenden agentischen Frameworks und Instrumentierungsbibliotheken:

Schritt 1: Erstellen und implementieren Sie Ihren Agenten

Folgen Sie der Anleitung „Erste Schritte“ für AgentCore Runtime, um Ihren Agenten zu erstellen und bereitzustellen. Weitere Beispiele finden Sie in den AgentCore Evaluationsbeispielen.

Schritt 2: Rufen Sie Ihren Agenten an

Rufen Sie Ihren Agenten mit dem folgenden Befehl auf und sehen Sie sich die Traces, Sessions und Metriken im GenAI Observability Dashboard an. CloudWatch

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

Schritt 3: Evaluieren Sie den Agenten

Sobald Sie einige Anrufe an Ihren Agenten gerichtet haben, können Sie ihn bewerten. Für Bewertungen benötigen wir:

Codebeispiele für AgentCore CLI und AgentCore SDK

Die folgenden Codebeispiele zeigen, wie Sie On-Demand-Evaluierungen mit verschiedenen Entwicklungsansätzen durchführen können. Wählen Sie die Methode, die am besten zu Ihrer Entwicklungsumgebung und Ihren Präferenzen passt.

AWS SDK

Laden Sie die Span-Logs von herunter CloudWatch

Bevor Sie die Evaluate API aufrufen, müssen Sie die Span-Protokolle von herunterladen. CloudWatch Sie können dazu den folgenden Python-Code verwenden und sie optional in einer JSON-Datei speichern. Dies macht es einfacher, die Anfrage für dieselbe Sitzung mit verschiedenen Evaluatoren zu stellen.

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)

Rufen Sie Evaluate auf

Sobald Sie die Eingabebereiche haben, können Sie die Evaluate API aufrufen. Bitte beachten Sie, dass die Antworten einige Zeit in Anspruch nehmen können, da ein umfangreiches Sprachmodell Ihre Spuren auswertet.

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

Wenn Sie den obigen Befehl verwenden und die Session-Spans in einer JSON-Datei speichern, können Sie anschließend auch die Auswertung wie folgt ausführen

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

Verwenden von Bewertungszielen

Um einen bestimmten Trace oder ein bestimmtes Tool innerhalb einer Sitzung auszuwerten, können Sie das Ziel mithilfe des evaluationTarget Parameters in Ihrer Anfrage angeben.

Session-level Evaluator

Da der Service nur eine Sitzung pro Evaluierung unterstützt, müssen Sie das Bewertungsziel nicht explizit festlegen.

Trace-level Evaluator

Für Evaluatoren auf Trace-Ebene (wie Builtin.Helpfulness oderBuiltin.Correctness) legen Sie die Trace-IDs im Parameter fest: evaluationTarget

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

Evaluator auf Tool-Aufrufeebene

Für Evaluatoren auf Span-Ebene (z. B.Builtin.ToolSelectionAccuracy) legen Sie die Span-IDs im Parameter fest: evaluationTarget

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

Schritt 4: Ergebnisse der Auswertung

Jeder Evaluate API-Aufruf gibt eine Antwort zurück, die eine Liste der Evaluator-Ergebnisse enthält. Da eine einzelne Sitzung mehrere Traces und Tool-Aufrufe beinhalten kann, werden diese Elemente als separate Entitäten ausgewertet. Folglich kann ein einziger API-Aufruf mehrere Bewertungsergebnisse zurückgeben.

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

Limit für Ergebnisse

Die Anzahl der pro API-Aufruf zurückgegebenen Bewertungen ist auf 10 Ergebnisse begrenzt. Wenn Sie beispielsweise eine Sitzung mit 15 Traces mit einem Evaluator auf Trace-Ebene auswerten, umfasst die Antwort maximal 10 Ergebnisse. Standardmäßig gibt die API die letzten 10 Bewertungen zurück, da diese in der Regel den Kontext enthalten, der für die Qualität der Bewertung am relevantesten ist.

Teilweise Ausfälle

Ein API-Aufruf kann n Bewertungen verarbeiten, während meine von ihnen fehlschlagen. Fehler können aus verschiedenen Gründen auftreten, darunter:

Im Falle eines teilweisen Fehlers umfasst die Antwort sowohl erfolgreiche als auch fehlgeschlagene Bewertungen. Fehlgeschlagene Ergebnisse enthalten einen Fehlercode und eine Fehlermeldung, die Ihnen bei der Diagnose des Problems helfen.

Umfassender Kontext

Jedes Evaluator-Ergebnis hat ein spanContext Feld, das die ausgewertete Entität identifiziert:

Beispiel für eine erfolgreiche Ergebniseingabe

Dies ist nur ein Eintrag. Wenn eine Sitzung mehrere Traces hat, werden Sie mehrere solcher Einträge sehen, einen für jeden Trace. Ähnliches gilt für Evaluatoren auf Tool-Ebene: Wenn es mehrere Tool-Calls gibt und ein Tool-Evaluator (wieBuiltin.ToolSelectionAccuracy) zur Verfügung steht, wird pro Tool-Spanne ein Ergebnis angezeigt.

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

Beispiel: Fehlgeschlagene Ergebniseingabe

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