On-demand jeu de données Runner

Il OnDemandEvaluationDatasetRunner orchestre l'ensemble du cycle de vie de l'évaluation côté client : invoque l'agent, attend l'ingestion de la télémétrie, collecte des intervalles et appelle l'API CloudWatch Evaluate, le tout en un seul appel. run()

Utilisez le moteur à la demande pour les itérations au moment du développement, les CI/CD pipelines et les petits ensembles de données pour lesquels vous avez besoin de détails par scénario et par évaluateur immédiatement dans la réponse.

Comment ça marche

Le runner traite les scénarios en trois phases :

Invocateur de l'agent

Le coureur a besoin d'un invocateur d'agent, un appelable qui invoque votre agent pour un seul tour. L'invocateur est indépendant du framework : vous pouvez appeler votre agent via boto3invoke_agent_runtime, un appel de fonction direct, une requête HTTP ou toute autre méthode.

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

Exemple

L'exemple suivant charge un ensemble de données à partir d'un fichier JSON et exécute l'évaluation à la demande. Pour le format du jeu de données, voir Schéma du jeu de données.

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

Résultats du processus :

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}")

Pour enregistrer les résultats dans un fichier :

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

Référence de configuration

Collecteur Span

Et AgentSpanCollector qui récupère les intervalles de télémétrie après l'invocation de l'agent. Le SDK est livré CloudWatchAgentSpanCollector avec :

from bedrock_agentcore.evaluation import CloudWatchAgentSpanCollector

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

Le collecteur interroge deux groupes de CloudWatch journaux (aws/spanspour les intervalles structurels et le groupe de journaux de l'agent pour le contenu des conversations), interroge jusqu'à ce que les intervalles apparaissent et les renvoie sous forme de liste plate.

Configuration d'évaluation

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
)

Structure des résultats

Le coureur renvoie un EvaluationResult avec la structure suivante :

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

Chaque entrée results est un dict de réponse brut de l'API Evaluate, contenant des champs tels que value labelexplanation,,context,tokenUsage, etignoredReferenceInputFields. Voir Commencer l'évaluation à la demande pour le format de réponse complet.

Un scénario avec statut FAILED indique qu'un problème structurel s'est produit (erreur d'invocation de l'agent, échec de la collecte des intervalles). Les erreurs individuelles d'un évaluateur dans un COMPLETED scénario sont enregistrées dans la results liste de l'évaluateur avec errorCode et errorMessage champs.