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

它在`OnDemandEvaluationDatasetRunner`客户端协调整个评估生命周期：调用代理、等待遥测摄取、从中收集跨度以及调用 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`都是来自 Evaluate API 的原始响应字典，其中包含诸如`value`、、`label`、`explanation``context``tokenUsage`、和之类的字段`ignoredReferenceInputFields`。有关完整的回复格式[，请参阅按需评估入门](getting-started-on-demand.md)。

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