On-demand gestore di set di dati

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

Utilizza il runner on-demand per l'iterazione in fase di sviluppo, le pipeline e i set di dati di piccole dimensioni in cui sono necessari dettagli per scenario e per valutatore immediatamente nella risposta CI/CD .

Come funziona

Il runner elabora gli scenari in tre fasi:

Agente invoker

Il runner richiede un agent invoker, un callable che richiama il tuo agente per un solo turno. L'invoker è indipendente dal framework: puoi chiamare il tuo agente tramite invoke_agent_runtime boto3, una chiamata di funzione diretta, 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.

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 l'invocazione dell'agente. L'CloudWatchAgentSpanCollectorSDK include:

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 sondaggi fino alla visualizzazione degli intervalli e li restituisce come elenco semplice.

Configurazione di 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 runner restituisce un oggetto 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 input results è un dict di risposta non elaborato dell'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 stato FAILED indica che si è verificato un problema strutturale (errore di invocazione dell'agente, errore di raccolta degli span). I singoli errori del valutatore all'interno di uno COMPLETED scenario vengono registrati nell'elenco del valutatore con i campi and. results errorCode errorMessage