Agent 评估与测试体系：从单轮评分到端到端流水线

为什么"看起来对了"不算评估

Agent 系统有一个让所有团队头疼的特性：非确定性。同一个 prompt 运行两次，可能得到完全不同的执行路径。一个 coding agent 这次成功修复了 bug，下次就可能把测试文件删了。

大多数团队在早期都用同一种方式"评估"：手动跑几个 case，看看输出"感觉对不对"。这在 demo 阶段够用，但一旦进入迭代周期就会暴露三个问题：

隐性回归无法发现。 你优化了 planning prompt，tool selection 准确率从 92% 降到 78%，但因为两个 case 恰好还跑通了，你完全没有感知。等到用户报告问题时，你已经推送了三个版本。

改进没有方向。 "Agent 变差了"是一句无用的话。差在哪里？是工具选错了？计划步骤多了？还是最终输出格式不对？没有分层指标，改进就是盲人摸象。

无法做权衡决策。 换一个更便宜但更弱的模型，成功率会掉多少？加一个工具选择验证步骤，延迟会增加多少？没有量化数据，这类决策只能靠猜。

三层评估体系

一个完整的 Agent 评估体系应该覆盖三个层级，每一层回答不同的问题：

第一层：组件级评估（Component Evaluation）

回答：每个子模块是否在独立工作？

需要测试的组件和对应指标：

组件	关键指标	测试方法
工具选择（Tool Selection）	准确率、F1	给定任务描述，验证是否选择了正确的工具和参数
规划（Planning）	步骤合理性、冗余度	给定目标，评估计划是否包含必要步骤且无多余步骤
输出格式（Output Formatting）	格式合规率	验证输出是否严格符合 JSON Schema / 函数签名
检索（Retrieval）	召回率、精确率	标准信息检索评估，适用于 RAG 类 Agent

组件级评估的核心价值是定位问题。当端到端测试失败时，组件级指标能告诉你是哪一层出了问题。

第二层：交互级评估（Interaction / Trajectory Evaluation）

回答：Agent 的执行路径是否合理？

交互级评估关注的是 trajectory —— Agent 从开始到结束走过的完整路径。关键指标：

• 轨迹正确率：Agent 是否走了最优路径？是否走了不必要的弯路？
• 步骤效率：完成任务实际用了多少步 vs 最少需要多少步
• 错误恢复率：当 Agent 犯错后，是否能自我纠正并回到正确路径
• 工具调用效率：是否调用了不必要的工具？参数是否准确？

这一层评估最难自动化，因为它需要定义"什么是正确的轨迹"。在实践中，通常用 LLM-as-judge 对比实际轨迹和参考轨迹。

第三层：结果级评估（Outcome Evaluation）

回答：任务最终是否完成了？

这是最重要的也是最终需要关注的层级：

• 任务完成率：任务是否真正完成（不是"输出了一段看起来像答案的文本"）
• 用户满意度：人工评估或隐式反馈（用户是否追问、是否采纳结果）
• 单任务成本：完成一个任务消耗的 token 数 / API 调用次数 / 延迟
• 成本-质量 Pareto：在给定预算下，哪种配置能达到最优质量

AWS 的 Agent Evaluation 框架就是从结果级入手，通过预定义的任务集和评判标准来衡量 Agent 在不同场景下的表现。

生产实践模式一：构建金标准评估数据集

评估的第一步不是选指标，而是有一份可靠的数据集。以下是构建数据集的完整流程，包含 LLM-as-judge 的实现。

import json
import os
from dataclasses import dataclass, asdict
from openai import OpenAI

client = OpenAI()

@dataclass
class EvalCase:
    task_id: str
    task_description: str
    expected_tools: list[str]
    expected_steps: list[str]
    expected_outcome: str
    difficulty: str  # easy / medium / hard
    category: str    # e.g. "code_search", "data_analysis"

@dataclass
class AgentResult:
    task_id: str
    trajectory: list[dict]  # List of {action, tool, input, output}
    final_output: str
    total_tokens: int
    total_steps: int

