OGX 测试录制系统深度解析：基于 API 录制回放的确定性集成测试方案

OGX 测试录制系统深度解析：基于 API 录制回放的确定性集成测试方案

【免费下载链接】ogxOpen GenAI Stack项目地址: https://gitcode.com/GitHub_Trending/ll/ogx

本文以 OGX（Open GenAI Stack）仓库中的测试录制系统为核心，系统讲解其目录结构、JSON 录音格式、字段规范化规则、四种运行模式以及背后的源码实现原理。读完本文，你将掌握如何使用OGX_TEST_INFERENCE_MODE环境变量在 replay、record-if-missing、record、live 四种模式间切换，理解请求哈希匹配与测试隔离机制，并能独立完成录音的新增、回放与批量重规范化操作。

一、为什么需要录制回放

OGX 的集成测试位于 tests/integration/ 目录，覆盖 inference、agents、responses、vector_io、tool_runtime 等众多 API 面。这些测试天然依赖 OpenAI、Ollama、Google Vertex AI、Tavily 等外部服务——如果每次跑测试都实时调用真实 API，会带来三个问题：

• 成本与稳定性：真实 API 调用产生费用，且受网络抖动、服务限流影响，测试结果不稳定；
• 确定性缺失：LLM 生成内容带有随机性，同样的请求两次运行可能返回不同文本，导致断言不可靠；
• 开发体验差：开发者本地没有 API Key 或无法访问外部服务时，测试无法运行。

录制回放（record/replay）系统的目标正是解决这些问题：预先将一次真实的 API 调用请求与响应配对，以 JSON 文件落盘；测试运行时不再访问真实服务，而是按请求哈希查找并重放录音，从而在完全离线、确定性、零成本的前提下验证全部行为。

该系统的入口说明位于 tests/integration/recordings/README.md，核心实现集中在 src/ogx/testing/api_recorder.py（约 1759 行），配套的批量规范化脚本为 scripts/normalize_recordings.py。

二、目录结构：录音放在哪里

README 指出，recordings/目录存放推理操作的请求/响应 JSON 对。实际上，录音并非集中在单一目录，而是按测试文件所在目录就近存放：

tests/integration/ ├── agents/recordings/ # agents 套件的录音（865+ 个 JSON 文件） ├── responses/recordings/ # responses 套件 ├── inference/recordings/ # inference 套件 ├── messages/recordings/ ├── interactions/recordings/ ├── tool_runtime/recordings/ └── common/recordings/ # 会话级（session 级）公共录音的默认目录

这一布局由 src/ogx/testing/api_recorder.py 中的ResponseStorage._get_test_dir()决定：对测试tests/integration/inference/test_foo.py::test_bar，录音目录解析为tests/integration/inference/recordings/。目录不预先创建，而是在首次写入时通过_ensure_directory()自动mkdir(parents=True, exist_ok=True)。

存储类ResponseStorage的默认基目录由常量DEFAULT_STORAGE_DIR = REPO_ROOT / "tests/integration/common"指定（api_recorder.py），即 session 级共享录音落在tests/integration/common/recordings。

三、录音格式：每个 JSON 文件长什么样

README 定义了录音文件的核心结构：request（规范化后的请求参数：method、endpoint、body）与response（响应体，由 Pydantic 模型序列化而来）。实际写盘时（见ResponseStorage.store_recording()，api_recorder.py），文件顶层还包含test_id与id_normalization_mapping字段。

以 tests/integration/agents/recordings/007a9180a7aa38e17c1135ebf1f75e0d5ce1ea58e2261deba8c41e51196078ec.json 为例，一个真实的 OpenAI 风格录音：

{ "test_id": "tests/integration/agents/test_openai_responses.py::test_output_guardrails_unsafe_content[openai_client-txt=ollama/llama3.2:3b-instruct-fp16-False]", "request": { "method": "POST", "url": "http://0.0.0.0:11434/v1/v1/chat/completions", "headers": {}, "body": { "model": "llama-guard3:1b", "messages": [ { "role": "user", "content": "Task: Check if there is unsafe content ..." } ], "stream": false, "temperature": 0.0 }, "endpoint": "/v1/chat/completions", "model": "llama-guard3:1b" }, "response": { "body": { "__type__": "openai.types.chat.chat_completion.ChatCompletion", "__data__": { "id": "rec-007a9180a7aa", "choices": [ { "finish_reason": "stop", "message": { "content": "safe", "role": "assistant" } } ], "created": 0, "model": "llama-guard3:1b", "object": "chat.completion", "system_fingerprint": "fp_ollama", "usage": { "completion_tokens": 2, "prompt_tokens": 414, "total_tokens": 416 } } }, "is_streaming": false }, "id_normalization_mapping": {} }

几个值得注意的细节：

