Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

On-demand jeu de données runner

Il OnDemandEvaluationDatasetRunner orchestre l'intégralité du cycle de vie d'évaluation côté client : invoquez l'agent, attendez l'ingestion de données télémétriques, collectez les spans et appelez l'API Evaluate CloudWatch, le tout en un seul appel. run()

Utilisez le lanceur à 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 coureur traite les scénarios en trois phases :

Invocateur d'agent

Le coureur a besoin d'un invocateur d'agent, un appelable qui invoque votre agent pendant 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'appel de l'agent. Le SDK fournit : CloudWatchAgentSpanCollector

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 structuraux 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 provenant de l'API Evaluate, contenant des champs tels que value labelexplanation,context,tokenUsage, etignoredReferenceInputFields. Consultez la section Commencer à utiliser l'évaluation à la demande pour connaître le format de réponse complet.

Un scénario avec statut FAILED signifie qu'un problème structurel est survenu (erreur d'appel de l'agent, échec de la collecte du span). Les erreurs de chaque évaluateur dans un COMPLETED scénario sont enregistrées dans la results liste de l'évaluateur avec les champs errorCode eterrorMessage.