JUDGE_PROMPT = """你是一个 Agent 评估专家。请根据以下标准评估 Agent 的执行结果。

任务描述：{task_description}
期望结果：{expected_outcome}
Agent 最终输出：{actual_output}

Agent 执行轨迹：
{trajectory}

请从以下维度评分（1-5分）：
1. 任务完成度（task_completion）：Agent 是否完成了任务的核心目标？
2. 过程合理性（process_quality）：执行过程是否高效、没有冗余步骤？
3. 输出质量（output_quality）：最终输出是否准确、完整、格式正确？

返回 JSON 格式：
{{
    "task_completion": <1-5>,
    "process_quality": <1-5>,
    "output_quality": <1-5>,
    "reasoning": "<简短说明扣分原因>"
}}"""

def judge_result(case: EvalCase, result: AgentResult) -> dict:
    trajectory_str = json.dumps(result.trajectory, ensure_ascii=False, indent=2)
    prompt = JUDGE_PROMPT.format(
        task_description=case.task_description,
        expected_outcome=case.expected_outcome,
        actual_output=result.final_output,
        trajectory=trajectory_str,
    )
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
        temperature=0,
    )
    scores = json.loads(response.choices[0].message.content)

    # 加权计算综合分
    weighted_score = (
        scores["task_completion"] * 0.5
        + scores["process_quality"] * 0.3
        + scores["output_quality"] * 0.2
    )
    scores["weighted_total"] = round(weighted_score, 2)
    return scores

def run_eval_dataset(
    cases: list[EvalCase],
    agent_fn,  # Your agent function: (task_description) -> AgentResult
) -> list[dict]:
    results = []
    for case in cases:
        agent_result = agent_fn(case.task_description)
        judge_scores = judge_result(case, agent_result)
        results.append({
            "task_id": case.task_id,
            "difficulty": case.difficulty,
            "tokens_used": agent_result.total_tokens,
            "steps_taken": agent_result.total_steps,
            **judge_scores,
        })

    # 汇总统计
    avg_score = sum(r["weighted_total"] for r in results) / len(results)
    pass_rate = sum(1 for r in results if r["task_completion"] >= 4) / len(results)
    print(f"Eval Summary: avg_score={avg_score:.2f}, pass_rate={pass_rate:.1%}")
    return results

# 使用示例
if __name__ == "__main__":
    cases = [
        EvalCase(
            task_id="search_001",
            task_description="在代码库中找到处理用户认证的函数，列出它的参数和返回类型",
            expected_tools=["code_search"],
            expected_steps=["搜索认证相关代码", "定位函数定义", "提取签名信息"],
            expected_outcome="包含函数名、参数列表、返回类型的结构化信息",
            difficulty="easy",
            category="code_search",
        ),
    ]
    # results = run_eval_dataset(cases, my_agent_fn)

关于数据集本身的几个建议：

• 分层采样：按难度和类别分布采样，不要全是简单 case
• 包含边界 case：任务描述模糊的、工具不可用的、需要多步推理的
• 版本管理：数据集和 Agent 版本绑定，避免用新 Agent 跑旧预期
• 定期刷新：每季度审核一次数据集，淘汰过时的 case

生产实践模式二：回归测试流水线

有了数据集，下一步是把它接入 CI，让每次 prompt 或模型变更都自动跑一轮评估。

import json
import os
from datetime import datetime, timezone

REGRESSION_THRESHOLD = 0.3  # 单项得分下降超过此值即报警
GLOBAL_PASS_LINE = 3.5      # 全局平均分低于此值即失败

def load_baseline(baseline_path: str) -> dict:
    with open(baseline_path) as f:
        return json.load(f)

