为什么需要自动化评测管线 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系统的"免疫系统"。核心设计要点：
...
