本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

# On-demand 数据集运行器
<a name="dataset-evaluations-on-demand"></a>

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

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

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

## 工作原理
<a name="ds-how-it-works"></a>

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

1.  **调用：**所有场景都使用线程池并行运行。每个场景都有一个唯一的会话 ID，并在一个按顺序执行的场景内切换，以维护对话上下文。

1.  **等待：**可配置的延迟（默认值：180 秒） CloudWatch 允许采集遥测数据。延迟一次性支付，而不是按场景支付。

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

## 代理调用者
<a name="ds-agent-invoker"></a>

跑步者需要一个代理调用器，一个在单回合内调用你的代理的可调用对象。调用者与框架无关：您可以通过 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`  | 代理的回应。 | 

## 示例
<a name="ds-example"></a>

以下示例从 JSON 文件加载数据集并运行按需评估。有关数据集格式，请参阅[数据集架构](dataset-evaluations-schema.md)。

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

## 配置参考
<a name="ds-components-reference"></a>

 **跨度收集器 ** 

`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`实例时设置。参见[用户模拟](user-simulation.md)。 | 

## 结果结构
<a name="ds-result-structure"></a>

运行器返回`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`、、`label``explanation``context``tokenUsage`、和等字段`ignoredReferenceInputFields`。有关完整的回复格式，请参阅按[需评估](getting-started-on-demand.md)入门。

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