企业官网建设流程全解析

模型服务流量录制回放：用真实请求构造回归测试集

一、模型升级后谁来保证推理结果不出偏

模型服务迭代频率高，版本升级、权重微调、框架更新都会影响推理输出。上线后发现问题往往已经造成业务影响——响应格式变化、精度下降、特定输入触发异常。传统单元测试覆盖的是代码逻辑，无法验证模型行为。模型本身的"黑盒"特性决定了，只有用真实输入才能发现真实问题。

基础设施不需要漂亮话，模型上线后的异常率不会因为"测试环境跑通了"而归零。流量录制回放的核心思路：在生产环境录制真实请求，存为测试集，在预发环境用新模型回放，对比输出差异。这不是测试框架的附加功能，而是模型服务 CI 的必要环节。

流量录制的实现方式有两种：网关层拦截和 SDK 嵌入。网关层拦截不侵入业务代码，但无法录制内部状态；SDK 嵌入可以录制更多上下文，但增加维护成本。推理场景中，录制输入输出足够，不需要内部状态，网关层拦截更合适。

二、流量录制回放的系统架构

录制回放系统分三个模块：录制器（Recorder）、存储器（Store）、回放器（Replayer）。录制器在网关层拦截请求并存储，存储器持久化请求-响应对，回放器在测试环境重放请求并对比结果。

flowchart TB subgraph Prod["生产环境"] GW["推理网关"] REC["录制器: 拦截请求"] INF_P["推理服务 Pod"] GW --> REC --> INF_P end subgraph Store["存储层"] S1["请求元数据: prompt, 参数"] S2["响应数据: 推理结果, 状态码"] S3["上下文: 模型版本, 节点信息"] REC --> S1 REC --> S2 INF_P --> S3 end subgraph Staging["预发环境"] RP["回放器: 读取测试集"] INF_S["新版本推理服务"] CMP["结果对比器"] RP --> INF_S --> CMP Store --> RP end style Prod fill:#e8f5e9 style Store fill:#e3f2fd style Staging fill:#fff3e0

录制策略的关键参数：

三、流量录制与回放实现

录制器基于 FastAPI 中间件实现，拦截请求和响应：

# recorder.py — 推理网关流量录制中间件 import json import logging import hashlib import random from datetime import datetime from fastapi import Request, Response from starlette.middleware.base import BaseHTTPMiddleware logging.basicConfig(level=logging.INFO) logger = logging.getLogger("traffic-recorder") class TrafficRecorderMiddleware(BaseHTTPMiddleware): """推理流量录制中间件，采样存储请求-响应对""" def __init__(self, app, sample_rate: float = 0.1, storage_path: str = "/data/recordings"): super().__init__(app) self.sample_rate = sample_rate # 采样率 self.storage_path = storage_path # 存储路径 async def dispatch(self, request: Request, call_next): # 判断是否采样本次请求 should_record = random.random() < self.sample_rate if not should_record: # 不录制，直接转发 return await call_next(request) # 录制请求部分 request_data = await self._capture_request(request) # 执行推理请求 response = await call_next(request) # 录制响应部分 response_data = await self._capture_response(response) # 组合录制记录并存储 record = { "id": hashlib.sha256( f"{request_data['timestamp']}{request_data['path']}".encode() ).hexdigest()[:16], "request": request_data, "response": response_data, "meta": { "recorded_at": datetime.now().isoformat(), "sample_rate": self.sample_rate, }, } # 异步存储，不影响请求延迟 await self._store_record(record) return response async def _capture_request(self, request: Request) -> dict: """提取请求关键信息""" body = await request.body() body_json = {} try: body_json = json.loads(body) except json.JSONDecodeError: body_json = {"raw_body_size": len(body)} # 脱敏处理：移除可能包含用户隐私的字段 sanitized = self._sanitize(body_json) return { "method": request.method, "path": request.url.path, "headers": dict(request.headers), # 仅保留非敏感 header "body": sanitized, "query_params": dict(request.query_params), "timestamp": datetime.now().isoformat(), } async def _capture_response(self, response: Response) -> dict: """提取响应关键信息""" # 读取响应体（需要特殊处理，因为 Response body 不可重复读取） response_body = b"" if hasattr(response, "body_iterator"): chunks = [] for chunk in response.body_iterator: if isinstance(chunk, str): chunks.append(chunk.encode()) else: chunks.append(chunk) response_body = b"".join(chunks) # 重新设置 body_iterator，让后续处理可以读取 response.body_iterator = iter([response_body]) response_json = {} try: response_json = json.loads(response_body) except json.JSONDecodeError: response_json = {"raw_body_size": len(response_body)} return { "status_code": response.status_code, "headers": dict(response.headers), "body": response_json, } def _sanitize(self, data: dict) -> dict: """脱敏处理，移除隐私字段""" sensitive_fields = {"user_id", "phone", "email", "session_token"} if isinstance(data, dict): return { k: "***" if k in sensitive_fields else ( self._sanitize(v) if isinstance(v, dict) else v ) for k, v in data.items() } return data async def _store_record(self, record: dict): """存储录制记录到文件系统（生产环境替换为对象存储）""" record_id = record["id"] file_path = f"{self.storage_path}/{record_id}.json" try: with open(file_path, "w", encoding="utf-8") as f: json.dump(record, f, ensure_ascii=False, indent=2) logger.info(f"录制记录已存储: {record_id}") except IOError as e: logger.error(f"录制存储失败: {e}")

