On-demand executor de conjunto de dados

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.

Como funciona

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

Agente invocador

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

Exemplo

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.

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

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

Estrutura de resultados

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 Evaluatevalue, contendo campos como labelexplanation,context,tokenUsage,, ignoredReferenceInputFields e. Consulte Introdução à avaliação sob demanda 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