As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

On-demand executor de conjuntos de dados

Ele OnDemandEvaluationDatasetRunner orquestra todo o ciclo de vida da avaliação do lado do cliente: invoque o agente, aguarde a ingestão de 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 exige um agente invocador, um chamável que invoque 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, 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

Um AgentSpanCollector que recupera extensões 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 o conteúdo da conversa), 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 executor 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 bruta da API Evaluatevalue, contendo campos como label explanationcontext,tokenUsage,, ignoredReferenceInputFields e. Consulte Introdução à avaliação sob demanda para ver o formato de resposta completo.

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