• Pydantic 类型保真：响应体不是普通字典，而是带__type__（完整模块路径 + 类名）和__data__的包装结构（_serialize_response，api_recorder.py）。回放时通过_deserialize_response找到原类并用model_validate/model_construct重建对象（api_recorder.py），保证测试拿到的仍是强类型 Pydantic 对象；
• 流式响应：当is_streaming: true时，body变为 chunk 列表，每个 chunk 同样按__type__/__data__包装，回放时通过生成器逐个 yield；
• 异常也可以被录制：录制模式下若真实调用抛异常，系统会通过serialize_exception把异常序列化存入录音（is_exception: true），回放时用deserialize_exception原样还原异常，从而验证错误路径；
• 模型列表特殊命名：对/v1/models、/api/tags等模型列表端点，文件名带有模型标识摘要，如models-{hash}-{digest}.json，用于区分不同服务端返回的不同模型集合（_model_identifiers_digest，api_recorder.py）。回放时_combine_model_list_responses还会把多份录音按模型 ID 做并集合并（api_recorder.py）。

四、规范化机制：让 git diff 干净如初

LLM 服务的响应中存在大量"每次运行都不同、但对测试行为无影响"的字段，如请求 ID、时间戳、耗时。若原样写入录音，每次重录都会产生巨大的 git diff，淹没真正有意义的变更。为此系统在写盘时自动执行规范化（_normalize_response，api_recorder.py）。

OpenAI 风格响应

Ollama 风格响应

Ollama 的 duration 字段依赖系统负载，是 diff 噪声的主要来源，因此全部归零。注意：只有当字段值非None时才覆盖，避免破坏结构语义。

更深层的哈希级规范化

除了响应字段，请求哈希计算阶段也做了多重建模，以保证同一次请求在录制与回放时哈希完全一致（_normalize_body_for_hash与normalize_inference_request，api_recorder.py 与 api_recorder.py）：

• 浮点数统一round(value, 5)；
• 字符串内嵌的长小数（4 位以上）统一四舍五入到 5 位，避免不同服务端浮点精度差异导致哈希漂移；
• 对 file_search 相关的向量检索分数、attributes 字典、document_id、citation 标记等运行期不稳定字段替换为占位符__NORMALIZED__（_normalize_file_search_metadata，api_recorder.py）；
• 对 Bedrock 的 OpenAI 兼容端点排除stream_options字段；
• 仅剥离extra_body/extra_query中的project_id（真实凭证与 dummy 凭证不同），但保留其他合法参数如 vLLM 的guided_choice。

这些规范化共同保证了"重录测试 → 只产生最小 diff"，让代码评审可以聚焦真实的行为变化。

五、四种运行模式与实战命令

README 给出了通过环境变量OGX_TEST_INFERENCE_MODE控制的四种模式，对应源码中的枚举APIRecordingMode（api_recorder.py）：live、record、replay、record-if-missing。模式读取逻辑在get_api_recording_mode()（api_recorder.py），未设置时默认replay，这与 tests/integration/conftest.py 中的 session 级兜底保持一致。

1. Replay 模式（默认）

OGX_TEST_INFERENCE_MODE=replay pytest tests/integration/

只允许使用已有录音，找不到录音即失败。此时api_recorder抛出RuntimeError，错误信息会附带 model、method、url 与提示命令：

Recording not found for request hash: ... Model: llama-guard3:1b | Request: POST http://0.0.0.0:11434/v1/v1/chat/completions Run './scripts/integration-tests.sh --inference-mode record-if-missing' with required API keys to generate.

这保证了 CI 与本地环境在无任何外部依赖时依然 100% 确定性地运行。

2. Record-if-missing 模式（新增测试推荐）

OGX_TEST_INFERENCE_MODE=record-if-missing pytest tests/integration/

存在录音则回放，不存在则现场调用真实 API 并落盘。这是迭代开发新测试的首选模式：先在record-if-missing下跑一遍生成全部录音，之后切回replay获得离线确定性。

3. Record 模式（强制重录）

OGX_TEST_INFERENCE_MODE=record pytest tests/integration/

强制录制所有 API 交互并覆盖已有录音。使用需谨慎：它可能因 LLM 输出变化、模型版本升级而改写录音内容，因此重录后应检查 git diff，确认只有预期内的变化。

4. Live 模式

OGX_TEST_INFERENCE_MODE=live pytest tests/integration/

完全跳过录音与回放，所有请求直连真实 API。setup_api_recording()在该模式下直接返回None，不安装任何 monkey patch（api_recorder.py）。此模式适合调试真实服务问题，但会失去确定性。

六、关键环境变量一览

结合 api_recorder.py 与 tests/integration/conftest.py，录制系统实际受以下环境变量控制：

七、请求哈希与匹配原理

录音文件以请求哈希命名（{request_hash}.json，完整 SHA256，见store_recording），回放时通过find_recording(request_hash)精确查找（api_recorder.py）。哈希由normalize_inference_request计算：method + endpoint(path) + 规范化 body + test_id组成的 JSON 经sort_keys=True排序后取 SHA256。

