On-demand corredor de conjuntos de datos

OnDemandEvaluationDatasetRunnerOrganiza todo el ciclo de vida de la evaluación desde el lado del cliente: invoca al agente, espera a que se ingiera la telemetría, recopila los intervalos y llama a la API de evaluación CloudWatch, todo en una sola llamada. run()

Utilice el programa bajo demanda para realizar iteraciones en tiempo de desarrollo, CI/CD canalizaciones y conjuntos de datos pequeños, en los que necesite información detallada por escenario y por evaluador de forma inmediata en la respuesta.

Funcionamiento

El ejecutor procesa los escenarios en tres fases:

Agente: invocador

El corredor requiere un invocador de agentes, un invocador que invoca a tu agente durante un solo turno. El invocador es independiente del marco: puedes llamar a tu agente mediante boto3invoke_agent_runtime, una llamada directa a una función, una solicitud HTTP o cualquier otro método.

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

Ejemplo

El siguiente ejemplo carga un conjunto de datos desde un archivo JSON y ejecuta la evaluación bajo demanda. Para conocer el formato del conjunto de datos, consulte Esquema del conjunto de datos.

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

Resultados del proceso:

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

Para guardar los resultados en un archivo:

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

Referencia de la configuración

Recopilador de spam

Y AgentSpanCollector que recupera los intervalos de telemetría tras la invocación del agente. CloudWatchAgentSpanCollectorEl SDK incluye:

from bedrock_agentcore.evaluation import CloudWatchAgentSpanCollector

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

El recopilador consulta dos grupos de CloudWatch registros (aws/spanspara los intervalos estructurales y el grupo de registro del agente para el contenido de las conversaciones), los sondea hasta que aparecen los intervalos y los devuelve como una lista plana.

Configuración de evaluación

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
)

Estructura de resultados

El corredor devuelve un EvaluationResult con la siguiente estructura:

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

Cada entrada results es un dictado de respuesta sin procesar de la API Evaluate, que contiene campos como value labelexplanation,context,tokenUsage, yignoredReferenceInputFields. Consulte Cómo empezar con la evaluación bajo demanda para ver el formato de respuesta completo.

Un escenario con estado FAILED significa que se ha producido un problema estructural (error de invocación del agente, error de recopilación de intervalos). Los errores de los evaluadores individuales dentro de un COMPLETED escenario se registran en la results lista del evaluador con errorCode los campos y. errorMessage