# 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. **Note** Le logiciel à la demande prend en charge tous les AgentCore évaluateurs, y compris tous les évaluateurs intégrés aux niveaux de session, de trace et d'appel d'outils, ainsi que les évaluateurs personnalisés. Le runner gère automatiquement la construction des demandes en fonction du niveau, le traitement par lots et le mappage de la vérité sur le terrain pour les évaluateurs que vous configurez. ## Comment ça marche Le runner traite les scénarios en trois phases : 1. **Invoke :** tous les scénarios s'exécutent simultanément à l'aide d'un pool de threads. Chaque scénario reçoit un identifiant de session unique et transforme un scénario exécuté de manière séquentielle pour maintenir le contexte de la conversation. 1. **Attendre :** un délai configurable (par défaut : 180 secondes) permet d' CloudWatch ingérer les données de télémétrie. Ce délai est payé une seule fois, et non par scénario. 1. **Évaluer :** les intervalles sont collectés auprès de chaque évaluateur CloudWatch et des demandes d'évaluation sont créées pour chaque évaluateur. Les champs Ground Truth de l'ensemble de données (`expected_response``assertions`,,`expected_trajectory`) sont automatiquement mappés aux entrées de référence d'API correctes. ## 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 boto3`invoke_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 = "" AGENT_ARN = "arn:aws:bedrock-agentcore:::runtime/" LOG_GROUP = "/aws/bedrock-agentcore/runtimes/-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)) ``` | Champ | Type | Description | | --- | --- | --- | | `AgentInvokerInput.payload` | `str` ou `dict` | Entrée du tour de l'ensemble de données. | | `AgentInvokerInput.session_id` | `str` | Stable à tous les virages d'un scénario. Transmettez-le à votre agent pour maintenir le contexte de la conversation. | | `AgentInvokerOutput.agent_output` | `Any` | La réponse de l'agent. | ## 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](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)") ``` 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/-DEFAULT", region=REGION, ) ``` Le collecteur interroge deux groupes de CloudWatch journaux (`aws/spans`pour 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 ) ``` | Champ | Par défaut | Description | | --- | --- | --- | | `evaluator_config.evaluator_ids` | — | Liste des identifiants d'évaluateurs (noms intégrés ou identifiants d'évaluateurs personnalisés). | | `evaluation_delay_seconds` | 180 | Quelques secondes à attendre après l'invocation pour CloudWatch ingérer des spans. Défini sur 0 si vous utilisez un CloudWatch non-collecteur. | | `max_concurrent_scenarios` | 5 | Nombre maximum de scénarios à invoquer et à évaluer en parallèle. | | `simulation_config` | Aucune | Configuration pour les scénarios simulés. Définissez le `SimulationConfig(model_id="…​")` moment où l'ensemble de données contient `SimulatedScenario` des instances. Voir [Simulation utilisateur](user-simulation.md). | ## 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` `label``explanation`,,`context`,`tokenUsage`, et`ignoredReferenceInputFields`. Voir [Commencer l'évaluation à la demande](getting-started-on-demand.md) 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.