这里有两个关键设计：

• test_id 隔离：哈希包含当前测试的 nodeid，因此不同测试即使发出完全相同的请求，哈希也不同，录音互不串扰。test_id 通过 src/ogx/core/testing_context.py 中的ContextVar维护，tests/integration/conftest.py 的_track_test_contextfixture 在每个测试运行前set_test_context(request.node.nodeid)；
• 模型列表端点例外：/v1/models、/api/tags等基础设施级请求把 test_id 置空（_is_shared_model_list_request，api_recorder.py），因为模型发现发生在 session 初始化阶段，必须跨测试共享同一份录音。

查找还有 fallback 机制：先查测试专属目录（tests/integration/<suite>/recordings/），找不到再回退到基目录tests/integration/common/recordings/，兼容"session 级录制、测试级回放"的场景。

在server 模式下，test_id 无法通过进程内 ContextVar 传递，系统通过patch_httpx_for_test_id()（api_recorder.py）patch OpenAI/OgxClient 的_prepare_request，把__test_id注入X-OGX-Provider-Data请求头，服务端再从 header 还原上下文（_get_test_context_with_fallback，api_recorder.py）。

此外，为了保证测试内生成的资源 ID（file-、vs_、call_ 等）在录制与回放间一致，系统还通过set_id_override覆盖 OGX 的 ID 生成器，按测试上下文为每种 ID 分配确定性连续块（_allocate_test_scoped_id，api_recorder.py）。

八、覆盖范围：哪些客户端被 patch

patch_inference_clients()（api_recorder.py）一次性安装全部 monkey patch，覆盖：

• OpenAI 客户端：chat/completions、completions、embeddings、models.list、responses.create；
• Ollama AsyncClient：generate、chat、embed、ps、pull、list；
• Google genai（可选）：generate_content、generate_content_stream、embed_content（仅当 google-genai 已安装）；
• 工具运行时：Tavily 搜索（invoke_tool），工具请求按 provider + tool_name + kwargs 哈希匹配；
• aiohttp：NVIDIA/vLLM 的 rerank 端点（/rerank），在 HTTP 层拦截，确保客户端侧后处理（如max_num_results）在回放时仍真实执行；
• httpx：/v1/messages、/interactions的 Messages API 直通路径，支持非流式与 SSE 流式（aiter_lines逐行录制/重放）。

值得注意的是，文件处理器（如 PyPDF）刻意不做录制回放（_patched_file_processor_method，api_recorder.py）：它们是本地确定性操作，且file_id每次运行随机生成会导致哈希查找失败，因此始终执行真实实现。unpatch_inference_clients()在上下文退出时恢复所有原始方法。

九、重新规范化已有录音

当你更新了规范化逻辑、或想清理历史录音中的噪声字段时，运行：

python scripts/normalize_recordings.py

该脚本递归扫描tests/下所有名为recordings的目录，对每个 JSON 重新应用 OpenAI/Ollama 字段规范化（scripts/normalize_recordings.py）。脚本刻意不修改请求体——因为那会改变请求哈希，导致录音查找失配（见脚本第 64-65 行注释）。执行前建议先预览：

python scripts/normalize_recordings.py --dry-run

--dry-run只输出"Would normalize"清单与Summary: N/M files modified统计，不落盘任何文件。

十、与集成测试运行脚本的配合

录制模式通常不直接通过 pytest 环境变量驱动，而是经 scripts/integration-tests.sh 统一封装，它内部把--inference-mode映射为OGX_TEST_INFERENCE_MODE并自动启动/停止 OGX server 或 docker 容器。常用组合：

# 生成/更新录音（需要配置好 API Key 与 --setup 对应的模型） ./scripts/integration-tests.sh --stack-config server:ci-tests --suite base --setup ollama --inference-mode record # 迭代开发：缺失才录制 ./scripts/integration-tests.sh --stack-config server:ci-tests --suite base --setup ollama --inference-mode record-if-missing # 日常确定性回归（默认 replay） ./scripts/integration-tests.sh --stack-config server:ci-tests --suite base --setup ollama

运行失败时，replay模式抛出的错误信息会直接建议使用--inference-mode record-if-missing配合对应 API Key 重新生成，形成完整的开发闭环。

结语

OGX 的测试录制系统是一套兼顾"离线确定性"与"真实行为保真"的工程化方案：请求哈希保证精确匹配，test_id 注入保证测试隔离，字段规范化保证 diff 可读，Pydantic 类型包装保证回放对象强类型，四模式切换覆盖从日常回归、新增测试到强制重录的全场景。对开发者而言，理解这套机制意味着：新增测试时用record-if-missing一跑即得录音，提交前切回replay即可获得稳定、离线、零成本的集成测试体验。

【免费下载链接】ogxOpen GenAI Stack项目地址: https://gitcode.com/GitHub_Trending/ll/ogx