企业官网建设流程全解析

AI 服务流量回放：用历史请求构建推理服务回归测试集

一、你的推理服务升级了新模型，测试集准确率涨了，线上用户投诉翻了倍

推理服务的回归测试有一个独特的难题：测试集的分布和真实流量的分布经常不一致。你的测试集覆盖了 100 类问题，每类 50 个样本——分布均匀。但真实流量中，80% 的问题集中在 3 个类别上，其他 97 个类别加一起只占 20%。如果模型在这 3 个高频类别上表现退化，你的均匀测试集根本检测不到。

流量回放（Traffic Replay）解决了这个问题：把生产环境的历史请求记录下来，在模型升级前用新模型重新推理一遍历史请求，对比输出结果。这不只是"检查正确率是否下降"——还要检查输出分布是否漂移、响应延迟是否增加、格式是否变化。

核心挑战是：推理请求的输入往往包含了用户问题 + 对话历史 + 检索上下文，数据量很大（单个请求可能 10KB-100KB）。如果按天全量录制，存储成本是一个大问题。需要采样 + 压缩 + 自动清洗（去掉 PII 个人身份信息）。

二、底层机制与原理剖析

流量回放的完整流程：

三、生产级代码实现

""" 推理服务流量回放系统 录制 → 清洗 → 回放 → 对比 """ from dataclasses import dataclass, field from typing import List, Dict, Optional, Any import hashlib import json import time import re from collections import Counter import numpy as np @dataclass class InferenceRecord: """单条推理记录""" request_id: str timestamp: float prompt: str model: str parameters: Dict[str, Any] # temperature, max_tokens 等 # 原始输出 original_output: str original_latency_ms: float # 元数据 metadata: Dict[str, Any] = field(default_factory=dict) @dataclass class ReplayResult: """回放对比结果""" record: InferenceRecord # 新模型输出 new_output: str new_latency_ms: float # 对比 output_match: bool # 输出完全一致 semantic_similarity: float # 语义相似度 latency_change_pct: float # 延迟变化百分比 # 是否有劣化标记 is_regression: bool regression_reasons: List[str] class TrafficRecorder: """流量录制器 在推理服务中拦截请求和响应，异步写入存储。 """ def __init__(self, storage, sample_rate: float = 0.1): self.storage = storage self.sample_rate = sample_rate # 类目计数器（用于分层采样） self._category_counter = Counter() def should_record(self, request: Dict) -> bool: """判定是否应该录制此请求""" # 基础采样 if np.random.random() > self.sample_rate: return False # 对于低频类目，提高采样率 category = request.get("metadata", {}).get("category", "unknown") count = self._category_counter.get(category, 0) if count < 100: return np.random.random() < 0.5 # 低频类目 50% 采样 self._category_counter[category] += 1 return True def record(self, request: Dict, response: Dict, latency_ms: float): """录制一条推理记录""" if not self.should_record(request): return record = InferenceRecord( request_id=request.get("request_id", ""), timestamp=time.time(), prompt=request.get("prompt", ""), model=request.get("model", ""), parameters=request.get("parameters", {}), original_output=response.get("output", ""), original_latency_ms=latency_ms, metadata=request.get("metadata", {}), ) # 异步写入存储 self.storage.save(record) class PIISanitizer: """PII 脱敏处理 在回放前清洗请求中的个人身份信息。 关键：不能修改语义内容，只替换标识符。 """ # Email 正则 EMAIL_PATTERN = re.compile(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}') # 手机号正则（中国大陆） PHONE_PATTERN = re.compile(r'1[3-9]\d{9}') # 身份证号 ID_PATTERN = re.compile(r'\d{17}[\dXx]') def sanitize(self, text: str) -> str: """脱敏文本""" text = self.EMAIL_PATTERN.sub('[EMAIL]', text) text = self.PHONE_PATTERN.sub('[PHONE]', text) text = self.ID_PATTERN.sub('[ID_NUMBER]', text) return text class RegressionComparator: """回归对比器 对比新旧模型在同一输入下的输出差异。 四个维度：输出一致性、语义相似度、延迟变化、格式合规。 """ def __init__(self, semantic_sim_fn=None): self.semantic_sim_fn = semantic_sim_fn # 语义相似度计算函数 def compare(self, record: InferenceRecord, new_output: str, new_latency: float) -> ReplayResult: """对比单条记录""" reasons = [] # 维度 1：输出一致性 exact_match = (record.original_output.strip() == new_output.strip()) # 维度 2：语义相似度 semantic_sim = 1.0 if self.semantic_sim_fn and not exact_match: try: semantic_sim = self.semantic_sim_fn( record.original_output, new_output ) except Exception: semantic_sim = -1.0 # 计算失败 # 维度 3：延迟变化 latency_change = 0.0 if record.original_latency_ms > 0: latency_change = (new_latency - record.original_latency_ms) / record.original_latency_ms # 判定是否劣化 if semantic_sim >= 0 and semantic_sim < 0.7: reasons.append(f"语义相似度过低: {semantic_sim:.2f}") if latency_change > 0.5: # 延迟增加超过 50% reasons.append(f"延迟显著增加: +{latency_change*100:.0f}%") # 格式检查：新输出是否包含异常标记（如截断） if new_output.endswith('...') and not record.original_output.endswith('...'): reasons.append("输出可能被截断") return ReplayResult( record=record, new_output=new_output, new_latency_ms=new_latency, output_match=exact_match, semantic_similarity=round(semantic_sim, 4) if semantic_sim >= 0 else -1.0, latency_change_pct=round(latency_change * 100, 1), is_regression=len(reasons) > 0, regression_reasons=reasons, ) class RegressionReport: """回归测试报告""" def __init__(self, low_sim_threshold: float = 0.7, latency_increase_threshold: float = 0.5): self.low_sim_threshold = low_sim_threshold self.latency_increase_threshold = latency_increase_threshold def generate(self, results: List[ReplayResult]) -> Dict: """生成回归报告""" total = len(results) regressions = [r for r in results if r.is_regression] # 统计 low_semantic = [r for r in results if 0 <= r.semantic_similarity < self.low_sim_threshold] high_latency = [r for r in results if r.latency_change_pct > self.latency_increase_threshold * 100] report = { "summary": { "total_samples": total, "regression_count": len(regressions), "regression_rate": round(len(regressions) / max(total, 1), 4), "exact_match_rate": round( sum(1 for r in results if r.output_match) / max(total, 1), 4 ), "avg_semantic_similarity": round( np.mean([r.semantic_similarity for r in results if r.semantic_similarity >= 0]), 4 ) if results else 0, "avg_latency_change_pct": round( np.mean([r.latency_change_pct for r in results]), 1 ), }, "issues": { "low_semantic_similarity": { "count": len(low_semantic), "threshold": self.low_sim_threshold, }, "high_latency_increase": { "count": len(high_latency), "threshold_pct": self.latency_increase_threshold * 100, }, }, # 分维度统计 "by_dimension": self._by_dimension(results, total), # 输出差异样例（最多 5 条） "sample_diffs": self._sample_diffs(regressions, max_samples=5), } # 通过判定 regression_rate = report["summary"]["regression_rate"] report["passed"] = regression_rate < 0.05 # 劣化率 < 5% 认为通过 return report def _by_dimension(self, results: List[ReplayResult], total: int) -> Dict: """按维度统计""" dims = {} for r in results: for reason in r.regression_reasons: # 提取原因的关键词 if "语义" in reason: dims["semantic"] = dims.get("semantic", 0) + 1 elif "延迟" in reason: dims["latency"] = dims.get("latency", 0) + 1 elif "截断" in reason: dims["truncation"] = dims.get("truncation", 0) + 1 return {k: {"count": v, "rate": round(v / max(total, 1), 4)} for k, v in dims.items()} def _sample_diffs(self, regressions: List[ReplayResult], max_samples: int = 5) -> List[Dict]: """采样输出差异样例""" samples = regressions[:max_samples] return [ { "prompt": r.record.prompt[:200], "original": r.record.original_output[:300], "new": r.new_output[:300], "reasons": r.regression_reasons, } for r in samples ]

