录制模型响应,让 CI 离线回放
文章介绍一个单文件、仅依赖标准库的模型请求录制与回放代理,用请求哈希索引已保存的响应。它将提示词组装测试与实时模型服务解耦,减少 CI 对网络、密钥和调用预算的依赖。
AI 深度解读
把模型接进测试后,CI 很容易多出三项依赖:供应商网络、API key 和调用预算。作者做了一个很小的 cassette proxy:录下一次真实请求的响应,后续测试按请求内容从磁盘回放,缺失记录就报错。
这个方案适合检查提示词和请求参数有没有被改坏,却不能证明模型回答质量没有退步。它值得读的地方,是把测试边界划得很清楚;但原文示例的模式切换和路径拼接也有需要修正的地方,不能直接照抄进 CI。
先分清:你要测请求组装,还是模型能力
提示词模块通常从纯函数开始:拼接 messages、填充 system prompt、按 token 预算裁剪历史。后来加上一个调用真实接口的端到端测试,整套测试就开始受外部服务影响。
供应商故障会让 CI 失败;fork PR 拿不到密钥,测试跑不起来;为了让流水线恢复,有人可能把关键测试跳过。作者认为,这类问题首先要从调用边界解决。
cassette 可以理解为「按请求归档的响应快照」。代理先计算请求的键,再决定访问上游还是读取文件:
record调用上游,保存响应,再返回结果。replay只读磁盘,命中返回 HTTP 200,缺失返回 HTTP 409,并附上cassette_miss和缓存键。off直接转发,用于检查真实模型行为。
回放缺失时必须失败,不能偷偷访问网络。
否则,新增请求仍会把网络波动和费用带回 CI。作者报告,第一周的 409 错误发现了两次意外的 payload 修改;这是一个有用的案例,但原文没有给出完整测试记录或对照数据。
工具只有一个 Python 文件,仅使用标准库,要求 Python 3.10+,面向 OpenAI-compatible chat endpoint。默认模式是 replay,监听 127.0.0.1:8089,数据放在 cassettes/。相关配置通过 CASSETTE_MODE、UPSTREAM_BASE、UPSTREAM_API_KEY、CASSETTE_DIR 和 PORT 传入。
请求内容才是契约,缓存键不能随便省字段
如果按测试名称保存响应,测试重命名会影响命中,而请求内容改了却可能继续拿到旧结果。作者选择对请求体规范化后计算哈希。
下面这段代码负责消除 JSON 键顺序的影响,同时排除指定字段:
VOLATILE = {"stream", "user", "metadata", "request_id"} # never part of the key
def canonical(payload: dict) -> str:
body = {k: v for k, v in payload.items() if k not in VOLATILE}
return json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
def key_for(payload: dict) -> str:
return hashlib.sha256(canonical(payload).encode("utf-8")).hexdigest()[:32]
要看清第一行和最后一行:排除哪些字段决定了测试能否发现变化;SHA-256 的十六进制结果只取前 32 个字符,作为文件名中的键。
system prompt 多一行、用户消息发生变化、采样参数不同,都应该产生新键。temperature=0 和 temperature=0.7 是不同请求,不能为了提高命中率把它们归为一类。
但 VOLATILE 也不是通用答案。代码只移除顶层字段,而且「不影响语义」取决于业务。如果 metadata 参与路由,或者测试关心 user 的传递,排除它们就会漏掉变化。stream 更直接决定响应形态;这个实现能够忽略它,前提是流式路径已经被排除在测试范围之外。
另一个边界是:缓存键没有包含 URL 路径和上游地址。同一个请求体发往不同路径,或者切换供应商后继续使用同一个目录,都可能命中旧记录。作者的方案适合受控的单端点测试,扩展到多个端点时,需要重新定义请求身份。
录制和回放能分开,示例配置却还没完全分开
作者建议先录制,再提交 cassette,之后让 CI 使用 dummy key 回放。示例测试保留了真实客户端和业务函数的调用路径:
def test_summary_payload(cassette_server):
client = make_client(base_url=cassette_server, api_key="cassette")
out = summarize_commits(client, ["fix: guard null", "feat: retry cap"])
assert out.startswith("Two commits")
这里要注意 base_url=cassette_server:客户端必须指向代理,才能把外部依赖替换为快照。api_key="cassette" 只是占位值。断言检查的是固定响应经过业务代码后能否得到预期结果。
原文用 session 级、自动启用的 pytest fixture 启动代理,通过 /healthz 等待就绪,再返回 http://127.0.0.1:8089/v1。它最多尝试 100 次,单次请求超时 0.5 秒,失败后等待 0.1 秒;结束时终止进程,等待超时为 10 秒。
不过,这套示例有几处必须先修:
第一,fixture 把子进程的 CASSETTE_MODE 写死为 replay。外面执行 CASSETTE_MODE=record pytest tests/test_end_to_end.py -q,并不能让它切换到录制。原文的 for m in replay off 计时循环,同样无法切换这个 fixture 启动的代理模式。
第二,录制命令先在后台启动代理,fixture 又使用同一个 8089 端口启动进程,会造成端口冲突,也让健康检查可能连到另一个进程。代理由谁启动,需要统一。
第三,默认 UPSTREAM_BASE 已包含 /v1,转发却直接拼接客户端路径。如果收到 /v1/chat/completions,就会得到重复的 /v1/v1/chat/completions。配置和路径规则必须一起调整。
上游请求超时为 120 秒,遇到 HTTP 429 时代码抛出 SystemExit。但请求运行在 ThreadingHTTPServer 的处理线程里,不能据此认定整个录制服务会退出;它也没有向客户端返回结构化的限流错误。
这些问题不否定回放方案,却说明「文件很短」和「接入已经可靠」是两回事。
我的判断
我认可它把稳定性测试与模型质量评估分开。检查请求组装时,每次支付模型调用成本,收益确实有限。与手写 mock 相比,真实响应录制能减少人工构造响应的工作;与实时端到端测试相比,回放更容易在无密钥环境中运行。
但快照只证明当前代码还能处理过去保存的响应。模型退役、供应商行为变化、回答质量下降,回放都看不见。原文也明确排除了质量评估、逐 token 流式聊天和一次性脚本。
作者没有发布性能基准,只建议测三项:测试总耗时、干净检出后的 cassette miss 数量,以及 CI 对模型上游的出站请求数。干净检出后出现 miss 应当排查,不能当成预热。至于这个工具比常见的客户端 mock、HTTP 录制库更划算,原文没有对照数据,仍属于经验判断。
数据管理也不能省略。每个文件保存 key、完整 request 和 response。UPSTREAM_API_KEY 从环境变量读取,不写入 cassette,但请求体里的个人信息、业务数据仍会落盘。提交前需要检查内容,提示词变化后的批量重录也应单独审查,避免把行为变化淹没在快照更新里。
原文披露,这是 MonkeyCode 产品推广的一部分。文中提到的 10M-token allowance 是撰写时的促销额度,不是永久承诺;免费服务器是否保持进程运行、文件系统是否持久化,也需要另行确认。把 cassette 打进构建产物可以减少持久化问题,但远程部署仍会引入网络依赖。对希望离线运行的 CI,本地启动代理更符合最初目标。
能拿来做什么
- 对维护提示词拼接模块的 Node 或全栈项目,可以先选一个非流式接口接入回放,覆盖 messages、system prompt 和采样参数变化。客户端连接代理,具体测试框架不必限定为 pytest。
- 把回放测试和实时模型检查分成独立入口。前者随 PR 运行,检查请求契约与响应处理;后者用于确认当前模型仍可调用、输出仍满足要求。
- 接入前先修正模式传递、端口管理和
/v1拼接规则,再验证修改一行 prompt 会返回 409。若无法稳定触发 miss,缓存键就没有守住测试边界。 - 将重录结果作为需要审查的数据变更,提交前检查敏感内容,并记录耗时、miss 数量和模型上游请求数,用本项目的数据判断维护成本是否值得。