On-demand Datensatz-Runner

Der OnDemandEvaluationDatasetRunner orchestriert den gesamten Evaluierungslebenszyklus auf der Client-Seite: Er ruft den Agenten auf, wartet auf die Telemetrieaufnahme, sammelt Spans von der Evaluate-API und ruft sie auf — CloudWatch alles in einem einzigen Aufruf. run()

Verwenden Sie den On-Demand-Runner für Iterationen während der Entwicklung, für CI/CD Pipelines und kleine Datensätze, bei denen Sie in der Antwort sofort Details für jedes Szenario und jeden Evaluator benötigen.

Funktionsweise

Der Runner verarbeitet Szenarien in drei Phasen:

Agentenaufrufer

Der Runner benötigt einen Agenten-Invoker, einen Callable, der Ihren Agenten für einen einzigen Zug aufruft. Der Invoker ist Framework-unabhängig: Sie können Ihren Agenten über boto3invoke_agent_runtime, einen direkten Funktionsaufruf, eine HTTP-Anfrage oder eine andere Methode aufrufen.

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

Beispiel

Im folgenden Beispiel wird ein Datensatz aus einer JSON-Datei geladen und die On-Demand-Auswertung ausgeführt. Informationen zum Datensatzformat finden Sie unter Datensatzschema.

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

Prozessergebnisse:

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

Um Ergebnisse in einer Datei zu speichern:

with open("results.json", "w") as f:
    f.write(result.model_dump_json(indent=2))

Konfigurationsreferenz

Span-Kollektor

Und der Telemetriebereiche nach dem Aufruf des Agenten abruft. AgentSpanCollector Das SDK wird ausgeliefert: CloudWatchAgentSpanCollector

from bedrock_agentcore.evaluation import CloudWatchAgentSpanCollector

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

Der Collector fragt zwei CloudWatch Protokollgruppen ab (aws/spansfür Strukturbereiche und die Protokollgruppe des Agenten für Konversationsinhalte), fragt ab, bis Spans angezeigt werden, und gibt sie als unformatierte Liste zurück.

Konfiguration für die Evaluierung

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
)

Struktur der Ergebnisse

Der Runner gibt eine EvaluationResult mit der folgenden Struktur zurück:

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

Jeder Eintrag in results ist ein rohes Antwortdiktat der Evaluate-API, das Felder wievalue,label,, explanation contexttokenUsage, und ignoredReferenceInputFields enthält. Das vollständige Antwortformat finden Sie unter Erste Schritte mit der On-Demand-Evaluierung.

Ein Szenario mit Status FAILED bedeutet, dass ein strukturelles Problem aufgetreten ist (Agentenaufruffehler, Span-Collection-Fehler). Einzelne Evaluatorfehler innerhalb eines COMPLETED Szenarios werden in der results Evaluatorliste mit errorCode den Feldern und aufgezeichnet. errorMessage