四、边界分析与架构权衡

流量录制的存储成本：

全量录制实时推理流量（如 1000 QPS），每条请求 10KB，一天的存储量是 864GB。即使采样 10%，也是 86GB/天。需要考虑压缩存储（gzip 后通常能压缩到原来的 10-20%）和自动过期清理。

比较的难度：

推理输出不是简单的"对/错"。两个输出可能语义相同但措辞完全不同。语义相似度计算本身就需要调用 embedding 模型——但这又引入了另一个模型的不确定性。建议用固定版本的 embedding 模型（如 text-embedding-3-large）做语义比较，确保比较本身的一致性。

适用边界：

最适合模型频繁升级（周度或双周迭代）的推理服务。也适合多模型 A/B 测试——用同一份流量回放数据作为基准比较多个候选模型。

禁用场景：

不适合对实时性要求高的流量录制（如每个请求都同步写存储）——录制操作不能阻塞推理路径。不适合创意性生成模型的回放（如 temperature=0.9 的对话），输出本身就不具有可比较性。

五、总结

流量回放解决了推理服务回归测试中"测试集分布与真实流量不一致"的问题。四个对比维度：输出一致性、语义相似度（≥0.7）、延迟变化（< +50%）、格式合规。录制时做分层采样保证低频类目不丢失，回放前做 PII 脱敏保证数据安全。回归通过标准：劣化率 < 5%。用生产流量做测试集，是模型上线前最后一道防线。