回放器与结果对比器实现：

# replayer.py — 流量回放与结果对比 import json import asyncio import logging from datetime import datetime from typing import List, Dict, Any from dataclasses import dataclass, field logging.basicConfig(level=logging.INFO) logger = logging.getLogger("traffic-replayer") @dataclass class ComparisonResult: """单条对比结果""" record_id: str match: bool differences: List[str] = field(default_factory=list) old_response: Dict[str, Any] = field(default_factory=dict) new_response: Dict[str, Any] = field(default_factory=dict) @dataclass class ReplayReport: """回放测试报告""" total: int = 0 matched: int = 0 mismatched: int = 0 errors: int = 0 details: List[ComparisonResult] = field(default_factory=list) @property def match_rate(self) -> float: if self.total == 0: return 0.0 return self.matched / self.total class TrafficReplayer: """流量回放器：读取录制数据，发送到新模型服务，对比结果""" def __init__(self, target_url: str, storage_path: str, concurrency: int = 10): self.target_url = target_url self.storage_path = storage_path self.concurrency = concurrency async def replay_all(self) -> ReplayReport: """回放所有录制数据并生成报告""" records = self._load_records() report = ReplayReport(total=len(records)) # 并发回放 sem = asyncio.Semaphore(self.concurrency) tasks = [self._replay_one(rec, sem) for rec in records] results = await asyncio.gather(*tasks, return_exceptions=True) for result in results: if isinstance(result, Exception): report.errors += 1 logger.error(f"回放异常: {result}") elif result.match: report.matched += 1 report.details.append(result) else: report.mismatched += 1 report.details.append(result) # 输出报告 logger.info(f"回放完成: 总数={report.total}, " f"匹配={report.matched}, " f"不匹配={report.mismatched}, " f"异常={report.errors}, " f"匹配率={report.match_rate:.2%}") return report async def _replay_one(self, record: dict, sem: asyncio.Semaphore) -> ComparisonResult: """回放单条记录""" async with sem: request_data = record["request"] expected_response = record["response"] # 构造回放请求 import aiohttp try: async with aiohttp.ClientSession() as session: async with session.post( f"{self.target_url}{request_data['path']}", json=request_data["body"], timeout=aiohttp.ClientTimeout(total=30), ) as resp: new_response_body = await resp.json() new_response = { "status_code": resp.status, "body": new_response_body, } except Exception as e: return ComparisonResult( record_id=record["id"], match=False, differences=[f"请求失败: {e}"], old_response=expected_response, new_response={}, ) # 对比结果 differences = self._compare_responses(expected_response, new_response) return ComparisonResult( record_id=record["id"], match=len(differences) == 0, differences=differences, old_response=expected_response, new_response=new_response, ) def _compare_responses(self, old: dict, new: dict) -> List[str]: """对比两次推理结果的差异""" diffs = [] # 状态码对比 if old.get("status_code") != new.get("status_code"): diffs.append(f"状态码变化: {old.get('status_code')} -> {new.get('status_code')}") old_body = old.get("body", {}) new_body = new.get("body", {}) # 响应格式对比（结构是否一致） old_keys = set(old_body.keys()) new_keys = set(new_body.keys()) if old_keys != new_keys: diffs.append(f"响应字段变化: 缺失={new_keys - old_keys}, 新增={old_keys - new_keys}") # 推理结果内容对比（允许文本差异，检查关键指标） if "result" in old_body and "result" in new_body: old_result = old_body["result"] new_result = new_body["result"] if old_result != new_result: # 文本推理结果不要求完全一致，但差异过大需要关注 # 使用简单的相似度检查 similarity = self._text_similarity(old_result, new_result) if similarity < 0.7: diffs.append(f"推理结果差异较大: 相似度={similarity:.2%}") if "error" in new_body and "error" not in old_body: diffs.append(f"新增错误: {new_body['error']}") return diffs def _text_similarity(self, text1: str, text2: str) -> float: """简易文本相似度计算（生产环境可替换为专业算法）""" if not text1 or not text2: return 0.0 # 基于字符级 Jaccard 相似度 set1 = set(text1) set2 = set(text2) intersection = len(set1 & set2) union = len(set1 | set2) return intersection / union if union > 0 else 0.0 def _load_records(self) -> List[dict]: """从存储路径加载所有录制记录""" import os records = [] if not os.path.exists(self.storage_path): logger.error(f"存储路径不存在: {self.storage_path}") return records for filename in os.listdir(self.storage_path): if filename.endswith(".json"): filepath = os.path.join(self.storage_path, filename) try: with open(filepath, "r", encoding="utf-8") as f: record = json.load(f) records.append(record) except (json.JSONDecodeError, IOError) as e: logger.error(f"加载记录失败 {filename}: {e}") logger.info(f"加载录制记录 {len(records)} 条") return records # 回放入口 async def main(): replayer = TrafficReplayer( target_url="http://staging-inference:8000", storage_path="/data/recordings", concurrency=10, ) report = await replayer.replay_all() # 生成回放报告 report_path = f"/data/replay-report-{datetime.now().strftime('%Y%m%d%H%M')}.json" with open(report_path, "w", encoding="utf-8") as f: json.dump({ "total": report.total, "matched": report.matched, "mismatched": report.mismatched, "errors": report.errors, "match_rate": f"{report.match_rate:.2%}", "mismatch_details": [ {"id": d.record_id, "differences": d.differences} for d in report.details if not d.match ], }, f, ensure_ascii=False, indent=2) # 匹配率低于阈值时输出告警 if report.match_rate < 0.95: logger.warning(f"匹配率低于 95%，需要人工审查: {report.match_rate:.2%}") if __name__ == "__main__": asyncio.run(main())

