Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

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 ejecutor bajo demanda para iteraciones en tiempo de desarrollo, CI/CD canalizaciones y pequeños conjuntos de datos para los que necesite información por escenario y evaluador de forma inmediata en la respuesta.

Funcionamiento

El ejecutor procesa los escenarios en tres fases:

Invocador de agentes

El ejecutor necesita un invocador de agente, un invocable que invoca a tu agente durante un solo turno. El invocador no depende 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

En el siguiente ejemplo, se carga un conjunto de datos desde un archivo JSON y se 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

Colector de spam

Un 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 registros del agente para ver el contenido de las conversaciones), 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 ejecutor 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 de evaluación, que contiene campos como value labelexplanation,context,tokenUsage, yignoredReferenceInputFields. Consulta 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 al invocar al agente, error al recopilar el intervalo). Los errores individuales de los evaluadores dentro de un COMPLETED escenario se registran en la results lista del evaluador con errorCode los campos y. errorMessage