模型服务流量录制回放用真实请求构造回归测试集一、模型升级后谁来保证推理结果不出偏模型服务迭代频率高版本升级、权重微调、框架更新都会影响推理输出。上线后发现问题往往已经造成业务影响——响应格式变化、精度下降、特定输入触发异常。传统单元测试覆盖的是代码逻辑无法验证模型行为。模型本身的黑盒特性决定了只有用真实输入才能发现真实问题。基础设施不需要漂亮话模型上线后的异常率不会因为测试环境跑通了而归零。流量录制回放的核心思路在生产环境录制真实请求存为测试集在预发环境用新模型回放对比输出差异。这不是测试框架的附加功能而是模型服务 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录制策略的关键参数参数说明推荐值录制比例生产流量的采样率避免全量存储10%-20%存储格式JSON 结构化存储便于对比分析请求响应元数据存储周期录制数据的保留时间7-30 天去敏规则脱敏处理避免存储用户隐私数据按业务规则配置回放并发测试环境的请求并发度生产 QPS 的 50%三、流量录制与回放实现录制器基于 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(levellogging.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, encodingutf-8) as f: json.dump(record, f, ensure_asciiFalse, indent2) 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(levellogging.INFO) logger logging.getLogger(traffic-replayer) dataclass class ComparisonResult: 单条对比结果 record_id: str match: bool differences: List[str] field(default_factorylist) old_response: Dict[str, Any] field(default_factorydict) new_response: Dict[str, Any] field(default_factorydict) dataclass class ReplayReport: 回放测试报告 total: int 0 matched: int 0 mismatched: int 0 errors: int 0 details: List[ComparisonResult] field(default_factorylist) 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(totallen(records)) # 并发回放 sem asyncio.Semaphore(self.concurrency) tasks [self._replay_one(rec, sem) for rec in records] results await asyncio.gather(*tasks, return_exceptionsTrue) 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]}, jsonrequest_data[body], timeoutaiohttp.ClientTimeout(total30), ) 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_idrecord[id], matchFalse, differences[f请求失败: {e}], old_responseexpected_response, new_response{}, ) # 对比结果 differences self._compare_responses(expected_response, new_response) return ComparisonResult( record_idrecord[id], matchlen(differences) 0, differencesdifferences, old_responseexpected_response, new_responsenew_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, encodingutf-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_urlhttp://staging-inference:8000, storage_path/data/recordings, concurrency10, ) 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, encodingutf-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_asciiFalse, indent2) # 匹配率低于阈值时输出告警 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。对比策略按模型类型调整文本推理用相似度阈值数值推理用误差范围。多轮对话场景需按会话分组录制和回放。脱敏和存储周期满足安全合规要求。录制回放不是替代单元测试而是补充模型行为的黑盒验证。模型上线前的回放测试通过率是可量化、可追踪的发布质量指标。
模型服务流量录制回放用真实请求构造回归测试集一、模型升级后谁来保证推理结果不出偏模型服务迭代频率高版本升级、权重微调、框架更新都会影响推理输出。上线后发现问题往往已经造成业务影响——响应格式变化、精度下降、特定输入触发异常。传统单元测试覆盖的是代码逻辑无法验证模型行为。模型本身的黑盒特性决定了只有用真实输入才能发现真实问题。基础设施不需要漂亮话模型上线后的异常率不会因为测试环境跑通了而归零。流量录制回放的核心思路在生产环境录制真实请求存为测试集在预发环境用新模型回放对比输出差异。这不是测试框架的附加功能而是模型服务 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录制策略的关键参数参数说明推荐值录制比例生产流量的采样率避免全量存储10%-20%存储格式JSON 结构化存储便于对比分析请求响应元数据存储周期录制数据的保留时间7-30 天去敏规则脱敏处理避免存储用户隐私数据按业务规则配置回放并发测试环境的请求并发度生产 QPS 的 50%三、流量录制与回放实现录制器基于 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(levellogging.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, encodingutf-8) as f: json.dump(record, f, ensure_asciiFalse, indent2) 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(levellogging.INFO) logger logging.getLogger(traffic-replayer) dataclass class ComparisonResult: 单条对比结果 record_id: str match: bool differences: List[str] field(default_factorylist) old_response: Dict[str, Any] field(default_factorydict) new_response: Dict[str, Any] field(default_factorydict) dataclass class ReplayReport: 回放测试报告 total: int 0 matched: int 0 mismatched: int 0 errors: int 0 details: List[ComparisonResult] field(default_factorylist) 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(totallen(records)) # 并发回放 sem asyncio.Semaphore(self.concurrency) tasks [self._replay_one(rec, sem) for rec in records] results await asyncio.gather(*tasks, return_exceptionsTrue) 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]}, jsonrequest_data[body], timeoutaiohttp.ClientTimeout(total30), ) 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_idrecord[id], matchFalse, differences[f请求失败: {e}], old_responseexpected_response, new_response{}, ) # 对比结果 differences self._compare_responses(expected_response, new_response) return ComparisonResult( record_idrecord[id], matchlen(differences) 0, differencesdifferences, old_responseexpected_response, new_responsenew_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, encodingutf-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_urlhttp://staging-inference:8000, storage_path/data/recordings, concurrency10, ) 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, encodingutf-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_asciiFalse, indent2) # 匹配率低于阈值时输出告警 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。对比策略按模型类型调整文本推理用相似度阈值数值推理用误差范围。多轮对话场景需按会话分组录制和回放。脱敏和存储周期满足安全合规要求。录制回放不是替代单元测试而是补充模型行为的黑盒验证。模型上线前的回放测试通过率是可量化、可追踪的发布质量指标。
