【智能体测评-08】动手搭一套Agent测评流水线：开源工具链实战
这是“智能体测评”系列的第 8 篇。前 7 篇我们讲了方法论——测什么、怎么判、怎么建测评集、怎么分析轨迹。这篇不一样：我们把这些方法论变成一套可以跑的代码。你将看到一个最小可用的 Agent 测评流水线，从任务定义到报告生成，全链路开源实现。
为什么需要流水线，而不是脚本
很多团队的“Agent 测评”停留在这个阶段：写一个 Jupyter Notebook，手动调 20 次 API，把结果打印在 cell 输出里，肉眼看一下，截个图发群里说“新版本效果不错”。
这不是测评，这是算命。
一个真正能支撑版本决策的测评流水线需要解决五个问题：
- 可复现：同一份代码、同一批任务、同一个模型版本，任何时候跑结果都一样
- 可扩展：从 20 条任务到 2000 条不需要重写架构
- 可并行：2000 条任务串行跑要 6 小时，并行跑只要 20 分钟
- 可追溯：每次运行的配置、结果、轨迹都持久化，可以和历史版本对比
- 可自动化：接入 CI/CD，每次 PR 自动跑，不通过就阻断合并
这篇文章我们用 Python 搭一个满足这五个条件的最小流水线。不依赖任何商用平台，全部用开源库。
整体架构
┌─────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ 任务加载器 │───▶│ Agent 执行器 │───▶│ 评分引擎 │───▶│ 报告生成器 │
│ TaskLoader │ │ Runner │ │ Evaluator │ │ Reporter │
└─────────────┘ └──────────────┘ └──────────────┘ └──────────────┘
│ │ │ │
▼ ▼ ▼ ▼
tasks/*.yaml agents/*.py scorers/*.py reports/*.html
(测评任务定义) (Agent 适配器) (规则+LLM+人工) (HTML+JSON)四个组件，职责清晰：
- TaskLoader：从 YAML 文件加载测评任务，支持标签过滤和分层抽样
- Runner：并发执行 Agent，收集轨迹和结果，处理超时和重试
- Evaluator：三层评分——规则校验 → LLM-as-Judge → 人工标注队列
- Reporter：生成 HTML 报告和 JSON 原始数据，支持版本对比
我们逐个实现。
项目结构
agent-eval/
├── tasks/ # 测评任务（YAML）
│ ├── golden/ # 黄金集
│ │ ├── search_001.yaml
│ │ └── coding_001.yaml
│ └── regression/ # 回归集
├── agents/ # Agent 适配器
│ ├── base.py # 抽象基类
│ └── my_agent.py # 你的 Agent
├── scorers/ # 评分器
│ ├── rule_based.py # 规则评分
│ ├── llm_judge.py # LLM 评分
│ └── trajectory.py # 轨迹指标
├── reports/ # 输出目录
├── eval.py # 主入口
├── config.yaml # 配置文件
└── requirements.txt第一步：定义任务格式
我们用 YAML 定义测评任务。每条任务是一个自包含的文件，包含输入、期望输出、评分配置和标签。
# tasks/golden/search_001.yaml
id: search_001
title: "查询深圳明天天气并给出穿衣建议"
tags: [search, simple, P0]
difficulty: L1
capability: [perception, tool_use]
input:
message: "深圳明天天气怎么样？需要穿外套吗？"
context:
location: "深圳"
date: "2026-08-26"
# 评分配置
evaluation:
# 规则校验：输出必须包含这些要素
must_contain:
- "深圳"
- "温度" # 或 "气温"
# 禁止出现
must_not_contain:
- "无法"
- "抱歉"
# LLM Judge 评分维度
llm_rubric:
- dimension: accuracy
description: "天气数据是否准确引用了天气 API 的返回结果"
weight: 0.4
- dimension: helpfulness
description: "穿衣建议是否具体、可操作，而不是泛泛而谈"
weight: 0.3
- dimension: completeness
description: "是否同时回答了天气和穿衣两个问题"
weight: 0.3
# 轨迹约束
trajectory:
max_steps: 5
max_tokens: 10000
required_tools: ["web_search"]这个格式的设计原则：
- 声明式：只描述“要测什么”和“怎么判”，不涉及“怎么跑”
- 自包含：一条任务的所有信息在一个文件里，方便增删和 review
- 多维度：同时支持规则校验、LLM 评分和轨迹约束
- 可标签化：标签和难度字段支持切片分析
第二步：实现任务加载器
# agents/base.py（先定义基类，加载器会引用）
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any
@dataclass
class ToolCall:
name: str
arguments: dict
output: Any = None
latency_ms: int = 0
status: str = "success" # success / error
@dataclass
class Trajectory:
steps: list[ToolCall] = field(default_factory=list)
total_tokens: int = 0
total_latency_ms: int = 0
@property
def step_count(self) -> int:
return len(self.steps)
@property
def tool_names(self) -> list[str]:
return [s.name for s in self.steps]
@property
def error_count(self) -> int:
return sum(1 for s in self.steps if s.status == "error")
@property
def duplicate_calls(self) -> int:
"""重复调用次数"""
seen = []
dupes = 0
for s in self.steps:
key = (s.name, str(s.arguments))
if key in seen:
dupes += 1
seen.append(key)
return dupes
@dataclass
class AgentResult:
task_id: str
output: str
trajectory: Trajectory
success: bool # Agent 是否正常运行完成（不是任务是否成功）
error: str | None = None
class BaseAgent(ABC):
@abstractmethod
async def run(self, message: str, context: dict) -> AgentResult:
...# eval.py — TaskLoader
import yaml
from pathlib import Path
from dataclasses import dataclass
from typing import Any
@dataclass
class EvalTask:
id: str
title: str
tags: list[str]
difficulty: str
capability: list[str]
input: dict[str, Any]
evaluation: dict[str, Any]
raw: dict[str, Any]
class TaskLoader:
def __init__(self, tasks_dir: str = "tasks"):
self.tasks_dir = Path(tasks_dir)
def load(
self,
sets: list[str] | None = None, # ["golden", "regression"]
tags: list[str] | None = None,
difficulty: list[str] | None = None,
) -> list[EvalTask]:
"""加载测评任务，支持按集合、标签、难度过滤。"""
tasks = []
search_dirs = [self.tasks_dir / s for s in (sets or ["golden"])]
for d in search_dirs:
if not d.exists():
continue
for f in sorted(d.glob("*.yaml")):
with open(f) as fh:
data = yaml.safe_load(fh)
# 标签过滤：任务必须包含所有请求的标签
if tags and not all(t in data.get("tags", []) for t in tags):
continue
if difficulty and data.get("difficulty") not in difficulty:
continue
tasks.append(EvalTask(
id=data["id"],
title=data["title"],
tags=data.get("tags", []),
difficulty=data.get("difficulty", "L2"),
capability=data.get("capability", []),
input=data["input"],
evaluation=data.get("evaluation", {}),
raw=data,
))
return tasks第三步：实现并发执行器
# eval.py — Runner
import asyncio
import time
from agents.base import BaseAgent, AgentResult
class Runner:
def __init__(
self,
agent: BaseAgent,
concurrency: int = 10,
timeout: int = 120,
retries: int = 1,
):
self.agent = agent
self.semaphore = asyncio.Semaphore(concurrency)
self.timeout = timeout
self.retries = retries
async def run_all(self, tasks: list[EvalTask]) -> list[AgentResult]:
"""并发执行所有任务，带超时和重试。"""
tasks_coroutines = [self._run_one(t) for t in tasks]
results = await asyncio.gather(*tasks_coroutines)
return list(results)
async def _run_one(self, task: EvalTask) -> AgentResult:
async with self.semaphore:
last_error = None
for attempt in range(self.retries + 1):
try:
start = time.monotonic()
result = await asyncio.wait_for(
self.agent.run(
message=task.input["message"],
context=task.input.get("context", {}),
),
timeout=self.timeout,
)
elapsed = time.monotonic() - start
result.trajectory.total_latency_ms = int(elapsed * 1000)
return result
except asyncio.TimeoutError:
last_error = f"timeout after {self.timeout}s"
except Exception as e:
last_error = str(e)
if attempt < self.retries:
await asyncio.sleep(1)
return AgentResult(
task_id=task.id,
output="",
trajectory=None,
success=False,
error=last_error,
)一个最小的 Agent 适配器长这样（以 OpenAI API 为例）：
# agents/my_agent.py
import openai
from agents.base import BaseAgent, AgentResult, Trajectory, ToolCall
class MyAgent(BaseAgent):
def __init__(self, model: str = "gpt-4o"):
self.client = openai.AsyncOpenAI()
self.model = model
async def run(self, message: str, context: dict) -> AgentResult:
# 这里简化为直接对话，实际 Agent 会有多轮工具调用循环
# 每一轮工具调用记录为一个 ToolCall，追加到 trajectory.steps
response = await self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "你是一个有用的助手。"},
{"role": "user", "content": message},
],
)
output = response.choices[0].message.content
return AgentResult(
task_id="",
output=output,
trajectory=Trajectory(
steps=[], # 真实实现中记录每步工具调用
total_tokens=response.usage.total_tokens,
),
success=True,
)第四步：三层评分引擎
规则评分
# scorers/rule_based.py
import re
from dataclasses import dataclass
@dataclass
class RuleScore:
passed: bool
score: float # 0-1
details: list[str]
def rule_score(output: str, eval_config: dict) -> RuleScore:
"""基于规则的评分：关键词包含/禁止、正则匹配。"""
details = []
checks = []
# must_contain: 支持列表的列表（OR 关系）
for pattern in eval_config.get("must_contain", []):
if isinstance(pattern, list):
found = any(p in output for p in pattern)
details.append(f"[{'✓' if found else '✗'}] 包含 {' 或 '.join(pattern)}")
else:
found = pattern in output
details.append(f"[{'✓' if found else '✗'}] 包含 '{pattern}'")
checks.append(found)
for pattern in eval_config.get("must_not_contain", []):
found = pattern in output
details.append(f"[{'✓' if not found else '✗'}] 不包含 '{pattern}'")
checks.append(not found)
# regex 匹配
for pattern in eval_config.get("must_match", []):
found = bool(re.search(pattern, output))
details.append(f"[{'✓' if found else '✗'}] 匹配 '{pattern}'")
checks.append(found)
if not checks:
return RuleScore(passed=True, score=1.0, details=["无规则约束"])
score = sum(checks) / len(checks)
return RuleScore(
passed=all(checks),
score=score,
details=details,
)LLM-as-Judge
# scorers/llm_judge.py
import json
import openai
from dataclasses import dataclass
JUDGE_PROMPT = """你是一个严格的测评评分员。请根据以下评分标准（rubric）对 Agent 的回答进行评分。
任务：{task_title}
用户输入：{user_input}
Agent 回答：
{agent_output}
评分标准：
{rubric_text}
请以 JSON 格式输出评分结果：
{{
"scores": {{
"<dimension>": {{ "score": <1-5>, "reason": "<理由>" }}
}},
"overall_comment": "<总体评价>"
}}
评分要求：
1. 每个维度 1-5 分：1=完全不满足，3=基本满足，5=超出预期
2. 必须给出具体理由，不能只给分数
3. 严格按 rubric 评分，不要加入 rubric 之外的标准
"""
@dataclass
class JudgeScore:
dimension: str
score: int
weight: float
reason: str
async def llm_judge(
task_title: str,
user_input: str,
agent_output: str,
rubric: list[dict],
model: str = "gpt-4o",
) -> tuple[float, list[JudgeScore], str]:
"""返回加权总分（0-1）、各维度分数和总体评价。"""
rubric_text = "\n".join(
f"- {r['dimension']}（权重 {r['weight']}）：{r['description']}"
for r in rubric
)
client = openai.AsyncOpenAI()
response = await client.chat.completions.create(
model=model,
messages=[{
"role": "user",
"content": JUDGE_PROMPT.format(
task_title=task_title,
user_input=user_input,
agent_output=agent_output,
rubric_text=rubric_text,
),
}],
response_format={"type": "json_object"},
temperature=0.1, # 低温度减少随机波动
)
result = json.loads(response.choices[0].message.content)
scores = []
weighted_sum = 0
total_weight = 0
for r in rubric:
dim = r["dimension"]
s = result["scores"].get(dim, {})
score_val = s.get("score", 1)
scores.append(JudgeScore(
dimension=dim,
score=score_val,
weight=r["weight"],
reason=s.get("reason", ""),
))
weighted_sum += score_val * r["weight"]
total_weight += r["weight"]
# 归一化到 0-1（原始分数 1-5）
final_score = (weighted_sum / total_weight - 1) / 4 if total_weight else 0
return final_score, scores, result.get("overall_comment", "")轨迹指标
# scorers/trajectory.py
from dataclasses import dataclass
from agents.base import Trajectory
@dataclass
class TrajectoryScore:
step_count: int
error_count: int
duplicate_calls: int
total_tokens: int
total_latency_ms: int
violations: list[str] # 违反轨迹约束的项
def score_trajectory(
traj: Trajectory,
eval_config: dict,
) -> TrajectoryScore:
"""计算轨迹指标，检查轨迹约束。"""
constraints = eval_config.get("trajectory", {})
violations = []
if "max_steps" in constraints and traj.step_count > constraints["max_steps"]:
violations.append(
f"步数 {traj.step_count} 超过上限 {constraints['max_steps']}"
)
if "max_tokens" in constraints and traj.total_tokens > constraints["max_tokens"]:
violations.append(
f"token {traj.total_tokens} 超过上限 {constraints['max_tokens']}"
)
for tool in constraints.get("required_tools", []):
if tool not in traj.tool_names:
violations.append(f"未使用必需工具：{tool}")
return TrajectoryScore(
step_count=traj.step_count,
error_count=traj.error_count,
duplicate_calls=traj.duplicate_calls,
total_tokens=traj.total_tokens,
total_latency_ms=traj.total_latency_ms,
violations=violations,
)组合评分
# eval.py — Evaluator
from dataclasses import dataclass, field
from scorers.rule_based import rule_score
from scorers.llm_judge import llm_judge
from scorers.trajectory import score_trajectory
@dataclass
class TaskScore:
task_id: str
final_score: float # 0-1
rule_score: float
llm_score: float | None
rule_passed: bool
trajectory_violations: list[str]
details: dict = field(default_factory=dict)
class Evaluator:
def __init__(self, judge_model: str = "gpt-4o"):
self.judge_model = judge_model
async def evaluate(
self,
task: EvalTask,
result: AgentResult,
) -> TaskScore:
if not result.success:
return TaskScore(
task_id=task.id,
final_score=0,
rule_score=0,
llm_score=None,
rule_passed=False,
trajectory_violations=[result.error or "execution failed"],
details={"error": result.error},
)
# 第一层：规则评分
rs = rule_score(result.output, task.evaluation)
# 第二层：LLM Judge（仅在有 rubric 时）
llm_s = None
rubric = task.evaluation.get("llm_rubric", [])
if rubric:
llm_s, judge_scores, comment = await llm_judge(
task_title=task.title,
user_input=task.input["message"],
agent_output=result.output,
rubric=rubric,
model=self.judge_model,
)
else:
judge_scores = []
comment = ""
# 轨迹评分
if result.trajectory:
ts = score_trajectory(result.trajectory, task.evaluation)
else:
ts = None
# 组合最终分数：
# 有 LLM rubric 时：规则 0.3 + LLM 0.7（规则做硬约束）
# 无 LLM rubric 时：规则 1.0
# 轨迹违规扣分
if llm_s is not None:
final = rs.score * 0.3 + llm_s * 0.7
else:
final = rs.score
# 轨迹违规每条扣 0.1，最多扣到 0
if ts:
final = max(0, final - len(ts.violations) * 0.1)
return TaskScore(
task_id=task.id,
final_score=round(final, 3),
rule_score=round(rs.score, 3),
llm_score=round(llm_s, 3) if llm_s is not None else None,
rule_passed=rs.passed,
trajectory_violations=ts.violations if ts else [],
details={
"rule_details": rs.details,
"judge_scores": [
{"dim": s.dimension, "score": s.score, "reason": s.reason}
for s in judge_scores
],
"trajectory": {
"steps": ts.step_count if ts else 0,
"errors": ts.error_count if ts else 0,
"duplicates": ts.duplicate_calls if ts else 0,
"tokens": ts.total_tokens if ts else 0,
"latency_ms": ts.total_latency_ms if ts else 0,
},
"comment": comment,
},
)第五步：报告生成
# eval.py — Reporter
import json
from datetime import datetime
from pathlib import Path
from collections import defaultdict
class Reporter:
def __init__(self, output_dir: str = "reports"):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
def generate(
self,
scores: list[TaskScore],
tasks: list[EvalTask],
agent_name: str,
model: str,
) -> str:
"""生成 JSON + HTML 报告，返回报告路径。"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
# 构建任务映射
task_map = {t.id: t for t in tasks}
# 汇总指标
summary = self._summarize(scores, task_map)
report = {
"timestamp": timestamp,
"agent": agent_name,
"model": model,
"total_tasks": len(scores),
"summary": summary,
"results": [
{
"task_id": s.task_id,
"title": task_map[s.task_id].title,
"difficulty": task_map[s.task_id].difficulty,
"tags": task_map[s.task_id].tags,
**s.__dict__,
}
for s in scores
],
}
# 写 JSON
json_path = self.output_dir / f"eval_{timestamp}.json"
with open(json_path, "w") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
# 写 HTML
html_path = self.output_dir / f"eval_{timestamp}.html"
html_path.write_text(self._render_html(report))
return str(html_path)
def _summarize(self, scores, task_map):
total = len(scores)
avg_score = sum(s.final_score for s in scores) / total if total else 0
passed = sum(1 for s in scores if s.rule_passed and not s.trajectory_violations)
# 按难度切片
by_difficulty = defaultdict(list)
for s in scores:
d = task_map[s.task_id].difficulty
by_difficulty[d].append(s.final_score)
# 按标签切片
by_tag = defaultdict(list)
for s in scores:
for tag in task_map[s.task_id].tags:
by_tag[tag].append(s.final_score)
# 轨迹指标汇总
all_steps = [s.details.get("trajectory", {}).get("steps", 0) for s in scores]
all_tokens = [s.details.get("trajectory", {}).get("tokens", 0) for s in scores]
all_latency = [s.details.get("trajectory", {}).get("latency_ms", 0) for s in scores]
return {
"avg_score": round(avg_score, 3),
"pass_rate": round(passed / total, 3) if total else 0,
"passed_count": passed,
"total_count": total,
"avg_steps": round(sum(all_steps) / total, 1) if total else 0,
"avg_tokens": int(sum(all_tokens) / total) if total else 0,
"avg_latency_ms": int(sum(all_latency) / total) if total else 0,
"by_difficulty": {
k: round(sum(v) / len(v), 3) for k, v in by_difficulty.items()
},
"by_tag": {
k: round(sum(v) / len(v), 3) for k, v in by_tag.items()
},
}
def _render_html(self, report: dict) -> str:
"""生成一个简洁的自包含 HTML 报告。"""
s = report["summary"]
rows = ""
for r in report["results"]:
color = "#16a34a" if r["final_score"] >= 0.7 else "#dc2626" if r["final_score"] < 0.4 else "#ca8a04"
traj = r["details"].get("trajectory", {})
rows += f"""
<tr>
<td>{r['task_id']}</td>
<td>{r['title']}</td>
<td>{r['difficulty']}</td>
<td style="color:{color};font-weight:bold">{r['final_score']:.1%}</td>
<td>{traj.get('steps', '-')}</td>
<td>{traj.get('tokens', '-')}</td>
<td>{traj.get('latency_ms', '-')}ms</td>
<td>{', '.join(r['trajectory_violations']) or '—'}</td>
</tr>"""
return f"""<!DOCTYPE html>
<html><head><meta charset="utf-8"><title>Eval Report</title>
<style>
body {{ font-family: -apple-system, sans-serif; max-width: 1100px; margin: 40px auto; padding: 0 20px; color: #1a1a1a; }}
h1 {{ font-size: 24px; }}
.cards {{ display: grid; grid-template-columns: repeat(4, 1fr); gap: 16px; margin: 24px 0; }}
.card {{ background: #f8fafc; border-radius: 8px; padding: 16px; }}
.card .label {{ font-size: 12px; color: #64748b; text-transform: uppercase; }}
.card .value {{ font-size: 28px; font-weight: 700; margin-top: 4px; }}
table {{ width: 100%; border-collapse: collapse; margin-top: 24px; font-size: 14px; }}
th, td {{ padding: 8px 12px; border-bottom: 1px solid #e2e8f0; text-align: left; }}
th {{ background: #f1f5f9; font-weight: 600; }}
</style></head><body>
<h1>Agent 测评报告 — {report['agent']} / {report['model']}</h1>
<p style="color:#64748b">{report['timestamp']}</p>
<div class="cards">
<div class="card"><div class="label">平均分</div><div class="value">{s['avg_score']:.1%}</div></div>
<div class="card"><div class="label">通过率</div><div class="value">{s['pass_rate']:.1%}</div></div>
<div class="card"><div class="label">平均步数</div><div class="value">{s['avg_steps']}</div></div>
<div class="card"><div class="label">平均 Token</div><div class="value">{s['avg_tokens']:,}</div></div>
</div>
<table>
<thead><tr><th>ID</th><th>任务</th><th>难度</th><th>得分</th><th>步数</th><th>Token</th><th>时延</th><th>违规</th></tr></thead>
<tbody>{rows}</tbody>
</table>
</body></html>"""第六步：主入口和 CI 集成
# eval.py — 主入口
import asyncio
import argparse
from agents.my_agent import MyAgent
async def main():
parser = argparse.ArgumentParser()
parser.add_argument("--sets", nargs="+", default=["golden"])
parser.add_argument("--tags", nargs="*")
parser.add_argument("--difficulty", nargs="*")
parser.add_argument("--concurrency", type=int, default=10)
parser.add_argument("--model", default="gpt-4o")
parser.add_argument("--threshold", type=float, default=0.7,
help="通过率低于此值则退出码为 1")
args = parser.parse_args()
# 1. 加载任务
loader = TaskLoader()
tasks = loader.load(sets=args.sets, tags=args.tags, difficulty=args.difficulty)
print(f"📋 加载了 {len(tasks)} 条任务")
# 2. 执行
agent = MyAgent(model=args.model)
runner = Runner(agent, concurrency=args.concurrency)
print(f"🚀 开始执行（并发 {args.concurrency}）...")
results = await runner.run_all(tasks)
# 3. 评分
evaluator = Evaluator()
scores = await asyncio.gather(*[
evaluator.evaluate(t, r) for t, r in zip(tasks, results)
])
# 4. 报告
reporter = Reporter()
report_path = reporter.generate(
scores, tasks,
agent_name="MyAgent",
model=args.model,
)
print(f"📊 报告已生成：{report_path}")
# 5. 门禁判断
summary = json.loads(open(report_path.replace(".html", ".json")).read())["summary"]
print(f" 平均分：{summary['avg_score']:.1%}")
print(f" 通过率：{summary['pass_rate']:.1%}")
if summary["avg_score"] < args.threshold:
print(f"❌ 平均分低于阈值 {args.threshold:.0%}，测评未通过")
exit(1)
print("✅ 测评通过")
if __name__ == "__main__":
asyncio.run(main())接入 GitHub Actions，每次 PR 自动跑黄金集：
# .github/workflows/eval.yml
name: Agent Eval
on: [pull_request]
jobs:
eval:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- run: pip install -r requirements.txt
- name: Run golden set
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: python eval.py --sets golden --threshold 0.7
- name: Upload report
if: always()
uses: actions/upload-artifact@v4
with:
name: eval-report
path: reports/版本对比
单次分数意义有限，真正的决策信号来自版本间对比。加一个对比脚本：
# compare.py
import json
import sys
def compare(old_path: str, new_path: str):
old = json.load(open(old_path))
new = json.load(open(new_path))
old_map = {r["task_id"]: r for r in old["results"]}
new_map = {r["task_id"]: r for r in new["results"]}
print(f"{'任务':<30} {'旧版':>6} → {'新版':<6} {'变化':>6}")
print("-" * 60)
regressions = []
improvements = []
for tid in sorted(set(old_map) & set(new_map)):
o = old_map[tid]["final_score"]
n = new_map[tid]["final_score"]
delta = n - o
symbol = "📈" if delta > 0.05 else "📉" if delta < -0.05 else " "
print(f"{old_map[tid]['title'][:28]:<30} {o:>5.0%} → {n:<5.0%} {symbol} {delta:+.0%}")
if delta < -0.05:
regressions.append(tid)
if delta > 0.05:
improvements.append(tid)
print(f"\n📈 改善：{len(improvements)} 条")
print(f"📉 退化：{len(regressions)} 条")
if regressions:
print(f"\n⚠️ 退化任务：{', '.join(regressions)}")
sys.exit(1)用法：
python eval.py --model gpt-4o # 旧版本
# 改了 prompt 或模型后
python eval.py --model gpt-4o-2026-08 # 新版本
python compare.py reports/eval_old.json reports/eval_new.json生产环境的扩展方向
上面的代码是一个最小可用版本，大约 500 行 Python。在真实生产环境中，你可能还需要：
| 需求 | 推荐方案 |
|---|---|
| 轨迹可视化 | 接入 Langfuse 或 LangSmith（OpenTelemetry 兼容） |
| 测评集管理 | 用 Braintrust 或自建 Web UI（YAML 文件 + Git 也能撑到 2000 条） |
| LLM Judge 缓存 | 相同输入+模型+prompt 命中缓存，省 60% 以上 judge 成本 |
| 分布式执行 | Celery / RQ / Ray，支持数千条任务跨机器并行 |
| 人工标注队列 | 集成 Label Studio 或 Argilla，LLM 评分低置信度的自动进人工队列 |
| 趋势看板 | 把 JSON 报告导入 Metabase / Grafana，画长期趋势图 |
| A/B 实验 | 同一批任务跑两个版本，配对 t 检验或 bootstrap 置信区间 |
| 故障注入 | 在 Agent 执行层 mock 工具返回（超时、空结果、错误码），测恢复能力 |
常见问题
Q：这个流水线和 LangSmith / Braintrust / Langfuse 有什么区别？
这些平台做的是“可观测性 + 测评”的整合，功能更全（轨迹可视化、数据集管理、团队协作），但都是 SaaS 或需要部署服务。我们的代码适合不想把数据发到第三方、想用 Git 管理测评任务、需要完全自定义评分逻辑的团队。两者不矛盾——可以先用这个最小流水线跑起来，规模大了再迁移到平台，任务格式和评分逻辑是可以复用的。
Q：LLM Judge 成本怎么控制？
1000 条任务，每条 judge 一次 gpt-4o，大约 $5-15。优化手段：
- 规则能判的不调 LLM
- Judge 用便宜模型（gpt-4o-mini 做初筛，只把分歧大的送 gpt-4o 仲裁）
- 加缓存：Agent 输出没变就不重新 judge
- 只在黄金集上全量 judge，回归集用规则为主
Q：测评要跑多久？
100 条任务，并发 10，平均每条 15 秒，大约 2-3 分钟。1000 条任务并发 20，大约 15-20 分钟。瓶颈通常是 Agent 执行（LLM 推理 + 工具调用），不是评分。
小结
这篇我们把前 7 篇的方法论落成了代码：
- TaskLoader 加载 YAML 任务，支持标签和难度过滤（第 7 篇的测评集管理）
- Runner 并发执行，带超时和重试（第 6 篇的轨迹采集）
- Evaluator 三层评分：规则 + LLM Judge + 轨迹约束（第 5、6 篇的评分方法）
- Reporter 生成 HTML 报告，支持按难度和标签切片（第 3 篇的四层框架）
- compare.py 做版本配对对比，退化自动告警
- GitHub Actions 接入 CI，黄金集不通过阻断 PR
这套代码的核心价值不是功能多全，而是它建立了一个可迭代的骨架——你可以从 50 条任务、一个 Agent 开始，随着团队成长逐步替换组件（加缓存、换平台、加分布式），而不需要推倒重来。
下一篇【智能体测评-09】三类典型场景的测评方案：Coding Agent、客服 Agent、深度研究 Agent，我们把这套流水线应用到三种不同类型的 Agent 上，看看评分标准、任务设计和轨迹指标在不同场景下有什么本质差异。