# 隨需資料集執行器
<a name="dataset-evaluations-on-demand"></a>

會`OnDemandEvaluationDatasetRunner`協調整個評估生命週期用戶端：叫用代理程式、等待遙測擷取、從 CloudWatch 收集範圍，以及呼叫評估 API，全都在單一`run()`呼叫中執行。

使用隨需執行器進行開發時間反覆運算、CI/CD 管道，以及小型資料集，其中您需要每個案例、每個評估器的回應詳細資訊。

**注意**  
隨需執行器支援所有 AgentCore 評估器，包括跨工作階段、追蹤和工具呼叫層級的所有內建評估器，以及自訂評估器。執行器會自動為您設定的任何評估器處理關卡感知請求建構、批次處理和 Ground Truth 映射。

## 運作方式
<a name="ds-how-it-works"></a>

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

1.  **調用：**所有案例都會使用執行緒集區同時執行。每個案例都會取得唯一的工作階段 ID，並在案例內依序執行，以維持對話內容。

1.  **等待：**可設定的延遲 （預設值：180 秒） 可讓 CloudWatch 擷取遙測資料。此延遲會支付一次，而非每個案例。

1.  **評估：**從 CloudWatch 收集跨度，並為每個評估器建置評估請求。來自資料集的 Ground Truth 欄位 (`expected_response`、`assertions`、`expected_trajectory`) 會自動映射至正確的 API 參考輸入。

## 客服人員叫用程式
<a name="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"

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


| 欄位 | Type | 說明 | 
| --- | --- | --- | 
|  `AgentInvokerInput.payload`  |  `str` 或 `dict`  | 來自資料集的轉彎輸入。 | 
|  `AgentInvokerInput.session_id`  |  `str`  | 在案例的所有轉彎中保持穩定。將此傳遞給您的客服人員，以維持對話內容。 | 
|  `AgentInvokerOutput.agent_output`  |  `Any`  | 代理程式的回應。 | 

## 範例
<a name="ds-example"></a>

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

## 組態參考
<a name="ds-components-reference"></a>

 **跨度收集器** 

`AgentSpanCollector` 擷取代理程式調用後遙測範圍的 。開發套件隨附 `CloudWatchAgentSpanCollector`：

```
from bedrock_agentcore.evaluation import CloudWatchAgentSpanCollector

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

收集器會查詢兩個 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)。 | 

## 結果結構
<a name="ds-result-structure"></a>

執行器會傳回`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`都是評估 API 的原始回應索引，其中包含 `value`、`label`、、`tokenUsage`、 `explanation` `context`和 等欄位`ignoredReferenceInputFields`。如需完整回應格式，請參閱[隨需評估入門](getting-started-on-demand.md)。

狀態為 的案例`FAILED`表示發生結構問題 （客服人員叫用錯誤、跨收集失敗）。`COMPLETED` 案例內的個別評估器錯誤會以 `errorCode`和 `errorMessage` 欄位記錄在評估器的`results`清單中。