On-demand pelari kumpulan data

OnDemandEvaluationDatasetRunnerIni mengatur seluruh sisi klien siklus hidup evaluasi: memanggil agen, menunggu konsumsi telemetri, mengumpulkan rentang dari CloudWatch, dan memanggil Evaluate API, semuanya dalam satu panggilan. run()

Gunakan runner sesuai permintaan untuk iterasi waktu pengembang, CI/CD saluran pipa, dan kumpulan data kecil di mana Anda memerlukan detail per skenario, per-evaluator segera dalam respons.

Cara kerjanya

Pelari memproses skenario dalam tiga fase:

Agen invoker

Pelari membutuhkan agen pemanggil, panggilan yang memanggil agen Anda untuk satu giliran. Invoker adalah framework-agnostic: Anda dapat memanggil agen Anda melalui boto3invoke_agent_runtime, panggilan fungsi langsung, permintaan HTTP, atau metode lainnya.

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

Contoh

Contoh berikut memuat dataset dari file JSON dan menjalankan evaluasi sesuai permintaan. Untuk format kumpulan data, lihat Skema kumpulan data.

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

Hasil proses:

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

Untuk menyimpan hasil ke file:

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

Referensi konfigurasi

Kolektor rentang

An AgentSpanCollector yang mengambil rentang telemetri setelah pemanggilan agen. SDK mengirimkanCloudWatchAgentSpanCollector:

from bedrock_agentcore.evaluation import CloudWatchAgentSpanCollector

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

Kolektor menanyakan dua grup CloudWatch log (aws/spansuntuk rentang struktural dan grup log agen untuk konten percakapan), polling hingga bentang muncul, dan mengembalikannya sebagai daftar datar.

Konfigurasi evaluasi

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
)

Struktur hasil

Pelari mengembalikan a EvaluationResult dengan struktur berikut:

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

Setiap entri di dalamnya results adalah dict respons mentah dari Evaluate API, yang berisi bidang sepertivalue,label,explanation,, contexttokenUsage, danignoredReferenceInputFields. Lihat Memulai evaluasi sesuai permintaan untuk format respons lengkap.

Skenario dengan status FAILED berarti terjadi masalah struktural (kesalahan pemanggilan agen, kegagalan pengumpulan rentang). Kesalahan evaluator individu dalam COMPLETED skenario dicatat dalam results daftar evaluator dengan errorCode dan errorMessage bidang.