온디맨드 데이터 세트 실행기

는 전체 평가 수명 주기 클라이언트 측을 OnDemandEvaluationDatasetRunner 오케스트레이션합니다. 즉, 에이전트 호출, 원격 측정 수집 대기, CloudWatch에서 범위 수집, 평가 API 호출이 모두 한 번의 run() 호출로 이루어집니다.

응답에서 즉시 시나리오별, 평가자별 세부 정보가 필요한 개발 시간 반복, CI/CD 파이프라인 및 소규모 데이터 세트에 온디맨드 실행기를 사용합니다.

작동 방식

실행기는 세 단계로 시나리오를 처리합니다.

에이전트 호출자

실행기에는 에이전트를 한 차례 호출하는 호출 가능한 에이전트 호출자가 필요합니다. 호출자는 프레임워크에 구애받지 않습니다. boto3invoke_agent_runtime, 직접 함수 호출, HTTP 요청 또는 기타 방법을 통해 에이전트를 호출할 수 있습니다.

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

예제

다음 예시에서는 JSON 파일에서 데이터 세트를 로드하고 온디맨드 평가를 실행합니다. 데이터 세트 형식은 데이터 세트 스키마를 참조하세요.

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/<agent-id>-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
)

결과 구조

실행기는 다음 구조의 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, labelexplanation, context, tokenUsage등의 필드를 포함합니다ignoredReferenceInputFields. 전체 응답 형식은 온디맨드 평가 시작하기를 참조하세요.

상태가 인 시나리오는 구조적 문제(에이전트 호출 오류, 스팬 수집 실패)가 발생했음을 FAILED 의미합니다. COMPLETED 시나리오 내의 개별 평가자 오류는 errorCode 및 errorMessage 필드와 results 함께 평가자 목록에 기록됩니다.