배치 데이터 세트 실행기

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

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

작동 방식

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

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

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

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

4. 폴링: 러너는 작업이 터미널 상태에 도달할 GetBatchEvaluation 때까지 폴링하고 집계 결과를 반환합니다.

에이전트 호출자

실행기에는 에이전트를 한 차례 호출하는 호출 가능한 에이전트 호출자가 필요합니다. 호출자는 프레임워크에 구애받지 않습니다. boto3invoke_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	에이전트의 응답입니다.

예제

다음 예시에서는 JSON 파일에서 데이터 세트를 로드하고 배치 평가를 실행합니다. 데이터 세트 형식은 데이터 세트 스키마를 참조하세요.

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}")

세션별 세부 정보 가져오기

집계 결과에는 모든 세션의 평균이 표시됩니다. 세션별, 평가자별 점수를 보려면 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()

구성 참조

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="…​") 경우를 설정합니다. 사용자 시뮬레이션을 참조하세요.

결과 구조

실행기는를 반환합니다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)를 사용하여 읽습니다.

오류 처리

• 시나리오 호출 실패는 로 기록FailedScenario되지만 배치 작업을 차단하지는 않습니다. 성공한 세션만 제출됩니다.

• 모든 시나리오가 실패하면 API를 호출ValueError하기 전에 실행기가 발생합니다.

• 폴링 제한 시간: 작업이를 초과하는 TimeoutError 경우polling_timeout_seconds.

• 작업 실패: 배치 평가 상태가 FAILED 또는 인 RuntimeError 경우STOPPED.