def compare_with_baseline(
    current_results: list[dict],
    baseline: dict,
) -> dict:
    regressions = []
    improvements = []

    for result in current_results:
        task_id = result["task_id"]
        if task_id not in baseline:
            continue

        prev = baseline[task_id]
        score_diff = result["weighted_total"] - prev["weighted_total"]

        if score_diff < -REGRESSION_THRESHOLD:
            regressions.append({
                "task_id": task_id,
                "previous": prev["weighted_total"],
                "current": result["weighted_total"],
                "delta": round(score_diff, 2),
                "reasoning": result.get("reasoning", ""),
            })
        elif score_diff > REGRESSION_THRESHOLD:
            improvements.append({
                "task_id": task_id,
                "previous": prev["weighted_total"],
                "current": result["weighted_total"],
                "delta": round(score_diff, 2),
            })

    avg_current = sum(r["weighted_total"] for r in current_results) / len(current_results)
    avg_baseline = sum(v["weighted_total"] for v in baseline.values()) / len(baseline)

    return {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "avg_current": round(avg_current, 2),
        "avg_baseline": round(avg_baseline, 2),
        "avg_delta": round(avg_current - avg_baseline, 2),
        "regressions": regressions,
        "improvements": improvements,
        "passed": avg_current >= GLOBAL_PASS_LINE and len(regressions) == 0,
    }

def save_as_baseline(results: list[dict], path: str):
    baseline = {r["task_id"]: r for r in results}
    baseline["_meta"] = {
        "created_at": datetime.now(timezone.utc).isoformat(),
        "num_cases": len(results),
    }
    with open(path, "w") as f:
        json.dump(baseline, f, ensure_ascii=False, indent=2)

# CI 集成示例
if __name__ == "__main__":
    # baseline_path = os.environ.get("EVAL_BASELINE_PATH", "eval_baselines/latest.json")
    # current = run_eval_dataset(cases, agent_fn)
    # baseline = load_baseline(baseline_path)
    # report = compare_with_baseline(current, baseline)
    # print(json.dumps(report, ensure_ascii=False, indent=2))
    # if not report["passed"]:
    #     sys.exit(1)
    print("Regression pipeline ready. Integrate into CI with EVAL_BASELINE_PATH env var.")

Weave 提供了类似功能的托管版本，自动追踪每次评估的结果并可视化趋势。如果你的团队已经在用 W&B 生态，Weave 是最省力的选择。

生产实践模式三：成本-质量 Pareto 分析

Agent 的成本不是线性的。从 GPT-4o 换成 GPT-4o-mini，成本可能降 10 倍，但成功率可能只降 5%。Pareto 分析帮你找到性价比最高的配置。

import json
from dataclasses import dataclass

@dataclass
class ModelConfig:
    name: str
    model_id: str
    cost_per_1k_input_tokens: float
    cost_per_1k_output_tokens: float

@dataclass
class EvalRun:
    config: ModelConfig
    avg_score: float
    pass_rate: float
    avg_input_tokens: int
    avg_output_tokens: int
    avg_latency_ms: int

def calculate_cost_per_task(run: EvalRun) -> float:
    input_cost = (run.avg_input_tokens / 1000) * run.config.cost_per_1k_input_tokens
    output_cost = (run.avg_output_tokens / 1000) * run.config.cost_per_1k_output_tokens
    return input_cost + output_cost

def pareto_analysis(runs: list[EvalRun]) -> list[dict]:
    analyzed = []
    for run in runs:
        cost = calculate_cost_per_task(run)
        analyzed.append({
            "model": run.config.name,
            "avg_score": run.avg_score,
            "pass_rate": run.pass_rate,
            "cost_per_task_usd": round(cost, 4),
            "avg_latency_ms": run.avg_latency_ms,
            "efficiency": round(run.avg_score / cost, 2) if cost > 0 else float("inf"),
        })

    # 按 cost 排序，找 Pareto 前沿
    analyzed.sort(key=lambda x: x["cost_per_task_usd"])

    pareto_frontier = []
    best_score = 0
    for item in analyzed:
        if item["avg_score"] > best_score:
            best_score = item["avg_score"]
            pareto_frontier.append(item)

    print("=== Cost-Quality Pareto Analysis ===")
    print(f"{'Model':<20} {'Score':>8} {'Pass%':>8} {'Cost/Task':>12} {'Efficiency':>12}")
    print("-" * 64)
    for item in analyzed:
        marker = " <-- Pareto" if item in pareto_frontier else ""
        print(
            f"{item['model']:<20} {item['avg_score']:>8.2f} "
            f"{item['pass_rate']:>7.1%} ${item['cost_per_task_usd']:>10.4f} "
            f"{item['efficiency']:>11.2f}{marker}"
        )

    return analyzed

