给 Agent 建一套能跟着长大的测试基准——Bedrock AgentCore 数据集管理实践

import boto3
import json
from datetime import datetime

client = boto3.client("bedrock-agentcore", region_name="us-east-1")

# 创建评估数据集
dataset_name = "customer-service-agent-baseline"
create_resp = client.create_evaluation_dataset(
    name=dataset_name,
    description="客服 Agent 离线回归基准——覆盖退款、改地址、查物流等核心场景",
)

dataset_id = create_resp["datasetId"]
print(f"数据集已创建: {dataset_id}")

# 写入第一批测试用例
test_cases = [
    {
        "testCaseId": "refund-standard",
        "input": {
            "userMessage": "我昨天买的鞋子想退款，订单号 CS-20240512-007",
        },
        "expectedBehavior": {
            "intent": "refund_request",
            "mustMention": ["退款流程", "预计到账时间"],
            "mustNotMention": ["换货"],
            "finalState": "refund_initiated",
        },
    },
    {
        "testCaseId": "address-change-with-refund",
        "input": {
            "userMessage": "我要改收货地址，但这个订单我已经申请退款了",
        },
        "expectedBehavior": {
            "intent": "address_change_with_conflict",
            "mustMention": ["退款进行中无法改地址", "建议取消退款后再改"],
            "mustNotMention": ["直接改地址成功"],
            "finalState": "conflict_explained",
        },
    },
    {
        "testCaseId": "shipping-status",
        "input": {
            "userMessage": "订单 CS-20240510-003 到哪了",
        },
        "expectedBehavior": {
            "intent": "shipping_query",
            "mustMention": ["当前物流状态", "预计送达日期"],
            "mustNotMention": [],
            "finalState": "status_provided",
        },
    },
]

for tc in test_cases:
    client.put_test_case(
        datasetId=dataset_id,
        testCase=tc,
    )

print(f"已写入 {len(test_cases)} 条测试用例")

# 锁定当前用例集合为 v1 版本——之后不可修改
version_resp = client.create_dataset_version(
    datasetId=dataset_id,
    versionLabel="v1",
    description="初始基准：3 条核心场景，2024-05-15 发布",
)

version_id = version_resp["versionId"]
print(f"基准版本已锁定: v1 → {version_id}")

# 对指定 Agent 跑 v1 基准的回归测试
eval_resp = client.start_evaluation_run(
    evaluationDatasetId=dataset_id,
    datasetVersionId=version_id,
    agentId="my-cs-agent",
    agentAliasId="prod-v2",          # 当前生产别名
    evaluationConfig={
        "outputFormat": "detailed",   # 每条用例的逐项判定
    },
)

run_id = eval_resp["runId"]
print(f"评估已启动: run={run_id}")

# 等待完成并拉取结果（简化示意，实际应轮询或用 EventBridge）
import time
time.sleep(30)  # 等待评估完成——生产代码请用 waiter

result = client.get_evaluation_run_results(runId=run_id)

# 打印每条用例的通过情况
for case_result in result["testCaseResults"]:
    tc_id = case_result["testCaseId"]
    passed = case_result["overallPassed"]
    details = case_result["criteriaResults"]
    print(f"  {tc_id}: {'✅ PASS' if passed else '❌ FAIL'}")
    for d in details:
        print(f"    - {d['criterion']}: {d['passed']} | {d['reason']}")

# 业务新增了"会员积分查询"场景，追加用例
new_case = {
    "testCaseId": "points-query",
    "input": {
        "userMessage": "我账户里还有多少积分可以用",
    },
    "expectedBehavior": {
        "intent": "points_balance_query",
        "mustMention": ["当前积分余额", "积分有效期"],
        "mustNotMention": [],
        "finalState": "points_displayed",
    },
}

client.put_test_case(datasetId=dataset_id, testCase=new_case)

# 发布 v2 基准
v2_resp = client.create_dataset_version(
    datasetId=dataset_id,
    versionLabel="v2",
    description="新增积分查询场景，共 4 条用例",
)

v2_version_id = v2_resp["versionId"]

# 用同一个 Agent 分别跑 v1 和 v2，对比分数差异
v1_run = client.start_evaluation_run(
    evaluationDatasetId=dataset_id,
    datasetVersionId=version_id,       # v1
    agentId="my-cs-agent",
    agentAliasId="prod-v3",            # 新版本 Agent
)

v2_run = client.start_evaluation_run(
    evaluationDatasetId=dataset_id,
    datasetVersionId=v2_version_id,    # v2
    agentId="my-cs-agent",
    agentAliasId="prod-v3",
)

# 两个 run 完成后对比：
# v1 分数 vs v2 分数 → 基准扩大的影响
# v1 分数(prod-v2) vs v1 分数(prod-v3) → Agent 改动的真实效果