Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

On-demand dataset runner

OnDemandEvaluationDatasetRunnerCoordina l'intero ciclo di vita della valutazione lato client: richiama l'agente, attende l'inserimento della telemetria, raccoglie gli intervalli e chiama l'API di valutazione, il tutto in un'unica chiamata. CloudWatch run()

Usa l'on-demand runner per iterazioni in fase di sviluppo, pipeline e piccoli set di dati in cui hai bisogno dei dettagli per scenario e per valutatore immediatamente nella risposta CI/CD .

Come funziona

Il corridore elabora gli scenari in tre fasi:

Agent invoker

Il runner richiede un agent invoker, un callable che invoca il tuo agente per un solo turno. L'invoker è indipendente dal framework: puoi chiamare il tuo agente tramite invoke_agent_runtime boto3, una chiamata diretta di funzione, una richiesta HTTP o qualsiasi altro metodo.

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

Esempio

L'esempio seguente carica un set di dati da un file JSON ed esegue la valutazione su richiesta. Per il formato del set di dati, vedere Schema del set di dati. Schema del set di dati

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

Risultati del processo:

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

Per salvare i risultati in un file:

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

Informazioni di riferimento sulla configurazione

Raccoglitore Span

E AgentSpanCollector che recupera gli intervalli di telemetria dopo la chiamata dell'agente. L'CloudWatchAgentSpanCollectorSDK fornisce:

from bedrock_agentcore.evaluation import CloudWatchAgentSpanCollector

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

Il raccoglitore interroga due gruppi di CloudWatch log (aws/spansper gli intervalli strutturali e il gruppo di log dell'agente per il contenuto delle conversazioni), esegue i sondaggi fino a visualizzare gli intervalli e li restituisce come un elenco semplice.

Configurazione della valutazione

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
)

Struttura dei risultati

Il corridore restituisce un risultato EvaluationResult con la seguente struttura:

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

Ogni voce results è una risposta non elaborata dall'API Evaluate, contenente campi comevalue,label, explanation contexttokenUsage, e. ignoredReferenceInputFields Vedi Guida introduttiva alla valutazione su richiesta per il formato di risposta completo.

Uno scenario con status FAILED indica che si è verificato un problema strutturale (errore di richiamo dell'agente, errore di raccolta dell'intervallo). Gli errori dei singoli valutatori all'interno di uno COMPLETED scenario vengono registrati nell'elenco del valutatore con i campi e. results errorCode errorMessage