if __name__ == "__main__":
    configs = [
        ModelConfig("gpt-4o-mini", "gpt-4o-mini", 0.00015, 0.0006),
        ModelConfig("gpt-4o", "gpt-4o", 0.0025, 0.01),
        ModelConfig("claude-sonnet", "claude-sonnet-4-20250514", 0.003, 0.015),
    ]
    # 模拟评估结果（实际应从 run_eval_dataset 获取）
    mock_runs = [
        EvalRun(configs[0], avg_score=3.6, pass_rate=0.72, avg_input_tokens=800, avg_output_tokens=300, avg_latency_ms=1200),
        EvalRun(configs[1], avg_score=4.2, pass_rate=0.88, avg_input_tokens=850, avg_output_tokens=350, avg_latency_ms=2500),
        EvalRun(configs[2], avg_score=4.3, pass_rate=0.90, avg_input_tokens=900, avg_output_tokens=320, avg_latency_ms=2800),
    ]
    pareto_analysis(mock_runs)

Pareto 分析的核心发现通常是：中等模型 + 好的 prompt 工程往往优于顶级模型 + 粗糙的 prompt。在做模型选择之前，先用现有 prompt 跑一轮 Pareto 分析，你会经常发现省钱的机会。

决策框架：什么时候用什么工具

场景	推荐工具	理由
快速原型评估，需要可视化	Weave	开箱即用的追踪和比较 UI
合规和安全审计	Giskard	内置偏见、幻觉、安全扫描
大规模基准对比	OpenHarness	标准化 benchmark 框架
Agent 行为 diff 追踪	Agent Diff	精确对比两次运行的轨迹差异
端到端生产评估	AWS Agent Eval	生产级流水线，与 AWS 生态集成
全链路可观测 + 评估	LMNR	Trace + Eval 一体化
定制化需求强	自建框架	用上面的代码模式搭建

选择依据不是"哪个功能多"，而是"哪个最贴合你当前的痛点"。早期用 Weave 快速看数据，中期建自有的回归流水线，后期根据合规需求引入 Giskard。

三个常见的坑

坑一：Judge-Model 对齐偏差

用 GPT-4o 当 judge 评估 GPT-4o 的输出，得分会系统性偏高。这在学术上叫 self-preference bias。

解决方案：

• Judge 模型和被评估模型用不同厂商（如用 Claude 评 GPT 的输出）
• 在评估 prompt 中明确评分标准，减少主观空间
• 定期用人工评估校准 judge 的打分倾向

坑二：评估数据集腐烂

评估数据集和代码一样会过期。当你的 Agent 增加了新工具、新能力时，旧的 test case 可能不再覆盖关键路径。

解决方案：

• 每次新增 Agent 能力时，同步新增对应的评估 case
• 设定"数据集保鲜期"，超过 3 个月的 case 需要重新审核
• 用 Agent Diff 追踪行为变化，识别数据集盲区

坑三：过拟合金标签示例

如果你的评估数据集只有 20-30 个 case，而你在上面反复调优 prompt，你实际上是在"应试"—— prompt 对这 20 个 case 表现完美，但对新场景可能完全崩溃。

解决方案：

• 评估集至少 100 个 case，覆盖不同难度和类别
• 划分 train/eval split，调优用的 case 和最终评估用的 case 不同
• 保留一组"隐藏集"不参与日常调优，只在发布前使用

总结

• Agent 评估必须分三层：组件级定位问题，交互级评估路径，结果级衡量最终效果。三层结合才能从"感觉变差了"变成"工具选择准确率下降 14%"。
• 先建数据集，再选指标。数据集的质量和覆盖率比评估算法的选择更重要。
• Judge 模型要和被评估模型异构，否则 self-preference bias 会让你的评估失去意义。
• 回归测试要接入 CI。不是"有空跑一下"，而是每次变更自动跑，和代码测试一个待遇。
• Pareto 分析能帮你省钱。跑一次成本-质量对比，你大概率会发现便宜的模型 + 好的 prompt 已经够用了。