オンデマンドデータセットランナー

は評価ライフサイクル全体をクライアント側OnDemandEvaluationDatasetRunnerでオーケストレーションします。エージェントを呼び出し、テレメトリの取り込みを待機し、CloudWatch からスパンを収集し、評価 API を呼び出します。これらはすべて 1 回のrun()呼び出しで行われます。

オンデマンドランナーは、開発時のイテレーション、CI/CD パイプライン、およびシナリオごと、評価者ごとの詳細がレスポンスですぐに必要な小さなデータセットに使用します。

仕組み

ランナーはシナリオを 3 つのフェーズで処理します。

エージェント呼び出し

ランナーには、エージェントを 1 ターン呼び出す呼び出し可能なエージェント呼び出し元が必要です。呼び出し元はフレームワークに依存しません。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))

例

次の例では、JSON ファイルからデータセットをロードし、オンデマンド評価を実行します。データセット形式については、「データセットスキーマ」を参照してください。

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

設定リファレンス

スパンコレクター

エージェントの呼び出し後にテレメトリスパンAgentSpanCollectorを取得する 。SDK には が付属していますCloudWatchAgentSpanCollector。

from bedrock_agentcore.evaluation import CloudWatchAgentSpanCollector

span_collector = CloudWatchAgentSpanCollector(
    log_group_name="/aws/bedrock-agentcore/runtimes/<agent-id>-DEFAULT",
    region=REGION,
)

コレクターは 2 つの 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
)

結果構造

ランナーは、次の構造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 からの raw レスポンスディクトであり、value、、label、explanation、tokenUsage、 contextなどのフィールドが含まれますignoredReferenceInputFields。フルレスポンス形式については、「オンデマンド評価の開始方法」を参照してください。

ステータスのシナリオは、構造上の問題 (エージェントの呼び出しエラー、スパン収集の失敗) が発生したFAILEDことを意味します。COMPLETED シナリオ内の個々の評価者エラーは、 フィールドerrorCodeと errorMessageフィールドを使用して評価者resultsのリストに記録されます。