View a markdown version of this page

On-demand 数据集运行器 - Amazon Bedrock AgentCore
工作原理代理调用者示例配置参考结果结构

On-demand 数据集运行器

它在OnDemandEvaluationDatasetRunner客户端协调整个评估生命周期：调用代理、等待遥测摄取、从中收集跨度以及调用 Evaluate API CloudWatch，所有这些都只需一次调用即可。run()

使用按需运行器处理开发时迭代、 CI/CD 管道和小型数据集，在这些数据集中，您需要立即在响应中显示每个场景、每个评估者的详细信息。

注意

按需运行器支持所有 AgentCore 赋值器，包括跨会话、跟踪和工具调用级别的所有内置赋值器，以及自定义赋值器。无论您配置哪个赋值器，运行器都会自动处理关卡感知请求构造、批处理和地面实况映射。

工作原理

运行器分三个阶段处理场景：

  1. 调用：所有场景都使用线程池并发运行。每个场景都有一个唯一的会话 ID，并在场景中按顺序执行以保持对话上下文。

  2. 等待：可配置的延迟（默认值：180 秒） CloudWatch 允许摄取遥测数据。这种延迟只能支付一次，而不是按场景支付。

  3. 评估：从每个评估者那里收集跨度 CloudWatch 并生成评估请求。数据集中的实况字段 (expected_response,assertions,expected_trajectory) 会自动映射到正确的 API 参考输入。

代理调用者

跑步者需要一个代理调用器，一个在单回合内调用你的代理的可调用对象。调用者不受框架限制：您可以通过 boto3 invoke_agent_runtime、直接函数调用、HTTP 请求或任何其他方法调用您的代理。

import json
import boto3
from bedrock_agentcore.evaluation import AgentInvokerInput, AgentInvokerOutput

REGION       = "<region-code>"
AGENT_ARN    = "arn:aws:bedrock-agentcore:<region-code>:<account-id>:runtime/<agent-id>"
LOG_GROUP    = "/aws/bedrock-agentcore/runtimes/<agent-id>-DEFAULT"

agentcore_client = boto3.client("bedrock-agentcore", region_name=REGION)

def agent_invoker(invoker_input: AgentInvokerInput) -> AgentInvokerOutput:
    payload = invoker_input.payload
    if isinstance(payload, str):
        payload = json.dumps({"prompt": payload}).encode()
    elif isinstance(payload, dict):
        payload = json.dumps(payload).encode()

    print(f"[{invoker_input.session_id}] > sending payload: {payload.decode()}")
    response = agentcore_client.invoke_agent_runtime(
        agentRuntimeArn=AGENT_ARN,
        runtimeSessionId=invoker_input.session_id,
        payload=payload,
    )
    response_body = response["response"].read()
    print(f"[{invoker_input.session_id}] < received response: {response_body.decode()}")
    return AgentInvokerOutput(agent_output=json.loads(response_body))
字段 Type 说明

AgentInvokerInput.payload

str 或 dict

来自数据集的转弯输入。

AgentInvokerInput.session_id

str

在场景中的所有回合中都保持稳定。将其传递给您的代理以维护对话上下文。

AgentInvokerOutput.agent_output

Any

代理的回应。

示例

以下示例从 JSON 文件加载数据集并运行按需评估。有关数据集格式，请参阅数据集架构。

from bedrock_agentcore.evaluation import (
    OnDemandEvaluationDatasetRunner,
    EvaluationRunConfig,
    EvaluatorConfig,
    FileDatasetProvider,
    CloudWatchAgentSpanCollector,
)

# Load dataset from a local file (see Dataset schema for format)
dataset = FileDatasetProvider("dataset.json").get_dataset()

# Or load from the Dataset Management service
from bedrock_agentcore.evaluation import DatasetClient, DatasetManagementServiceProvider
ds_client = DatasetClient(region_name=REGION)
dataset = DatasetManagementServiceProvider(dataset_id="my-dataset-id", client=ds_client).get_dataset()

