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

# 批量数据集运行器
<a name="dataset-evaluations-batch"></a>

`BatchEvaluationRunner`委托人通过和 `GetBatchEvaluation` API 将收集和评估完全跨越到服务中`StartBatchEvaluation`。在为每个场景调用代理后，运行器会提交批处理作业并进行轮询直至完成，返回汇总结果。

当您需要汇总多个会话的分数而无需自己管理跨度收集时，可使用批处理运行器；用于基线测量、大型数据集和 pre/post 比较。

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

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

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

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

1.  **提交：运行**者使用 CloudWatch 日志组、调`StartBatchEvaluation`用阶段的会话 ID、评估者 ID 和数据集中的真实情况进行调用。

1.  **轮询：运行**者轮询`GetBatchEvaluation`直到作业达到终端状态并返回汇总结果。

## 代理调用者
<a name="batch-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"
SERVICE_NAME = "<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="batch-ds-example"></a>

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

```
from bedrock_agentcore.evaluation import (
    BatchEvaluationRunner,
    BatchEvaluationRunConfig,
    BatchEvaluatorConfig,
    CloudWatchDataSourceConfig,
    FileDatasetProvider,
)

# 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()

# Configure the batch evaluation
config = BatchEvaluationRunConfig(
    batch_evaluation_name="dataset-batch-eval",
    evaluator_config=BatchEvaluatorConfig(
        evaluator_ids=[
            "Builtin.GoalSuccessRate",
            "Builtin.Correctness",
            "Builtin.TrajectoryExactOrderMatch",
            "Builtin.Helpfulness",
        ],
    ),
    data_source=CloudWatchDataSourceConfig(
        service_names=[SERVICE_NAME],
        log_group_names=[LOG_GROUP],
        ingestion_delay_seconds=180,
    ),
    polling_timeout_seconds=1800,
    polling_interval_seconds=30,
)

# Run
runner = BatchEvaluationRunner(region=REGION)
result = runner.run_dataset_evaluation(
    agent_invoker=agent_invoker,
    dataset=dataset,
    config=config,
)

# Display aggregate results
print(f"Status: {result.status}")
print(f"Batch evaluation ID: {result.batch_evaluation_id}")

if result.evaluation_results:
    er = result.evaluation_results
    print(f"Sessions completed: {er.number_of_sessions_completed}")
    print(f"Sessions failed:    {er.number_of_sessions_failed}")
    print(f"Total sessions:     {er.total_number_of_sessions}")

    for summary in er.evaluator_summaries or []:
        avg = summary.statistics.average_score if summary.statistics else None
        print(f"  {summary.evaluator_id}: avg={avg}")
```

## 获取每个会话的详细信息
<a name="batch-ds-per-session-detail"></a>

汇总结果显示所有会话的平均值。要查看每个会话、每个评估者的分数，请从以下地址获取评估事件： CloudWatch

```
if result.output_data_config:
    events = runner.fetch_evaluation_events(result)
    print(f"\nEvaluation events: {len(events)}")
    for ev in events:
        attrs = ev.get("attributes", {})
        print(f"  session: {attrs.get('session.id', '')[:40]}")
        print(f"  evaluator: {attrs.get('gen_ai.evaluation.name')}")
        print(f"  score: {attrs.get('gen_ai.evaluation.score.value')}")
        print(f"  label: {attrs.get('gen_ai.evaluation.score.label')}")
        print()
```

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

```
BatchEvaluationRunConfig(
    batch_evaluation_name="my-batch-eval",   # Job name
    evaluator_config=BatchEvaluatorConfig(
        evaluator_ids=["Builtin.GoalSuccessRate"],
    ),
    data_source=CloudWatchDataSourceConfig(
        service_names=["MyAgent.DEFAULT"],   # Exactly 1 service name
        log_group_names=[LOG_GROUP],         # 1-5 log group names
        ingestion_delay_seconds=180,         # Wait for CW ingestion (default: 180)
    ),
    polling_timeout_seconds=1800,            # Max wait for job completion (default: 1800)
    polling_interval_seconds=30,             # Poll interval (default: 30)
    simulation_config=None,                  # Set SimulationConfig for simulated scenarios
)
```


| 字段 | 默认值 | 说明 | 
| --- | --- | --- | 
|  `batch_evaluation_name`  | — | 批量评估作业的名称。 | 
|  `evaluator_config.evaluator_ids`  | — | 评估器 ID 列表（内置或自定义）。 | 
|  `data_source.service_names`  | — | 服务名称，用于标识您的代理的踪迹 CloudWatch。 | 
|  `data_source.log_group_names`  | — | CloudWatch 存储代理遥测数据的日志组名称。 | 
|  `data_source.ingestion_delay_seconds`  | 180 | 调用后等待摄取跨度 CloudWatch 所需的秒数。 | 
|  `polling_timeout_seconds`  | 1800 | 等待批处理作业完成的最大秒数。 | 
|  `polling_interval_seconds`  | 30 | 轮询请求之间的秒数。 | 
|  `simulation_config`  | 无 | 模拟场景的配置。`SimulationConfig(model_id="…​")`当数据集包含`SimulatedScenario`实例时设置。参见[用户模拟](user-simulation.md)。 | 

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

跑步者返回`BatchEvaluationResult`：

```
BatchEvaluationResult
  ├── batch_evaluation_id: str
  ├── batch_evaluation_arn: str
  ├── batch_evaluation_name: str
  ├── status: str
  ├── created_at: datetime
  ├── evaluation_results: Optional[BatchEvaluationSummary]
  │     ├── number_of_sessions_completed: int
  │     ├── number_of_sessions_in_progress: int
  │     ├── number_of_sessions_failed: int
  │     ├── number_of_sessions_ignored: int
  │     ├── total_number_of_sessions: int
  │     └── evaluator_summaries: List
  │           ├── evaluator_id: str
  │           ├── statistics.average_score: float
  │           ├── total_evaluated: int
  │           └── total_failed: int
  ├── error_details: Optional[List[str]]
  ├── agent_invocation_failures: List[FailedScenario]
  └── output_data_config: Optional[CloudWatchOutputDataConfig]
        ├── log_group_name: str
        └── log_stream_name: str
```
+  `agent_invocation_failures`列出了在提交批处理作业之前代理调用失败的场景。这些会话不包含在批量评估中。
+  `output_data_config`指向写入每个会话详细信息的 CloudWatch 日志流。`runner.fetch_evaluation_events(result)`用来阅读。

## 错误处理
<a name="batch-ds-error-handling"></a>
+  **场景调用失败**被记录为`FailedScenario`但不会阻塞批处理作业；只会提交成功的会话。
+ 如果**所有场景都失败**，则运行者`ValueError`在调用 API 之前加注。
+  **轮询超时：**`TimeoutError`如果任务超过`polling_timeout_seconds`。
+  **作业失败：**`RuntimeError`如果批量评估状态为`FAILED`或`STOPPED`。