模型回归测试：新版本上线前要用历史 Case 跑一遍

模型回归测试：新版本上线前要用历史 Case 跑一遍

一、新模型上线最大的风险：你修了一个 Case，引入了十个 Bug

模型迭代是 AI 项目的常态，但每次模型升级都可能是一次无声的"灾难回滚"。

典型场景是这样的：算法团队调了一个月参数，在公开评测集上跑出了历史最佳的 Rouge-L 和 BLEU 分数，信心满满地提了个上线 PR。上线三小时后，用户反馈开始涌入——"之前能正确回答的数学题现在算错了"、"财务报告的摘要质量大幅下降"、"特定行业术语的翻译全乱了"。运维拉出日志一看，新模型在七个历史通过的测试用例上全部失败。

这不是假设，这是真实发生过的事。根本原因在于：公开评测集只能覆盖模型能力的"平均面"，而生产环境的真实流量分布集中在"长尾面"。新模型在平均面上的提升，往往以长尾面上的退化为代价。

把每个历史 Bad Case 视为一份"技术债"来管理，是一种务实的做法。每一次模型更新，都应该把这些历史欠账单拿出来逐笔对账。基础设施不需要漂亮话，它需要的是一个在任何情况下都不会丢失历史能力的验证流水线。

二、回归测试流水线的三层防御架构

回归测试的核心不是"跑一遍脚本"，而是构建一套可追溯、可审计、可比较的防御体系。

graph TD A[模型训练完成] --> B[第一层：单元级测试] B --> C{通过?} C -->|否| D[阻断发布] C -->|是| E[第二层：Case级回归] E --> F[加载历史Case库] F --> G[并发执行推理] G --> H[比对输出一致性] H --> I{差异度 < 阈值?} I -->|否| J[生成差异报告] I -->|是| K[第三层：A/B 影子测试] K --> L[采样线上流量] L --> M[新旧模型并行推理] M --> N[对比关键指标] N --> O{指标退化 < 5%?} O -->|否| P[人工评审] O -->|是| Q[放行上线] J --> P D --> R[回滚/修复] P --> R

三层防御的设计逻辑：

第一层：单元级。对模型的基础能力做快速冒烟检测，包括格式输出合规性、最大 Token 限制响应、特殊字符处理。这一层在 30 秒内完成，失败的代价最低。

第二层：Case 级。这是回归测试的核心。维护一个结构化 Case 库，每个 Case 包含：输入 Prompt、期望输出（或期望的数值区间）、测试理由（对应历史上的哪个故障）、优先级标签。执行时并发调用推理 API，将新模型输出与历史基线对比。

第三层：A/B 影子测试。在真实流量上并行跑新老模型，对比关键指标但不影响用户结果。这一层最接近生产环境，但也最昂贵。

三层递进的设计遵循一个原则：越靠前的层执行越快、成本越低、阻断越早。模型部署的核心不是"一次测试通过"，而是"在成本最低的环节发现问题"。

三、生产级回归测试框架实现

以下是一个可落地的 Python 回归测试框架骨架，核心设计包括：Case 管理器、并发执行器、差异比较器。

