On-demand 数据集运行器

它在客户端OnDemandEvaluationDatasetRunner协调整个评估生命周期：调用代理、等待遥测数据采集、从评估API收集跨度并调用Evaluate API CloudWatch，所有这些都在一次调用中完成。run()

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

注意

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

工作原理

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

调用：所有场景都使用线程池并行运行。每个场景都有一个唯一的会话 ID，并在一个按顺序执行的场景内切换，以维护对话上下文。
等待：可配置的延迟（默认值：180 秒） CloudWatch 允许采集遥测数据。延迟一次性支付，而不是按场景支付。
评估：从每个评估者那里 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都是来自评估 API 的原始响应字典，包含value、、labelexplanationcontexttokenUsage、和等字段ignoredReferenceInputFields。有关完整的回复格式，请参阅按需评估入门。

具有状态的场景FAILED表示发生了结构性问题（代理调用错误、跨度收集失败）。COMPLETED场景中的个别评估者错误将记录在评估者的results列表中，其中包含errorCode和errorMessage字段。

Javascript 在您的浏览器中被禁用或不可用。

要使用 Amazon Web Services 文档，必须启用 Javascript。请参阅浏览器的帮助页面以了解相关说明。

文档惯例

数据集加密

批量数据集运行器