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_factorydict) 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_idrequest.get(request_id, ), timestamptime.time(), promptrequest.get(prompt, ), modelrequest.get(model, ), parametersrequest.get(parameters, {}), original_outputresponse.get(output, ), original_latency_mslatency_ms, metadatarequest.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(r1[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_fnNone): 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( recordrecord, new_outputnew_output, new_latency_msnew_latency, output_matchexact_match, semantic_similarityround(semantic_sim, 4) if semantic_sim 0 else -1.0, latency_change_pctround(latency_change * 100, 1), is_regressionlen(reasons) 0, regression_reasonsreasons, ) 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_samples5), } # 通过判定 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 测试——用同一份流量回放数据作为基准比较多个候选模型。禁用场景不适合对实时性要求高的流量录制如每个请求都同步写存储——录制操作不能阻塞推理路径。不适合创意性生成模型的回放如 temperature0.9 的对话输出本身就不具有可比较性。五、总结流量回放解决了推理服务回归测试中测试集分布与真实流量不一致的问题。四个对比维度输出一致性、语义相似度≥0.7、延迟变化 50%、格式合规。录制时做分层采样保证低频类目不丢失回放前做 PII 脱敏保证数据安全。回归通过标准劣化率 5%。用生产流量做测试集是模型上线前最后一道防线。