# オンデマンドデータセットランナー
は評価ライフサイクル全体をクライアント側`OnDemandEvaluationDatasetRunner`でオーケストレーションします。エージェントを呼び出し、テレメトリの取り込みを待機し、CloudWatch からスパンを収集し、評価 API を呼び出します。これらはすべて 1 回の`run()`呼び出しで行われます。
オンデマンドランナーは、開発時のイテレーション、CI/CD パイプライン、およびシナリオごと、評価者ごとの詳細がレスポンスですぐに必要な小さなデータセットに使用します。
**注記**
オンデマンドランナーは、セッションレベル、トレースレベル、ツールコールレベルにわたるすべての組み込みエバリュエーターを含むすべての AgentCore エバリュエーターとカスタムエバリュエーターをサポートします。ランナーは、設定した評価者のレベル対応リクエストの構築、バッチ処理、グラウンドトゥルースマッピングを自動的に処理します。
## 仕組み
ランナーはシナリオを 3 つのフェーズで処理します。
1. **呼び出し:** すべてのシナリオは、スレッドプールを使用して同時に実行されます。各シナリオは一意のセッション ID を取得し、シナリオ内で順番に実行して会話コンテキストを維持します。
1. **待機:** 設定可能な遅延 (デフォルト: 180 秒) により、CloudWatch はテレメトリデータを取り込むことができます。この遅延はシナリオごとではなく 1 回支払われます。
1. **評価:** スパンは CloudWatch から収集され、評価リクエストは評価者ごとに構築されます。データセットのグラウンドトゥルースフィールド (`expected_response`、`assertions`、`expected_trajectory`) は、正しい API リファレンス入力に自動的にマッピングされます。
## エージェント呼び出し
ランナーには、エージェントを 1 ターン呼び出す呼び出し可能なエージェント呼び出し元が必要です。呼び出し元はフレームワークに依存しません。boto3 、直接関数呼び出し`invoke_agent_runtime`、HTTP リクエスト、またはその他の方法でエージェントを呼び出すことができます。
```
import json
import boto3
from bedrock_agentcore.evaluation import AgentInvokerInput, AgentInvokerOutput
REGION = ""
AGENT_ARN = "arn:aws:bedrock-agentcore:::runtime/"
LOG_GROUP = "/aws/bedrock-agentcore/runtimes/-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))
```
| フィールド | タイプ | 説明 |
| --- | --- | --- |
| `AgentInvokerInput.payload` | `str` または `dict` | データセットからのターン入力。 |
| `AgentInvokerInput.session_id` | `str` | シナリオのすべてのターンで安定しています。これをエージェントに渡して、会話コンテキストを維持します。 |
| `AgentInvokerOutput.agent_output` | `Any` | エージェントのレスポンス。 |
## 例
次の例では、JSON ファイルからデータセットをロードし、オンデマンド評価を実行します。データセット形式については、[「データセットスキーマ](dataset-evaluations-schema.md)」を参照してください。
```
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/-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
)
```
| フィールド | デフォルト | 説明 |
| --- | --- | --- |
| `evaluator_config.evaluator_ids` | — | 評価者 IDs (組み込み名またはカスタム評価者 IDs) のリスト。 |
| `evaluation_delay_seconds` | 180 | CloudWatch がスパンを取り込むまで、呼び出し後に待機する秒数。CloudWatch 以外のコレクターを使用する場合は、 を 0 に設定します。 |
| `max_concurrent_scenarios` | 5 | 並列で呼び出して評価するシナリオの最大数。 |
| `simulation_config` | なし | シミュレートされたシナリオの設定。データセットに`SimulatedScenario`インスタンスが含まれている`SimulationConfig(model_id="…")`場合に を設定します。[「ユーザーシミュレーション](user-simulation.md)」を参照してください。 |
## 結果構造
ランナーは、次の構造`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`。フルレスポンス形式については[、「オンデマンド評価の開始方法](getting-started-on-demand.md)」を参照してください。
ステータスのシナリオは、構造上の問題 (エージェントの呼び出しエラー、スパン収集の失敗) が発生した`FAILED`ことを意味します。`COMPLETED` シナリオ内の個々の評価者エラーは、 フィールド`errorCode`と `errorMessage`フィールドを使用して評価者`results`のリストに記録されます。