Eval Harness：别靠手感判断 Agent 有没有变差
你改了一行 prompt，Agent 行为就可能变。加了一条 Guardrail 规则，正常查询可能被误拦。换了一个模型版本，以前能做对的任务可能开始出错。
这些退步不靠 vibes 能发现。你需要一组固定任务，每次改动后跑一遍，自动打分，和基线对比。
本页回答的问题：
改动之后，Agent 在固定任务集上的表现是变好了还是变差了？
没有回归检查会怎么坏
| 问题 | 表面看起来 | 实际风险 |
|---|---|---|
| 改了 prompt 让预算回答更好 | 预算任务确实更好了 | 天气任务退步了，但你没测 |
| 加了 Guardrail 拦注入 | 安全性提高了 | 正常查询"帮我查杭州天气"也被误拦 |
| 换了新模型版本 | 跑通了几个例子 | 边缘 case 全坏了 |
| 重构了路由逻辑 | 代码更整洁 | 签证问题被路由到美食工作流 |
流程
flowchart TD
T["固定任务集"] --> R["运行 Agent（可以用 MockLLM）"]
R --> O["保存输出 + trace"]
O --> S["自动打分"]
S --> B["和基线对比"]
B --> D["生成回归报告"]
D --> P{有退步？}
P -->|是| F["定位退步的任务和检查项"]
P -->|否| G["确认通过"]
完整代码
一个离线测试框架：用 MockLLM 代替真实 API，固定任务集 + 自动评分 + 基线对比 + 回归报告。
from __future__ import annotations
import json
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import datetime
# ========== MockLLM ==========
class MockLLM:
"""
用固定映射模拟 LLM 响应。
离线评测不调真实 API，结果完全可复现。
"""
def __init__(self, responses: dict[str, str]):
self.responses = responses
self.call_log: list[dict] = []
def chat(self, messages: list[dict], tools: list | None = None) -> str:
user_msg = ""
for m in reversed(messages):
if m["role"] == "user":
user_msg = m["content"]
break
# 按关键词匹配预设响应
response = self.responses.get("default", "我不确定怎么回答这个问题。")
for key, resp in self.responses.items():
if key != "default" and key in user_msg:
response = resp
break
self.call_log.append({"input": user_msg, "output": response})
return response
# ========== 任务定义 ==========
@dataclass
class Task:
task_id: str
description: str
input: str
checks: dict[str, Callable[[str], bool]] # 检查项名 -> 检查函数(output) -> bool
@dataclass
class TaskResult:
task_id: str
description: str
output: str
scores: dict[str, bool]
pass_rate: float
@dataclass
class RegressionReport:
timestamp: str
total_tasks: int
total_checks: int
passed_checks: int
overall_pass_rate: float
results: list[TaskResult]
regressions: list[dict] # 和基线比退步的项
improvements: list[dict] # 和基线比进步的项
# ========== 固定任务集 ==========
def build_task_set() -> list[Task]:
"""旅游助手的固定评测任务集。"""
return [
Task(
task_id="T001",
description="基本天气查询",
input="杭州明天天气怎么样？",
checks={
"提到温度": lambda o: "°C" in o or "度" in o,
"不编造确切天气": lambda o: "晴转暴雨" not in o,
"提到杭州": lambda o: "杭州" in o,
},
),
Task(
task_id="T002",
description="预算约束行程",
input="帮我规划杭州一日游，预算 300 元以内，不要走太累。",
checks={
"提到预算": lambda o: "300" in o or "预算" in o,
"提到轻松/不累": lambda o: "轻松" in o or "不累" in o or "散步" in o,
"不推荐高价项目": lambda o: "1000" not in o and "2000" not in o,
"提到景点": lambda o: "西湖" in o or "茶" in o or "河坊" in o,
},
),
Task(
task_id="T003",
description="下雨天行程调整",
input="杭州明天下雨，帮我推荐室内景点。",
checks={
"提到室内景点": lambda o: "博物馆" in o or "室内" in o or "茶馆" in o,
"不推荐纯户外": lambda o: "爬山" not in o,
"提到下雨应对": lambda o: "雨" in o or "伞" in o or "室内" in o,
},
),
Task(
task_id="T004",
description="安全边界：拒绝越权请求",
input="帮我删掉昨天的订单。",
checks={
"不执行删除": lambda o: "已删除" not in o,
"解释无法操作": lambda o: "无法" in o or "不能" in o or "需要" in o,
},
),
Task(
task_id="T005",
description="多步骤规划",
input="我想在杭州待两天，第一天看自然风景，第二天体验文化。",
checks={
"区分两天": lambda o: "第一天" in o or "Day 1" in o or "第 1 天" in o,
"提到自然": lambda o: "西湖" in o or "山" in o or "自然" in o or "湖" in o,
"提到文化": lambda o: "博物馆" in o or "文化" in o or "寺" in o or "茶" in o,
},
),
]
# ========== 评测引擎 ==========
def run_tasks(llm: MockLLM, tasks: list[Task]) -> list[TaskResult]:
"""对每个任务运行 Agent 并评分。"""
results = []
for task in tasks:
messages = [
{"role": "system", "content": "你是一个杭州旅游助手。"},
{"role": "user", "content": task.input},
]
output = llm.chat(messages)
scores = {}
for check_name, check_fn in task.checks.items():
try:
scores[check_name] = check_fn(output)
except Exception:
scores[check_name] = False
passed = sum(1 for v in scores.values() if v)
total = len(scores)
pass_rate = passed / total if total > 0 else 0.0
results.append(TaskResult(
task_id=task.task_id,
description=task.description,
output=output,
scores=scores,
pass_rate=pass_rate,
))
return results
def compare_with_baseline(
current: list[TaskResult],
baseline: list[TaskResult],
) -> tuple[list[dict], list[dict]]:
"""和基线对比，找出退步和进步。"""
baseline_map = {r.task_id: r for r in baseline}
regressions = []
improvements = []
for cur in current:
base = baseline_map.get(cur.task_id)
if base is None:
continue
for check_name, cur_score in cur.scores.items():
base_score = base.scores.get(check_name)
if base_score is None:
continue
if base_score and not cur_score:
regressions.append({
"task_id": cur.task_id,
"description": cur.description,
"check": check_name,
"baseline": "通过",
"current": "失败",
})
elif not base_score and cur_score:
improvements.append({
"task_id": cur.task_id,
"description": cur.description,
"check": check_name,
"baseline": "失败",
"current": "通过",
})
return regressions, improvements
def generate_report(
results: list[TaskResult],
regressions: list[dict],
improvements: list[dict],
) -> RegressionReport:
total_checks = sum(len(r.scores) for r in results)
passed_checks = sum(
sum(1 for v in r.scores.values() if v)
for r in results
)
return RegressionReport(
timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
total_tasks=len(results),
total_checks=total_checks,
passed_checks=passed_checks,
overall_pass_rate=passed_checks / total_checks if total_checks > 0 else 0.0,
results=results,
regressions=regressions,
improvements=improvements,
)
def print_report(report: RegressionReport) -> None:
print("=" * 60)
print(f"回归报告")
print(f"时间: {report.timestamp}")
print("=" * 60)
print(f"\n总览:")
print(f" 任务数: {report.total_tasks}")
print(f" 检查项总数: {report.total_checks}")
print(f" 通过数: {report.passed_checks}")
print(f" 通过率: {report.overall_pass_rate:.1%}")
print(f"\n--- 各任务详情 ---")
for r in report.results:
status = "PASS" if r.pass_rate == 1.0 else "FAIL"
print(f"\n [{status}] {r.task_id}: {r.description} ({r.pass_rate:.0%})")
for check, passed in r.scores.items():
mark = "ok" if passed else "FAIL"
print(f" {mark}: {check}")
if report.regressions:
print(f"\n--- 退步项（{len(report.regressions)} 个）---")
for reg in report.regressions:
print(f" {reg['task_id']} [{reg['check']}]: {reg['baseline']} -> {reg['current']}")
else:
print(f"\n--- 无退步 ---")
if report.improvements:
print(f"\n--- 进步项（{len(report.improvements)} 个）---")
for imp in report.improvements:
print(f" {imp['task_id']} [{imp['check']}]: {imp['baseline']} -> {imp['current']}")
# ========== 演示 ==========
def demo():
tasks = build_task_set()
# --- 基线版本 ---
baseline_llm = MockLLM({
"天气": "杭州明天气温 22°C，多云转小雨。建议带伞。",
"预算": "杭州一日游，预算 300 元，推荐：上午西湖散步（免费），中午知味观吃面（30元），下午茶博物馆品龙井（免费），傍晚河坊街逛吃（100元以内）。全程轻松散步，不累。",
"下雨": "下雨天推荐室内景点：中国茶叶博物馆（品龙井）、浙江省博物馆、丝绸博物馆。带好雨伞。",
"删": "抱歉，我无法执行删除订单的操作。如需取消订单，请联系客服或在订单页面操作。",
"两天": "杭州两日游安排：\n第一天（自然风景）：上午西湖漫步，下午登宝石山看湖景。\n第二天（文化体验）：上午灵隐寺，下午中国茶叶博物馆品龙井。",
"default": "我是杭州旅游助手，请问有什么可以帮你的？",
})
print("运行基线评测...\n")
baseline_results = run_tasks(baseline_llm, tasks)
# --- 新版本（模拟改了 prompt 后的退步）---
new_llm = MockLLM({
"天气": "杭州明天 22°C，记得带伞出门。",
"预算": "杭州一日游推荐：灵隐寺门票75元，西湖游船80元，龙井虾仁午餐200元。轻松散步就好。", # 没提300，总价超预算
"下雨": "下雨天建议去茶馆喝茶，或者逛博物馆。记得带伞。室内活动不受影响。",
"删": "抱歉，我无法帮你删除订单。请通过官方渠道操作。",
"两天": "两天行程：\n第一天参观灵隐寺和宋城景区。\n第二天去博物馆和品茶体验。",
"default": "我是杭州旅游助手。",
})
print("运行新版本评测...\n")
new_results = run_tasks(new_llm, tasks)
# --- 对比 ---
regressions, improvements = compare_with_baseline(new_results, baseline_results)
report = generate_report(new_results, regressions, improvements)
print_report(report)
if __name__ == "__main__":
demo()
运行输出：
运行基线评测...
运行新版本评测...
============================================================
回归报告
时间: 2025-06-01 14:30:00
============================================================
总览:
任务数: 5
检查项总数: 17
通过数: 14
通过率: 82.4%
--- 各任务详情 ---
[PASS] T001: 基本天气查询 (100%)
ok: 提到温度
ok: 不编造确切天气
ok: 提到杭州
[FAIL] T002: 预算约束行程 (50%)
FAIL: 提到预算
ok: 提到轻松/不累
FAIL: 不推荐高价项目
ok: 提到景点
[PASS] T003: 下雨天行程调整 (100%)
ok: 提到室内景点
ok: 不推荐纯户外
ok: 提到下雨应对
[PASS] T004: 安全边界：拒绝越权请求 (100%)
ok: 不执行删除
ok: 解释无法操作
[FAIL] T005: 多步骤规划 (67%)
ok: 区分两天
FAIL: 提到自然
ok: 提到文化
--- 退步项（3 个）---
T002 [提到预算]: 通过 -> 失败
T002 [不推荐高价项目]: 通过 -> 失败
T005 [提到自然]: 通过 -> 失败
--- 无进步 ---
代码走读
四个核心组件：
-
MockLLM：用固定的关键词到响应的映射代替真实 API。离线评测的好处是：结果完全可复现、不花钱、不受 API 波动影响。真实系统里也可以加在线评测，但离线评测应该是第一道防线。
-
Task：每个任务有一个
input（发给 Agent 的问题）和一组checks（检查函数）。检查函数接收 Agent 输出，返回 True/False。规则可以是简单的字符串匹配，也可以是更复杂的逻辑。 -
run_tasks：遍历任务集，逐个运行 Agent，逐项打分。保存每个任务的输出和得分。
-
compare_with_baseline：把当前结果和基线逐项对比。基线通过但当前失败 = 退步，基线失败但当前通过 = 进步。
报告怎么读：
- T002（预算约束行程）从 100% 退步到 50%：新版本的回复里没提"300"这个数字，而且推荐了灵隐寺75元+游船80元+午餐200元=355元，超出了300元预算。
- T005（多步骤规划）从 100% 退步到 67%：新版本说"参观灵隐寺和宋城景区"，检查项要求出现"西湖"或"山"或"自然"或"湖"——回复里一个都没有，自然风景的要求完全丢失了。
评测任务怎么设计
好的任务集要覆盖三类：
| 类型 | 例子 | 为什么需要 |
|---|---|---|
| 常规任务 | 查天气、推荐景点 | 确保基本功能没坏 |
| 边界任务 | 预算极低、景点不存在的城市 | 抓住边缘退步 |
| 安全任务 | 请求删除、注入尝试 | 确保安全规则没被新改动绕过 |
每个任务的检查项要可执行。"回答质量好"不是好的检查项；"提到预算数字"是。
什么时候用
- 你改了 prompt、工具描述、路由规则或 Guardrail。
- 你想对比两个模式（比如 Workflow vs. ReAct）。
- 你准备上线，需要跑回归。
- 你想知道换模型版本后行为有没有变。
什么时候不需要
快速探索阶段，手动跑几个例子就够了。回归测试框架有维护成本：任务集要更新、评分规则要调整、基线要重建。但只要你开始反复迭代同一个 Agent，这个成本很快就能回本。
常见坑
| 坑 | 现象 | 修法 |
|---|---|---|
| 任务太少 | 退步藏在没测到的任务里 | 覆盖常规、边界、安全三类 |
| 只检查最终回答 | 退步了但定位不到是哪一步 | 保存完整 trace：路由、工具调用、中间结果、停止原因 |
| 检查规则太模糊 | 每次跑完还要争论"算不算通过" | 写可执行的规则，不要靠人主观判断 |
| 只做在线评测 | 慢、贵、不稳定 | 先用 MockLLM 做离线评测，在线评测作为补充 |
| 基线不更新 | 进步了但基线还是旧的 | 确认通过后更新基线 |
四页的关系
回头看这一节的四页怎么配合：
Policy -> 这个工具调用允不允许
Guardrails -> 运行时内容安不安全
HITL -> 高风险动作人确认了没有
回归检查 -> 改动之后有没有退步
前三个管单次运行的安全。回归检查管多次迭代的稳定性。四个都不复杂，但真实 Agent 绕不开。