# 배치 데이터 세트 실행기
<a name="dataset-evaluations-batch"></a>

`BatchEvaluationRunner` 대리인은 `StartBatchEvaluation` 및 `GetBatchEvaluation` APIs를 통해 수집 및 평가를 서비스에 전적으로 적용합니다. 각 시나리오에 대해 에이전트를 호출한 후 실행기는 배치 작업을 제출하고 완료될 때까지 폴링하여 집계 결과를 반환합니다.

스팬 수집을 직접 관리하지 않고 여러 세션에서 집계 점수가 필요한 경우 기준 측정, 대규모 데이터 세트 및 사전/사후 비교를 위해 배치 실행기를 사용합니다.

## 작동 방식
<a name="batch-ds-how-it-works"></a>

실행기는 4단계로 시나리오를 처리합니다.

1.  **간접 호출:** 모든 시나리오는 스레드 풀을 사용하여 동시에 실행됩니다. 각 시나리오는 고유한 세션 ID를 가져오고 시나리오 내에서 순차적으로 실행되어 대화 컨텍스트를 유지합니다.

1.  **대기:** 구성 가능한 수집 지연(기본값: 180초)을 통해 CloudWatch는 원격 측정 데이터를 수집할 수 있습니다. 이 지연은 시나리오당이 아니라 한 번 지불됩니다.

1.  **제출:** 실행기가 CloudWatch 로그 그룹, 호출 단계의 세션 IDs, 평가자 IDs 및 데이터 세트의 실측 정보를 `StartBatchEvaluation` 사용하여를 호출합니다.

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


| Field | 유형 | 설명 | 
| --- | --- | --- | 
|  `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
)
```


| Field | 기본값 | 설명 | 
| --- | --- | --- | 
|  `batch_evaluation_name`  | — | 배치 평가 작업의 이름입니다. | 
|  `evaluator_config.evaluator_ids`  | — | 평가자 IDs 목록(기본 제공 또는 사용자 지정). | 
|  `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`  | 없음 | 시뮬레이션된 시나리오에 대한 구성입니다. 데이터 세트에 `SimulatedScenario` 인스턴스가 포함된 `SimulationConfig(model_id="…​")` 경우를 설정합니다. [사용자 시뮬레이션](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`되지만 배치 작업을 차단하지는 않습니다. 성공한 세션만 제출됩니다.
+ **모든 시나리오가 실패**하면 API를 호출`ValueError`하기 전에 실행기가 발생합니다.
+  **폴링 제한 시간:** 작업이를 초과하는 `TimeoutError` 경우`polling_timeout_seconds`.
+  **작업 실패:** 배치 평가 상태가 `FAILED` 또는 인 `RuntimeError` 경우`STOPPED`.