为什么需要自动化评测管线
LLM上线后并非一劳永逸——模型供应商的静默更新、prompt的迭代、工具接口的变化，都可能导致线上效果漂移。人工评测周期长、成本高、不可重复。自动化评测管线是保障线上质量的唯一可行路径。
一条完整的评测管线应该做到：定时执行、自动评分、异常告警、趋势可视化。
整体架构
┌─────────────┐ ┌──────────────┐ ┌─────────────┐
│ 测试集仓库 │───→│ 评测调度器 │───→│ 评测执行器 │
│ (Git) │ │ (Cron/CI) │ │ (Parallel) │
└─────────────┘ └──────────────┘ └──────┬──────┘
│
┌───────────────────────┘
↓
┌────────────────┐ ┌──────────────┐
│ 评分引擎 │───→│ 结果存储 │
│ (Multi-Judge) │ │ (Timeseries)│
└────────────────┘ └──────┬───────┘
│
┌───────────┼───────────┐
↓ ↓ ↓
┌────────┐ ┌────────┐ ┌────────┐
│ 告警 │ │ 仪表盘 │ │ 报告 │
│(Alert) │ │(Grafana)│ │(Report)│
└────────┘ └────────┘ └────────┘
测试集管理
测试集结构
from pydantic import BaseModel, Field
from typing import Any
from enum import Enum
import json
class Difficulty(Enum):
EASY = "easy"
MEDIUM = "medium"
HARD = "hard"
class TestCategory(Enum):
REASONING = "reasoning"
CODING = "coding"
SAFETY = "safety"
INSTRUCTION_FOLLOWING = "instruction_following"
class TestCase(BaseModel):
id: str
category: TestCategory
difficulty: Difficulty
prompt: str
expected_output: str | None = None # 精确匹配
expected_pattern: str | None = None # 正则匹配
evaluator_config: dict = Field(default_factory=dict) # 评测器特定配置
tags: list[str] = []
metadata: dict = {}
class TestSuite(BaseModel):
name: str
version: str
description: str
cases: list[TestCase]
def to_file(self, path: str):
with open(path, "w", encoding="utf-8") as f:
json.dump(self.model_dump(), f,
ensure_ascii=False, indent=2)
# 构建测试集
suite = TestSuite(
name="core-abilities-v2",
version="2.1.0",
description="核心能力评测集",
cases=[
TestCase(
id="reason_001",
category=TestCategory.REASONING,
difficulty=Difficulty.MEDIUM,
prompt="一个房间有3个开关控制隔壁房间的3盏灯。你只能去隔壁房间一次。如何确定每个开关对应哪盏灯？",
expected_output="打开开关1等5分钟，关掉开关1打开开关2，去隔壁房间。亮的灯对应开关2，摸起来热的对应开关1，凉的对应开关3。",
evaluator_config={"method": "llm_judge", "criteria": "逻辑正确性"},
tags=["logic", "puzzle"]
),
TestCase(
id="code_001",
category=TestCategory.CODING,
difficulty=Difficulty.EASY,
prompt="写一个Python函数，输入列表返回第二大的元素",
expected_pattern=r"def\s+\w+\(.*\).*:.*\n.*return.*\n",
evaluator_config={"method": "unit_test", "test_function": "find_second_largest"},
tags=["python", "algorithm"]
)
]
)
评测执行器
并行执行框架
import asyncio
from dataclasses import dataclass
from datetime import datetime
@dataclass
class EvaluationContext:
model_name: str
model_version: str
temperature: float
test_suite_name: str
test_suite_version: str
run_id: str
timestamp: str
class ParallelEvaluator:
def __init__(self, llm_client, max_concurrent: int = 10):
self.llm = llm_client
self.semaphore = asyncio.Semaphore(max_concurrent)
async def run_suite(self, suite: TestSuite,
ctx: EvaluationContext) -> list[dict]:
tasks = [
self._run_single(case, ctx)
for case in suite.cases
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [r for r in results if not isinstance(r, Exception)]
async def _run_single(self, case: TestCase,
ctx: EvaluationContext) -> dict:
async with self.semaphore:
start = datetime.now()
# 调用LLM
response = await self.llm.chat(
model=ctx.model_name,
messages=[{"role": "user", "content": case.prompt}],
temperature=ctx.temperature
)
elapsed = (datetime.now() - start).total_seconds()
# 评测
score, detail = await self._evaluate(case, response)
return {
"case_id": case.id,
"category": case.category.value,
"difficulty": case.difficulty.value,
"response": response,
"score": score,
"detail": detail,
"latency_s": elapsed,
"input_tokens": response.usage.prompt_tokens,
"output_tokens": response.usage.completion_tokens,
"run_id": ctx.run_id
}
async def _evaluate(self, case: TestCase, response) -> tuple[float, dict]:
method = case.evaluator_config.get("method", "exact")
if method == "exact":
return self._exact_match(response.content, case.expected_output), {"method": "exact"}
elif method == "regex":
return self._regex_match(response.content, case.expected_pattern), {"method": "regex"}
elif method == "llm_judge":
return await self._llm_judge(response.content, case.evaluator_config)
elif method == "unit_test":
return await self._unit_test(response.content, case.evaluator_config)
return 0.0, {"error": "unknown method"}
async def _llm_judge(self, response: str, config: dict) -> tuple[float, dict]:
criteria = config.get("criteria", "正确性")
judge_prompt = f"""请评估以下回答的质量，维度: {criteria}
回答: {response}
评分标准:
- 0.0-0.3: 完全错误
- 0.4-0.6: 部分正确，有缺陷
- 0.7-0.8: 基本正确，小瑕疵
- 0.9-1.0: 完全正确且优秀
只输出一个0到1的浮点数。"""
judge_resp = await self.llm.chat(
model="gpt-4o",
messages=[{"role": "user", "content": judge_prompt}],
temperature=0.0
)
try:
score = float(judge_resp.content.strip())
except ValueError:
score = 0.0
return score, {"method": "llm_judge", "raw": judge_resp.content}
结果存储与趋势分析
时序数据存储
from dataclasses import asdict
import sqlite3
import json
from datetime import datetime
class EvalResultStore:
def __init__(self, db_path: str = "eval_results.db"):
self.conn = sqlite3.connect(db_path)
self._init_db()
def _init_db(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS eval_results (
run_id TEXT,
case_id TEXT,
category TEXT,
difficulty TEXT,
score REAL,
latency_s REAL,
input_tokens INTEGER,
output_tokens INTEGER,
model_name TEXT,
model_version TEXT,
test_suite TEXT,
timestamp TEXT,
detail TEXT
)
""")
self.conn.execute("""
CREATE INDEX IF NOT EXISTS idx_run ON eval_results(run_id)
""")
self.conn.commit()
def save(self, results: list[dict]):
for r in results:
self.conn.execute(
"INSERT INTO eval_results VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
(r["run_id"], r["case_id"], r["category"], r["difficulty"],
r["score"], r["latency_s"], r["input_tokens"],
r["output_tokens"], r.get("model_name",""),
r.get("model_version",""), r.get("test_suite",""),
datetime.now().isoformat(), json.dumps(r["detail"]))
)
self.conn.commit()
def get_trend(self, category: str, days: int = 30) -> list[dict]:
rows = self.conn.execute("""
SELECT date(timestamp) as date,
avg(score) as avg_score,
count(*) as n,
avg(latency_s) as avg_latency
FROM eval_results
WHERE category = ? AND timestamp >= date('now', ?)
GROUP BY date(timestamp)
ORDER BY date
""", (category, f"-{days} days")).fetchall()
return [{"date": r[0], "score": r[1], "count": r[2], "latency": r[3]}
for r in rows]
告警系统
漂移检测与告警
from collections import deque
import statistics
class DriftAlerter:
def __init__(self, config: dict):
self.window = deque(maxlen=config.get("window_size", 50))
self.threshold = config.get("threshold", 0.05)
self.alert_channels = config.get("channels", [])
self.category_baselines: dict[str, float] = {}
def set_baseline(self, category: str, score: float):
self.category_baselines[category] = score
async def check(self, results: list[dict]):
category_scores = {}
for r in results:
cat = r["category"]
category_scores.setdefault(cat, []).append(r["score"])
alerts = []
for cat, scores in category_scores.items():
avg = statistics.mean(scores)
baseline = self.category_baselines.get(cat)
if baseline and (baseline - avg) > self.threshold:
alerts.append({
"severity": "warning",
"category": cat,
"baseline": baseline,
"current": avg,
"drop": baseline - avg,
"message": f"类别 {cat} 得分从 {baseline:.3f} 下降到 {avg:.3f}"
})
# 绝对值告警
if avg < 0.5:
alerts.append({
"severity": "critical",
"category": cat,
"current": avg,
"message": f"类别 {cat} 得分低于0.5: {avg:.3f}"
})
for alert in alerts:
await self._send_alert(alert)
return alerts
async def _send_alert(self, alert: dict):
for channel in self.alert_channels:
if channel["type"] == "webhook":
await self._send_webhook(channel["url"], alert)
elif channel["type"] == "email":
await self._send_email(channel["address"], alert)
调度器：CI/CD集成
Cron + GitHub Actions混合调度
# .github/workflows/llm-eval.yml
name: LLM Evaluation Pipeline
on:
schedule:
- cron: "0 2,14 * * *" # 每天2:00和14:00执行
workflow_dispatch: {} # 手动触发
jobs:
evaluate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install Dependencies
run: pip install -r eval/requirements.txt
- name: Run Evaluation
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
run: |
python -m eval.pipeline \
--suite test-suites/core-abilities-v2.json \
--models config/models.yaml \
--output results/$(date +%Y%m%d_%H%M).json
- name: Check Alerts
run: python -m eval.alerter --results results/*.json
- name: Update Dashboard
if: always()
run: python -m eval.dashboard --update --results results/*.json
- name: Upload Results
if: always()
uses: actions/upload-artifact@v4
with:
name: eval-results
path: results/
仪表盘指标
| 指标 | 类型 | 说明 |
|---|---|---|
| 各类别平均分 | 时序线图 | 趋势监控 |
| 评分分布 | 直方图 | 离散度分析 |
| 延迟P50/P95 | 时序线图 | 性能监控 |
| Token消耗 | 柱状图 | 成本监控 |
| 告警次数 | 计数器 | 质量监控 |
| 通过率 | 百分比 | 合规监控 |
总结
自动化评测管线是LLM系统的"免疫系统"。核心设计要点：
- 测试集管理：版本化、分类别、难度梯度
- 并行执行：异步并发 + 多评测方法（精确/正则/LLM-Judge/单测）
- 时序存储：每次结果入库，支持趋势分析
- 自动告警：漂移检测 + 多渠道通知
- CI集成：定时调度 + 手动触发 + 结果归档
没有评测管线，LLM系统就像没有仪表盘的飞机——飞着飞着可能就出事了。建议在系统上线第一天就搭建这套管线，宁可初期测试集小一些，也不要等到出问题才补建。