四、录制回放的边界与限制

场景一：时序依赖的请求。部分推理请求依赖上下文状态（如多轮对话），单条录制无法重现依赖关系。解法：按会话 ID 分组录制，回放时按序重放整个会话，而非独立请求。存储代价增加，但覆盖了状态依赖场景。

场景二：推理结果的模糊对比。模型输出不是精确值，文本推理的"正确性"缺乏严格定义。完全一致的对比要求不现实，需要定义相似度阈值。对话类模型关注语义一致性，数值类模型关注误差范围。对比策略按模型类型定制，统一阈值会放过关键问题或误报过多。

场景三：录制数据的安全合规。生产流量可能包含用户隐私数据。录制前必须脱敏，脱敏规则按业务字段配置。存储周期不超过合规要求的天数，过期自动清理。对象存储设置访问权限，只有回放系统可以读取。

场景四：录制影响生产性能。中间件拦截增加请求延迟，全量录制时影响更大。采样率控制在 10%-20%，延迟增加不超过 1ms。存储操作异步执行，不阻塞请求返回路径。

五、总结

流量录制回放为模型服务提供回归验证手段：在生产环境采样真实请求，存为结构化数据，预发环境回放对比结果差异。实现分为录制中间件、存储层、回放器三个模块，采样率 10%-20%，存储格式统一为 JSON。对比策略按模型类型调整：文本推理用相似度阈值，数值推理用误差范围。多轮对话场景需按会话分组录制和回放。脱敏和存储周期满足安全合规要求。录制回放不是替代单元测试，而是补充模型行为的黑盒验证。模型上线前的回放测试通过率，是可量化、可追踪的发布质量指标。