# 온디맨드 데이터 세트 실행기 는 전체 평가 수명 주기 클라이언트 측을 `OnDemandEvaluationDatasetRunner` 오케스트레이션합니다. 즉, 에이전트 호출, 원격 측정 수집 대기, CloudWatch에서 범위 수집, 평가 API 호출이 모두 한 번의 `run()` 호출로 이루어집니다. 응답에서 즉시 시나리오별, 평가자별 세부 정보가 필요한 개발 시간 반복, CI/CD 파이프라인 및 소규모 데이터 세트에 온디맨드 실행기를 사용합니다. **참고** 온디맨드 실행기는 세션, 추적 및 도구 호출 수준 전반의 모든 기본 제공 평가자와 사용자 지정 평가자를 포함하여 모든 AgentCore 평가자를 지원합니다. 실행기는 구성한 평가자에 대한 레벨 인식 요청 구성, 일괄 처리 및 실측 정보 매핑을 자동으로 처리합니다. ## 작동 방식 실행기는 세 단계로 시나리오를 처리합니다. 1. **간접 호출:** 모든 시나리오는 스레드 풀을 사용하여 동시에 실행됩니다. 각 시나리오는 고유한 세션 ID를 가져오고 시나리오 내에서 순차적으로 실행되어 대화 컨텍스트를 유지합니다. 1. **대기:** 구성 가능한 지연(기본값: 180초)을 통해 CloudWatch는 원격 측정 데이터를 수집할 수 있습니다. 이 지연은 시나리오당이 아니라 한 번 지불됩니다. 1. **평가:** CloudWatch에서 스팬을 수집하고 각 평가자에 대해 평가 요청을 작성합니다. 데이터 세트(`expected_response`, `assertions`, `expected_trajectory`)의 실측 정보 필드는 올바른 API 참조 입력에 자동으로 매핑됩니다. ## 에이전트 호출자 실행기에는 에이전트를 한 차례 호출하는 호출 가능한 에이전트 호출자가 필요합니다. 호출자는 프레임워크에 구애받지 않습니다. boto3`invoke_agent_runtime`, 직접 함수 호출, HTTP 요청 또는 기타 방법을 통해 에이전트를 호출할 수 있습니다. ``` 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)) ``` | Field | 유형 | 설명 | | --- | --- | --- | | `AgentInvokerInput.payload` | `str` 또는 `dict` | 데이터 세트의 회전 입력입니다. | | `AgentInvokerInput.session_id` | `str` | 시나리오의 모든 턴에서 안정적입니다. 대화 컨텍스트를 유지하려면 이를 에이전트에게 전달합니다. | | `AgentInvokerOutput.agent_output` | `Any` | 에이전트의 응답입니다. | ## 예제 다음 예시에서는 JSON 파일에서 데이터 세트를 로드하고 온디맨드 평가를 실행합니다. 데이터 세트 형식은 [데이터 세트 스키마](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)") ``` 프로세스 결과: ``` 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}") ``` 결과를 파일에 저장하려면: ``` with open("results.json", "w") as f: f.write(result.model_dump_json(indent=2)) ``` ## 구성 참조 **스팬 수집기** 에이전트 호출 후 원격 측정을 검색`AgentSpanCollector`하는 입니다. SDK는를 제공합니다`CloudWatchAgentSpanCollector`. ``` from bedrock_agentcore.evaluation import CloudWatchAgentSpanCollector span_collector = CloudWatchAgentSpanCollector( log_group_name="/aws/bedrock-agentcore/runtimes/-DEFAULT", region=REGION, ) ``` 수집기는 두 개의 CloudWatch 로그 그룹(`aws/spans`구조 스팬의 경우 , 대화 콘텐츠의 경우 에이전트의 로그 그룹)을 쿼리하고 스팬이 나타날 때까지 폴링한 다음 플랫 목록으로 반환합니다. **평가 구성** ``` 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 ) ``` | Field | 기본값 | 설명 | | --- | --- | --- | | `evaluator_config.evaluator_ids` | — | 평가자 IDs 목록(기본 제공 이름 또는 사용자 지정 평가자 IDs). | | `evaluation_delay_seconds` | 180 | CloudWatch가 스팬을 수집하기 위해 호출 후 대기하는 초 단위입니다. CloudWatch가 아닌 수집기를 사용하는 경우 0으로 설정합니다. | | `max_concurrent_scenarios` | 5 | 병렬로 호출하고 평가할 최대 시나리오 수입니다. | | `simulation_config` | 없음 | 시뮬레이션된 시나리오에 대한 구성입니다. 데이터 세트에 `SimulatedScenario` 인스턴스가 포함된 `SimulationConfig(model_id="…​")` 경우를 설정합니다. [사용자 시뮬레이션](user-simulation.md)을 참조하세요. | ## 결과 구조 실행기는 다음 구조의 `EvaluationResult`를 반환합니다. ``` 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 ``` 의 각 항목은 평가 API의 원시 응답 명령`results`으로, , `value`, `label``explanation`, `context`, `tokenUsage`등의 필드를 포함합니다`ignoredReferenceInputFields`. 전체 응답 형식[은 온디맨드 평가 시작하기](getting-started-on-demand.md)를 참조하세요. 상태가 인 시나리오는 구조적 문제(에이전트 호출 오류, 스팬 수집 실패)가 발생했음을 `FAILED` 의미합니다. `COMPLETED` 시나리오 내의 개별 평가자 오류는 `errorCode` 및 `errorMessage` 필드와 `results` 함께 평가자 목록에 기록됩니다.