按需评估入门

按照以下步骤设置和运行您的首次按需评估。

先决条件

要使用 AgentCore 评估 OnDemand 评估功能，您需要：

支持的框架

AgentCore 评估目前支持以下代理框架和仪器库：

步骤 1：创建和部署代理

按照 AgentCore 运行时入门指南创建和部署代理。您可以在AgentCore 评估样本中找到其他示例。

第 2 步：调用您的代理

使用以下命令调用您的代理，并在上的 GenAI 可观察性仪表板上查看跟踪、会话和指标。 CloudWatch

示例 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)

步骤 3：评估代理

一旦你对代理进行了几次调用，就可以对其进行评估了。为了进行评估，我们需要：

AgentCore CLI 和 AgentCore SDK 的代码示例

以下代码示例演示了如何使用不同的开发方法进行按需评估。选择最适合您的开发环境和偏好的方法。

AWS SDK

从中下载 span-log CloudWatch

在调用 Evaluate API 之前，您需要从中下载跨度日志 CloudWatch。你可以使用下面的 Python 代码来执行此操作，也可以选择将它们保存在 JSON 文件中。这样可以更轻松地向不同的评估者提出同一个会话的请求。

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)

致电评估

有了输入跨度后，就可以调用 Evaluate API 了。请注意，回复可能需要一点时间，因为大型语言模型正在对你的踪迹进行评分。

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

如果你在上面使用并将会话跨度转储到一个 json 文件中，你也可以随后运行评估，如下所示

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

使用评估目标

要评估会话中的特定跟踪或工具，您可以使用请求中的evaluationTarget参数指定目标。

Session-level 评估者

由于该服务每次评估仅支持一个会话，因此您无需明确设置评估目标。

Trace-level 评估者

对于跟踪级别的赋值器（例如Builtin.Helpfulness或Builtin.Correctness），请在参数中设置跟踪 ID：evaluationTarget

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

工具调用等级评估器

对于跨度级别赋值器（例如Builtin.ToolSelectionAccuracy），请在参数中设置跨度 ID：evaluationTarget

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

步骤 4：评估结果

每个 Evaluate API 调用都会返回一个包含评估器结果列表的响应。由于单个会话可能包含多条轨迹和工具调用，因此这些元素将作为单独的实体进行评估。因此，单个 API 调用可能会返回多个评估结果。

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

结果限制

每次 API 调用返回的评估数量限制为 10 个结果。例如，如果您使用跟踪级别赋值器评估包含 15 条轨迹的会话，则响应最多包含 10 个结果。默认情况下，API 会返回最后 10 次评估，因为这些评估通常包含与评估质量最相关的上下文。

部分故障

一个 API 调用可能处理 n 个评估，而其中 m 个评估失败。故障可能由于各种原因而发生，包括：

在部分失败的情况下，响应包括成功和失败的评估。失败的结果包括错误代码和错误消息，可帮助您诊断问题。

跨度上下文

每个赋值器结果都有一个spanContext字段，用于标识所评估的实体：

成功输入结果示例

这只是一个条目。如果一个会话有多条跟踪，你将看到多个这样的条目，每条记录一个。同样，对于工具级赋值器，如果有多个工具调用并且提供了一个工具赋值器（例如Builtin.ToolSelectionAccuracy），则每个工具跨度将有一个结果。

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

结果输入失败示例

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