# On-demand executor de conjunto de dados
<a name="dataset-evaluations-on-demand"></a>

Ele `OnDemandEvaluationDatasetRunner` orquestra todo o ciclo de vida da avaliação no lado do cliente: invoque o agente, aguarde a ingestão da telemetria, colete períodos e chame a API Evaluate, tudo em uma única chamada. CloudWatch `run()`

Use o executor sob demanda para iteração em tempo de desenvolvimento, CI/CD pipelines e pequenos conjuntos de dados em que você precisa de detalhes por cenário e por avaliador imediatamente na resposta.

**nota**  
O executor sob demanda oferece suporte a todos os AgentCore avaliadores, incluindo todos os avaliadores integrados nos níveis de sessão, rastreamento e chamada de ferramentas, bem como avaliadores personalizados. O executor gerencia automaticamente a construção de solicitações com reconhecimento de nível, o agrupamento em lotes e o mapeamento da verdade básica para qualquer avaliador que você configurar.

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

O executor processa cenários em três fases:

1.  **Invocar:** todos os cenários são executados simultaneamente usando um pool de threads. Cada cenário recebe um ID de sessão exclusivo e, dentro de um cenário, é executado sequencialmente para manter o contexto da conversa.

1.  **Espera:** um atraso configurável (padrão: 180 segundos) permite CloudWatch a ingestão dos dados de telemetria. Esse atraso é pago uma vez, não por cenário.

1.  **Avalie:** os intervalos são coletados CloudWatch e as solicitações de avaliação são criadas para cada avaliador. Os campos de verdade básica do conjunto de dados (`expected_response`,`assertions`,`expected_trajectory`) são mapeados automaticamente para as entradas de referência corretas da API.

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

O corredor precisa de um agente invocador, um chamável que invoca seu agente por um único turno. O invocador é independente da estrutura: você pode chamar seu agente via boto3`invoke_agent_runtime`, uma chamada direta de função, uma solicitação HTTP ou qualquer outro 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 | 
| --- | --- | --- | 
|  `AgentInvokerInput.payload`  |  `str` ou `dict`  | A entrada de turno do conjunto de dados. | 
|  `AgentInvokerInput.session_id`  |  `str`  | Estável em todas as curvas em um cenário. Transmita isso ao seu agente para manter o contexto da conversa. | 
|  `AgentInvokerOutput.agent_output`  |  `Any`  | A resposta do agente. | 

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

O exemplo a seguir carrega um conjunto de dados de um arquivo JSON e executa a avaliação sob demanda. Para o formato do conjunto de dados, consulte Esquema do [conjunto de dados.](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 do 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}")
```

Para salvar os resultados em um arquivo:

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

## Referência da configuração
<a name="ds-components-reference"></a>

 **Coletor Span** 

E `AgentSpanCollector` que recupera os intervalos de telemetria após a invocação do agente. O SDK fornece: `CloudWatchAgentSpanCollector`

```
from bedrock_agentcore.evaluation import CloudWatchAgentSpanCollector

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

O coletor consulta dois grupos de CloudWatch registros (`aws/spans`para extensões estruturais e o grupo de registros do agente para conteúdo de conversas), pesquisa até que os intervalos apareçam e os retorna como uma lista simples.

 **Configuração de avaliação** 

```
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 | Padrão | Description | 
| --- | --- | --- | 
|  `evaluator_config.evaluator_ids`  | — | Lista de IDs de avaliador (nomes integrados ou IDs de avaliador personalizados). | 
|  `evaluation_delay_seconds`  | 180 | Segundos de espera após a invocação para CloudWatch ingerir os intervalos. Defina como 0 se estiver usando um não CloudWatch coletor. | 
|  `max_concurrent_scenarios`  | 5 | Número máximo de cenários a serem invocados e avaliados paralelamente. | 
|  `simulation_config`  | Nenhum | Configuração para cenários simulados. Defina `SimulationConfig(model_id="…​")` quando o conjunto de dados contém `SimulatedScenario` instâncias. Consulte [Simulação do usuário](user-simulation.md). | 

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

O corredor retorna um `EvaluationResult` com a seguinte estrutura:

```
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` é um ditado de resposta bruto da API Evaluate`value`, contendo campos como `label``explanation`,`context`,`tokenUsage`,, `ignoredReferenceInputFields` e. Consulte [Introdução à avaliação sob demanda](getting-started-on-demand.md) para ver o formato completo da resposta.

Um cenário com status `FAILED` significa que ocorreu um problema estrutural (erro de invocação do agente, falha na coleta de intervalos). Os erros individuais do avaliador em um `COMPLETED` cenário são registrados na `results` lista do avaliador com os campos `errorCode` e. `errorMessage`