批次資料集執行器

BatchEvaluationRunner 委派代表會透過 和 GetBatchEvaluation APIs，將收集StartBatchEvaluation和評估完全跨到服務。為每個案例叫用您的代理程式後，執行器會提交批次任務並輪詢，直到完成為止，並傳回彙總結果。

當您需要跨多個工作階段彙總分數而不自行管理跨度收集時，請使用批次執行器；用於基準測量、大型資料集和前/後比較。

運作方式

執行器會以四個階段處理案例：

客服人員叫用程式

執行器需要代理程式叫用程式，這是可呼叫的，可叫用您的代理程式一次。叫用者與架構無關：您可以透過 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))

範例

下列範例會從 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
)

結果結構

執行器會傳回 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

錯誤處理