七月评测体系建设复盘：从零到可重复评测流水线的搭建之路
一、评测不是跑分，是建立信任
一个模型上线前经过了"充分评测"：MMLU 得分 72.3，C-Eval 得分 68.1，人工评估 10 个 case 全部通过。上线后第三个小时，用户反馈"模型总把'苹果'翻译成'iPhone'"——评测集里根本没有中文多义词的测试用例。
这就是评测体系面临的核心问题：离线评测的分数，与线上表现之间的差距，往往大到不可接受。问题不在于分数本身，在于评测的设计是否覆盖了真实场景。
七月从零开始重新搭建了一套评测体系。目标是三个：可复现（同样的模型和数据集，两次跑出来分数一致）、可比较（不同模型之间可以公平对比）、可追溯（每次评测的参数和结果有完整记录）。
见证奇迹的时刻：当评测流水线第一次跑完并输出标准化的 JSON 报告时，三个月的模型迭代第一次有了可量化的"好坏"标准——不再是"感觉变好了"，而是"MMLU 提升了 1.2% ± 0.3%"。
二、评测体系的三层架构
三层评测体系：基准层确保通用能力、业务层确保上线价值、安全层确保发布底线。基础设施层的版本追踪是整个系统的"地基"——没有版本追踪，所有评测结果都不可信。
见证奇迹的时刻出现在基础设施层跑通的那个下午：任何一个模型的任意一次评测，通过哈希值可以追溯到当时的模型权重、数据集版本、Prompt 版本和评测参数——评测终于从"玄学"变成了"工程"。
三、可重复评测流水线的完整实现
""" 可重复评测流水线的核心组件。 设计原则：任何一次评测的输入（模型+数据+Prompt）和输出（分数+置信区间+时间戳） 都通过哈希绑定，形成不可伪造的评测记录。 """ import json import hashlib import time from pathlib import Path from dataclasses import dataclass, field, asdict from typing import List, Dict, Optional, Any from datetime import datetime import numpy as np # ==================== 1. 版本追踪 ==================== @dataclass class EvalContext: """评测上下文：记录本次评测的所有输入信息。 设计原因：将所有影响评测结果的变量显式记录， 任一变化都会产生新的context_hash，保证可追溯性。""" model_name: str model_hash: str # 模型权重的SHA256 dataset_name: str dataset_version: str dataset_hash: str # 数据集内容的SHA256 prompt_template: str prompt_hash: str # Prompt模板的SHA256 eval_params: Dict[str, Any] # 评估参数（few-shot、温度等） started_at: str = field(default_factory=lambda: datetime.now().isoformat()) @property def context_hash(self) -> str: """生成评测上下文的唯一标识。 设计原因：SHA256确保任何输入的微小变化都会产生不同的hash， 这是"可复现"的技术前提。""" content = json.dumps({ "model_hash": self.model_hash, "dataset_hash": self.dataset_hash, "prompt_hash": self.prompt_hash, "params": self.eval_params, }, sort_keys=True) return hashlib.sha256(content.encode()).hexdigest()[:16] def validate(self) -> bool: """验证上下文完整性。 设计原因：缺少任一hash意味着无法保证复现性，应阻止评测执行。""" required = ["model_hash", "dataset_hash", "prompt_hash"] return all(getattr(self, f) for f in required) # ==================== 2. 评测结果存储 ==================== @dataclass class EvalResult: """单次评测的完整结果。 设计原因：同时存储原始分数和统计信息（均值、标准差、置信区间）， 原始分数用于后续的假设检验，统计信息用于快速对比。""" context_hash: str task_name: str metric_name: str score: float std_error: float # Bootstrap估计的标准误 ci_95_low: float # 95%置信区间下界 ci_95_high: float # 95%置信区间上界 num_samples: int raw_scores: List[float] # 每个样本的原始分数（用于后续分析） finished_at: str = field(default_factory=lambda: datetime.now().isoformat()) def to_dict(self) -> dict: return asdict(self) def is_significantly_better_than(self, other: "EvalResult", alpha: float = 0.05) -> bool: """统计显著性检验。 设计原因：仅比较分数均值没有意义，需要考虑方差。 使用Bootstrap的两样本t检验判断差异是否显著。""" from scipy import stats t_stat, p_value = stats.ttest_ind( self.raw_scores, other.raw_scores, random_state=42, ) return p_value < alpha and self.score > other.score # ==================== 3. 评测调度器 ==================== class EvalScheduler: """评测任务调度器。 设计原因：集中管理评测任务，避免重复执行相同配置的评测。 通过context_hash去重，节省GPU资源和时间。""" def __init__(self, storage_dir: str = "./eval_results"): self.storage_dir = Path(storage_dir) self.storage_dir.mkdir(parents=True, exist_ok=True) self.history: Dict[str, List[EvalResult]] = {} self._load_history() def _load_history(self): """加载历史评测记录。 设计原因：启动时加载已有记录到内存，快速判断新任务是否需要执行。""" for result_file in self.storage_dir.glob("*.jsonl"): with open(result_file, "r") as f: for line in f: result = EvalResult(**json.loads(line)) if result.context_hash not in self.history: self.history[result.context_hash] = [] self.history[result.context_hash].append(result) def should_run(self, context: EvalContext) -> bool: """判断是否需要执行评测。 设计原因：如果相同context已有结果，跳过执行。 'force'参数允许覆盖已有结果。""" if not context.validate(): print("⚠️ 评测上下文不完整，必须运行") return True if context.context_hash in self.history: existing = self.history[context.context_hash] latest = max(r.finished_at for r in existing) print(f"✅ 已有评测结果 (完成于 {latest})，跳过") return False return True def save_result(self, result: EvalResult): """保存评测结果。 设计原因：同时写入JSONL文件和更新内存缓存， JSONL格式支持追加写入，无需全量重写。""" result_path = self.storage_dir / f"{result.context_hash}.jsonl" with open(result_path, "a") as f: f.write(json.dumps(result.to_dict(), ensure_ascii=False) + "\n") if result.context_hash not in self.history: self.history[result.context_hash] = [] self.history[result.context_hash].append(result) def get_trend(self, task_name: str, metric_name: str, limit: int = 20) -> List[dict]: """获取评测趋势数据，用于Dashboard展示。 设计原因：按时间排序展示分数变化趋势， 这是判断模型迭代方向是否正确的最直观方式。""" trend = [] for context_hash, results in self.history.items(): for result in results: if result.task_name == task_name and result.metric_name == metric_name: trend.append({ "date": result.finished_at[:10], "score": result.score, "ci_low": result.ci_95_low, "ci_high": result.ci_95_high, "context_hash": context_hash, }) return sorted(trend, key=lambda x: x["date"])[-limit:] # ==================== 4. 评测指标计算 ==================== def compute_with_confidence( scores: List[float], num_bootstrap: int = 1000, ci_level: float = 0.95, ) -> Dict[str, float]: """使用Bootstrap方法计算置信区间。 设计原因：Bootstrap不依赖正态分布假设， 适用于任何分布的评测分数。1000次重采样是速度和精度的平衡点。""" scores = np.array(scores) n = len(scores) mean = np.mean(scores) # Bootstrap重采样 bootstrap_means = [] rng = np.random.RandomState(42) for _ in range(num_bootstrap): indices = rng.randint(0, n, size=n) bootstrap_means.append(np.mean(scores[indices])) bootstrap_means = np.array(bootstrap_means) std_error = np.std(bootstrap_means, ddof=1) # 百分位数法计算置信区间 alpha = (1 - ci_level) / 2 ci_low = np.percentile(bootstrap_means, alpha * 100) ci_high = np.percentile(bootstrap_means, (1 - alpha) * 100) return { "mean": mean, "std_error": std_error, f"ci_{int(ci_level*100)}_low": ci_low, f"ci_{int(ci_level*100)}_high": ci_high, } # ==================== 5. 评测报告生成 ==================== class EvalReportGenerator: """评测报告生成器。 设计原因：标准化报告格式让不同模型的评测结果可直接对比， Markdown格式便于在代码评审和日报中引用。""" @staticmethod def generate_markdown_report( model_name: str, results: List[EvalResult], baseline_results: Optional[List[EvalResult]] = None, ) -> str: """生成Markdown格式的评测报告""" report = [ f"# 模型评测报告", f"**模型**: {model_name}", f"**评测时间**: {datetime.now().isoformat()}", f"**评测任务数**: {len(results)}", "", "## 评测结果", "", "| 任务 | 指标 | 分数 | 95% CI | 样本数 |", "|------|------|------|--------|--------|", ] for r in results: report.append( f"| {r.task_name} | {r.metric_name} | " f"{r.score:.4f} | [{r.ci_95_low:.4f}, {r.ci_95_high:.4f}] | " f"{r.num_samples} |" ) # 与基线对比 if baseline_results: report.extend([ "", "## 与基线对比", "", "| 任务 | 当前 | 基线 | 变化 | 显著性 |", "|------|------|------|------|--------|", ]) baseline_dict = {(r.task_name, r.metric_name): r for r in baseline_results} for r in results: key = (r.task_name, r.metric_name) if key in baseline_dict: baseline = baseline_dict[key] change = r.score - baseline.score significant = r.is_significantly_better_than(baseline) report.append( f"| {r.task_name} | {r.score:.4f} | {baseline.score:.4f} | " f"{change:+.4f} | {'✅ 显著' if significant else '—'} |" ) return "\n".join(report) # 使用示例 if __name__ == "__main__": scheduler = EvalScheduler() context = EvalContext( model_name="Qwen2.5-7B-v2", model_hash="abc123def456", dataset_name="MMLU", dataset_version="v1.0", dataset_hash="xyz789", prompt_template="standard", prompt_hash="tmpl_hash_001", eval_params={"num_fewshot": 5, "temperature": 0.0}, ) if scheduler.should_run(context): print(f"开始评测: {context.context_hash}") # ... 执行评测逻辑 ... # 生成趋势数据 trend = scheduler.get_trend("MMLU", "accuracy") print(f"MMLU 评测趋势（最近{len(trend)}次）:") for t in trend: print(f" {t['date']}: {t['score']:.4f} [{t['ci_low']:.4f}, {t['ci_high']:.4f}]")四、评测体系建设的核心权衡
覆盖广度 vs 评测深度
理论上评测集越大、维度越多越好。但 GPU 资源有限。一个完整的 MMLU 评测（57 个学科）在 8×A100 上需要 4 小时。基准层的核心评测集（MMLU + C-Eval + GSM8K）是性价比最高的组合。
自动化 vs 人工评测
自动评测覆盖 90% 的场景，但语义质量、创造性、安全性等"软指标"必须人工参与。见证奇迹的时刻：自动化评测显示两版模型得分完全一致（差异 < 0.1%），但人工评测发现新版在口语化表达上明显更优——这是自动评测完全看不到的维度。
离线 vs 在线评测
离线评测可复现、成本低。在线评测（A/B 测试）反映真实用户行为，但成本高、周期长。合理的策略是离线评测做初筛，只有通过离线标准的模型才能进入在线评测阶段。
五、总结
七月评测体系建设实现了三个核心目标：可复现（通过 SHA256 哈希绑定模型、数据集、Prompt 的版本）、可比较（通过 Bootstrap 置信区间和统计显著性检验）、可追溯（每次评测的完整上下文和结果持久化存储）。三层评测架构（基准层-业务层-安全层）覆盖了从通用能力到线上安全的全链路。评测调度器通过 context_hash 去重避免重复执行。Bootstrap 方法提供不依赖分布假设的置信区间。自动评测覆盖硬性指标，语义质量和用户体验等软指标仍需人工参与。评测体系的建设不是一次性工程，而是随着业务场景和模型能力的演进持续迭代的基础设施。