# Create span collector
span_collector = CloudWatchAgentSpanCollector(
    log_group_name=LOG_GROUP,
    region=REGION,
)

# Configure evaluators
config = EvaluationRunConfig(
    evaluator_config=EvaluatorConfig(
        evaluator_ids=[
            "Builtin.GoalSuccessRate",
            "Builtin.TrajectoryExactOrderMatch",
            "Builtin.Correctness",
            "Builtin.Helpfulness",
        ],
    ),
    evaluation_delay_seconds=180,
    max_concurrent_scenarios=5,
)

# Run
runner = OnDemandEvaluationDatasetRunner(region=REGION)
result = runner.run(
    agent_invoker=agent_invoker,
    dataset=dataset,
    span_collector=span_collector,
    config=config,
)

print(f"Completed: {len(result.scenario_results)} scenario(s)")

处理结果：

for scenario in result.scenario_results:
    print(f"\nScenario: {scenario.scenario_id} ({scenario.status})")
    if scenario.error:
        print(f"  Error: {scenario.error}")
        continue
    for evaluator in scenario.evaluator_results:
        print(f"  {evaluator.evaluator_id}:")
        for r in evaluator.results:
            print(f"    Score: {r.get('value')}, Label: {r.get('label')}")
            ignored = r.get("ignoredReferenceInputFields", [])
            if ignored:
                print(f"    Ignored fields: {ignored}")

要将结果保存到文件中，请执行以下操作：

with open("results.json", "w") as f:
    f.write(result.model_dump_json(indent=2))

配置参考

跨度收集器

AgentSpanCollector在代理调用后检索遥测跨度的。该软件开发工具包提供CloudWatchAgentSpanCollector：

from bedrock_agentcore.evaluation import CloudWatchAgentSpanCollector

span_collector = CloudWatchAgentSpanCollector(
    log_group_name="/aws/bedrock-agentcore/runtimes/<agent-id>-DEFAULT",
    region=REGION,
)

收集器查询两个 CloudWatch 日志组（aws/spans用于结构跨度，代理的日志组用于对话内容），轮询直到出现跨度，然后将它们作为平面列表返回。

评估配置 

from bedrock_agentcore.evaluation import EvaluationRunConfig, EvaluatorConfig

config = EvaluationRunConfig(
    evaluator_config=EvaluatorConfig(
        evaluator_ids=["Builtin.Correctness", "Builtin.GoalSuccessRate"],
    ),
    evaluation_delay_seconds=180,  # Wait for CloudWatch ingestion (default: 180)
    max_concurrent_scenarios=5,    # Thread pool size (default: 5)
    simulation_config=None,        # Set SimulationConfig for simulated scenarios
)
字段 默认值 说明

evaluator_config.evaluator_ids

—

评估者 ID 列表（内置名称或自定义赋值器 ID）。

evaluation_delay_seconds

180

调用后等待摄取跨 CloudWatch 度的秒数。如果使用非CloudWatch 收集器，则设置为 0。

max_concurrent_scenarios

5

并行调用和评估的最大场景数。

simulation_config

无

模拟场景的配置。设置数据集SimulationConfig(model_id="…​")何时包含SimulatedScenario实例。参见用户模拟。

结果结构

运行器返回EvaluationResult具有以下结构的：

EvaluationResult
  └── scenario_results: List[ScenarioResult]
        ├── scenario_id: str
        ├── session_id: str
        ├── status: "COMPLETED" | "FAILED"
        ├── error: Optional[str]
        └── evaluator_results: List[EvaluatorResult]
              ├── evaluator_id: str
              └── results: List[Dict]   # Raw API responses

中的每个条目results都是来自 Evaluate API 的原始响应字典，其中包含诸如value、、label、explanationcontexttokenUsage、和之类的字段ignoredReferenceInputFields。有关完整的回复格式，请参阅按需评估入门。

带有状态的场景FAILED意味着发生了结构性问题（代理调用错误、跨度收集失败）。COMPLETED场景中的单个评估者错误会记录在评估者的results列表中，errorCode并带有和errorMessage字段。