# On-demand corredor de conjuntos de datos
<a name="dataset-evaluations-on-demand"></a>

`OnDemandEvaluationDatasetRunner`Organiza 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.

**nota**  
El procesador bajo demanda es compatible con todos los AgentCore evaluadores, incluidos todos los evaluadores integrados en todos los niveles de sesión, seguimiento y uso de herramientas, así como con los evaluadores personalizados. El ejecutor gestiona automáticamente la creación de solicitudes en función de los niveles, el procesamiento por lotes y el mapeo de datos básicos para los evaluadores que configure.

## Funcionamiento
<a name="ds-how-it-works"></a>

El ejecutor procesa los escenarios en tres fases:

1.  **Invocar:** todos los escenarios se ejecutan simultáneamente mediante un grupo de subprocesos. Cada escenario recibe un identificador de sesión único y, dentro de un escenario, se ejecuta secuencialmente para mantener el contexto de la conversación.

1.  **Espera:** un retraso configurable (predeterminado: 180 segundos) permite CloudWatch ingerir los datos de telemetría. Este retraso se paga una vez, no por escenario.

1.  **Evaluar:** se recopilan los intervalos CloudWatch y se crean las solicitudes de evaluación para cada evaluador. Los campos de información básica del conjunto de datos (`expected_response`,`assertions`,`expected_trajectory`) se asignan automáticamente a las entradas de referencia de la API correctas.

## Agente: invocador
<a name="ds-agent-invoker"></a>

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 boto3`invoke_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))
```


| Campo | Tipo | Description (Descripción) | 
| --- | --- | --- | 
|  `AgentInvokerInput.payload`  |  `str` o `dict`  | El turno es la entrada del conjunto de datos. | 
|  `AgentInvokerInput.session_id`  |  `str`  | Estable en todos los giros de un escenario. Transmita esta información a su agente para mantener el contexto de la conversación. | 
|  `AgentInvokerOutput.agent_output`  |  `Any`  | La respuesta del agente. | 

## Ejemplo
<a name="ds-example"></a>

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](dataset-evaluations-schema.md).

```
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
<a name="ds-components-reference"></a>

 **Recopilador de spam** 

Y `AgentSpanCollector` que recupera los intervalos de telemetría tras la invocación del agente. `CloudWatchAgentSpanCollector`El 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/spans`para 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
)
```


| Campo | Predeterminado | Description (Descripción) | 
| --- | --- | --- | 
|  `evaluator_config.evaluator_ids`  | — | Lista de identificadores de evaluadores (nombres integrados o identificadores de evaluadores personalizados). | 
|  `evaluation_delay_seconds`  | 180 | Hay que esperar segundos después de la invocación para CloudWatch ingerir los intervalos. Configúrelo en 0 si utiliza un recopilador que no sea. CloudWatch  | 
|  `max_concurrent_scenarios`  | 5 | Número máximo de escenarios para invocar y evaluar en paralelo. | 
|  `simulation_config`  | Ninguno | Configuración para escenarios simulados. Se establece `SimulationConfig(model_id="…​")` cuándo el conjunto de datos contiene `SimulatedScenario` instancias. Consulte [Simulación de usuario](user-simulation.md). | 

## Estructura de resultados
<a name="ds-result-structure"></a>

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` `label``explanation`,`context`,`tokenUsage`, y`ignoredReferenceInputFields`. Consulte [Cómo empezar con la evaluación bajo demanda](getting-started-on-demand.md) 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`