import json import asyncio import hashlib from dataclasses import dataclass, field from typing import Optional, Callable from enum import Enum class CasePriority(Enum): """Case 优先级：P0 阻断上线，P1 需人工确认，P2 仅告警""" P0_BLOCKING = 0 P1_REVIEW = 1 P2_WARNING = 2 @dataclass class TestCase: """回归测试用例的完整定义""" case_id: str priority: CasePriority input_prompt: str expected_output: Optional[str] = None # 精确匹配模式 expected_range: Optional[tuple] = None # 数值区间匹配，如 (0.8, 1.0) reference: str = "" # 历史故障记录链接 metadata: dict = field(default_factory=dict) class RegressionRunner: """回归测试执行器 — 并发调用推理接口，逐 Case 比对""" def __init__(self, infer_func: Callable, concurrency: int = 10): self.infer = infer_func # 推理函数签名: (prompt: str) -> str self.semaphore = asyncio.Semaphore(concurrency) async def run_all(self, cases: list[TestCase]) -> dict: """并发执行所有 Case，返回按优先级分组的差异报告""" tasks = [self._run_one(c) for c in cases] results = await asyncio.gather(*tasks, return_exceptions=True) # 按优先级分组汇总 report = {"P0_BLOCKING": [], "P1_REVIEW": [], "P2_WARNING": []} for case, result in zip(cases, results): if isinstance(result, Exception): report[f"P{case.priority.value}_" + case.priority.name.split("_")[-1]].append({ "case_id": case.case_id, "status": "ERROR", "error": str(result), }) elif result.get("mismatch"): group = f"P{case.priority.value}_" + case.priority.name.split("_")[-1] report[group].append(result) return report async def _run_one(self, case: TestCase) -> dict: """执行单个 Case：推理 + 对比 + 返回差异信息""" async with self.semaphore: try: actual = await asyncio.wait_for( asyncio.to_thread(self.infer, case.input_prompt), timeout=120.0 ) except asyncio.TimeoutError: return {"case_id": case.case_id, "mismatch": True, "reason": "推理超时（120s）"} except Exception as e: return {"case_id": case.case_id, "mismatch": True, "reason": f"推理异常: {e}"} mismatch, reason = self._compare(case, actual) return { "case_id": case.case_id, "mismatch": mismatch, "reason": reason, "actual": actual[:500], # 截断防止报告过大 "expected": case.expected_output[:500] if case.expected_output else str(case.expected_range), } def _compare(self, case: TestCase, actual: str) -> tuple[bool, str]: """核心比对逻辑：支持精确匹配 + 数值区间两种模式""" # 数值区间匹配（适用于 Rouge-L、准确率等可量化指标） if case.expected_range is not None: try: value = float(actual.strip()) low, high = case.expected_range if low <= value <= high: return False, "在期望区间内" return True, f"数值 {value} 不在区间 [{low}, {high}]" except ValueError: return True, f"无法将输出解析为数值: {actual[:100]}" # 精确文本匹配 if case.expected_output is not None: if actual.strip() == case.expected_output.strip(): return False, "精确匹配" # 计算归一化相似度作为辅助信息 sim = self._text_similarity(case.expected_output, actual) return True, f"文本不匹配，相似度: {sim:.2%}" return False, "no assertion" @staticmethod def _text_similarity(a: str, b: str) -> float: """基于字符集 Jaccard 的快速相似度（不替代语义对比）""" set_a, set_b = set(a), set(b) if not set_a and not set_b: return 1.0 return len(set_a & set_b) / len(set_a | set_b) # ---- 使用示例：加载 Case 库并执行回归 ---- def load_cases(path: str) -> list[TestCase]: """从 JSON 文件加载 Case 库""" with open(path, "r") as f: raw = json.load(f) return [ TestCase( case_id=c["id"], priority=CasePriority[c["priority"]], input_prompt=c["prompt"], expected_output=c.get("expected"), expected_range=tuple(c["range"]) if c.get("range") else None, reference=c.get("bug_link", ""), ) for c in raw ] async def main(): def my_infer(prompt: str) -> str: # 替换为实际推理 API 调用 import openai client = openai.OpenAI() resp = client.chat.completions.create( model="my-model-v2", messages=[{"role": "user", "content": prompt}], timeout=60, ) return resp.choices[0].message.content cases = load_cases("regression_cases.json") runner = RegressionRunner(infer_func=my_infer, concurrency=10) report = await runner.run_all(cases) # 阻断逻辑：任何 P0 Case 不通过则返回非 0 if report["P0_BLOCKING"]: print(f"阻断上线：{len(report['P0_BLOCKING'])} 个 P0 Case 失败") for item in report["P0_BLOCKING"]: print(f" - {item['case_id']}: {item['reason']}") exit(1) print(f"P0 通过，P1 需人工审核 {len(report['P1_REVIEW'])} 个 Case")

这个框架的关键设计决策：

1. 并发控制：asyncio.Semaphore限制同时发起的推理请求数，防止打爆推理服务。生产环境通常设置 10-20 并发。
2. 超时保护：每个 Case 120 秒超时，单个 Case 卡死不影响整体流水线。
3. 优先级阻断：P0（阻断级）Case 失败直接 exit(1)，P1 生成报告供人工评审。这是务实的权衡——不是所有回归都能自动化判定。
4. Case 编号归一化：用case_id做唯一标识，方便在 CI 日志中 grep 定位失败点。

四、回归测试的成本边界与不能做的事情

回归测试很容易从一个好习惯演变成"运维噩梦"。

Case 膨胀问题。一个运行了两年的模型服务，Case 库可能积累 2000+ 条用例。全量跑一遍的推理成本可能高达数百美元，耗时超过 2 小时。解决思路是分层抽样：P0 Case 全跑，P1 Case 随机抽 30%，P2 Case 仅在新模型在 P0/P1 层出现异常时才激活。同时建立 Case 退役机制——连续 10 次通过的 P2 Case 自动降级归档。

语义等价的误判。文本匹配不是语义匹配。"将文件复制到目标路径"和"把文件拷贝到目标目录"在语义上等价，但字符串比对会报告差异。这时候需要引入语义相似度模型（如 Sentence-BERT）做二次判定，但这又引入了一个"谁测试测试者"的递归问题。务实的做法是：P0 Case 使用精确文本匹配确保严格一致性，P1/P2 Case 使用嵌入相似度阈值（如 cosine_sim > 0.95 视为通过）。

新能力的假阳性。如果新模型对某个 Prompt 的回答确实比旧模型更优（例如修复了旧模型的幻觉），回归测试会错误地标记为退化。这需要人工评审机制介入——框架生成差异报告，由算法工程师逐条确认是否是"有意义的偏离"。不能为了自动化而自动化。

Case 库维护成本。Case 库不是一次性建设的，它需要持续维护：新增线上用户反馈的 Bad Case、淘汰已修复且长期稳定的 Case、更新期望值以反映模型行为的合理演进。建议指定一位轮值工程师每周做 Case 库清理，否则三个月后 Case 库就会变成无人敢动的"历史黑洞"。

五、总结

回归测试的本质是"用历史错误防御未来退化"。它不追求完美覆盖，而是追求在最经济的成本下保护最关键的能力。

落地建议分三步走：第一步，从线上最近 30 天的用户反馈中提取 20 个高频 Bad Case 作为 P0 种子集；第二步，实现本文的 RegressionRunner 并接入 CI 流水线，每次模型 PR 自动触发回归；第三步，建立 Case 库的增删改流程和周度评审机制。

三层防御体系（单元 → Case → A/B 影子测试）的设计原则是"越早失败越便宜"。P0 Case 不通过，连代码合并都不应该发生。基础设施的可靠性不是靠一次测试保证的，而是靠每一次变更都被充分验证的工程纪律堆出来的。