View a markdown version of this page

隨需資料集執行器 - Amazon Bedrock AgentCore
運作方式客服人員叫用程式範例組態參考結果結構

隨需資料集執行器

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

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

注意

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

運作方式

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

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

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

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

客服人員叫用程式

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

代理程式的回應。

範例

下列範例會從 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 擷取代理程式調用後遙測範圍的 。開發套件隨附 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="…​")時設定 。請參閱使用者模擬。

結果結構

執行器會傳回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。如需完整回應格式，請參閱隨需評估入門。

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