构建 LLM 网关 Fake-Upstream 回放校验器：Friend 仓库的结构性 Oracle 测试全解析

构建 LLM 网关 Fake-Upstream 回放校验器Friend 仓库的结构性 Oracle 测试全解析【免费下载链接】FriendAI that sees your screen, listens to your conversations and tells you what to do项目地址: https://gitcode.com/GitHub_Trending/fr/Friend本篇技术指南聚焦 Friend 仓库中backend/testing/replay_harness_llm_gateway_fake_upstream/目录所实现的LLM Gateway Fake-Upstream Oracle——一套确定性、本地闭环的结构性测试它用「回环假上游 FastAPI 依赖覆盖」驱动真实网关路由链路验证 OpenAI 兼容 Provider 终端的路径契约。读完本文你将掌握该 Oracle 的架构原理、运行方式、契约边界与数据安全设计并能在自己的 LLM 网关项目中复刻同类最小可验证的接线测试。一、它是什么一条永久生命周期的顾问型结构测试在 Friend 仓库的后端中backend/llm_gateway/是一个独立的 LLM 网关服务见 backend/llm_gateway/main.py对外提供 OpenAI 兼容的/v1/chat/completions接口内部经过路由解析resolver→ 执行器executor→ Provider 适配器三层后再调用真实的 OpenAI / OpenRouter / Perplexity / Gemini(Vertex) / Anthropic 等上游见 backend/llm_gateway/routers/dependencies.py 中注册的五个 Provider。问题随之而来如何在不触碰任何真实供应商、不泄露任何业务数据的前提下验证这条网关 → 上游的接线没有被悄悄改坏答案就是这个 Fake-Upstream Oracle。它在 backend/testing/replay_harness_llm_gateway_fake_upstream/README.md 中被明确定义为LIFECYCLE: permanent—— 一个确定性的、顾问性质的advisory结构性测试用于 LLM 网关真实路由与 Provider 终端路径。它的核心思想是用一个仅监听 127.0.0.1 回环地址的假上游fake upstream通过 FastAPI 的测试依赖覆盖dependency override把它注入真实网关进程然后驱动真实的网关路由、路由解析器、执行器和OpenAICompatibleChatCompletionProvider完成一次受控闭环调用。调用方httpx AsyncClient ASGITransport │ POST /v1/chat/completions ▼ 真实 FastAPI 网关 appllm_gateway.main │ 路由解析 resolve_chat_completion_route ▼ 真实执行器 execute_chat_completion ProviderRegistry被 override │ POST {base_url}/chat/completions ▼ 回环假上游 ThreadingHTTPServer127.0.0.1:0仅存结构性证据二、快速上手如何运行仓库根目录的 package.json 注册了 npm 脚本test:replay-llm-fake-upstream: backend/testing/replay_harness_llm_gateway_fake_upstream/run.sh因此只需在仓库根目录执行npm run test:replay-llm-fake-upstream脚本 backend/testing/replay_harness_llm_gateway_fake_upstream/run.sh 内部做了两件事前置条件校验要求存在backend/.venv/bin/python否则提示Missing backend/.venv. Run make setup first.并退出码为 1即需要先通过仓库根目录的make setup初始化 Python 虚拟环境。以模块方式运行回到仓库根目录后执行PYTHONPATHbackend backend/.venv/bin/python -m testing.replay_harness_llm_gateway_fake_upstream.oracle让llm_gateway包可被导入。成功后Oracle 会在 stdout 打印唯一一行结构化证据 JSON并以退出码 0 结束失败时打印一行错误说明并返回退出码 1。由于 stdout 被保留给证据输出运行过程中所有第三方库日志包括被 pin 住的 Starlette/httpx 组合在 import 时的告警都被主动抑制。三、核心契约它证明什么、不证明什么README 用两节Contract / 明确排除项把该测试的边界划得清清楚楚。这正是它作为结构性测试而非一致性测试的关键。3.1 假上游只证明四件事#契约点含义1真实 Provider 适配器使用预期的 chat-completions 路径假上游校验收到的请求路径必须是/v1/chat/completions2一次路由后的网关请求恰好完成一次代理隔离、有界的本地闭环往返请求经由网关 → 假上游 → 网关全程不离开本进程3网关与上游之间的事件顺序保持稳定事件序列被固定为固定四元组见下文4入站请求不能选择或重定向上游目标伪造upstream_url字段会被路由层以 4xx 拒绝且不触达 Provider 终端3.2 它明确不做什么README 强调它刻意不建立以下任何一项OpenAI / Anthropic 的 SSE 流式保真度SSE fidelityProvider 一致性provider conformance或生产端点行为客户端兼容性发布门禁release gate——即它只是顾问性测试不拦截发布捕获/回放capture/replay能力Phase 0BREADME 中指向的回放体系的后续阶段。也就是说跑通这条测试 ≠ 你的 Provider 适配器与 OpenAI 完全兼容它只证明接线没断、路径没变、事件没乱、上游不可被调用方劫持。四、源码级原理拆解核心实现在 backend/testing/replay_harness_llm_gateway_fake_upstream/oracle.py约 300 行下面按执行流水线逐段讲解。4.1 五个关键常量契约的锚点GATEWAY_PATH /v1/chat/completions ROUND_TRIP_DEADLINE_SECONDS 5.0 MAX_LOOPBACK_REQUEST_BYTES 16 * 1024 # 16 KiB 请求体上限 EXPECTED_PROVIDER_REQUEST_KEYS (messages, model, reasoning_effort, stream) EXPECTED_GATEWAY_RESPONSE_KEYS (choices, created, id, model, object)GATEWAY_PATH同时是网关路由路径与假上游期望的终端路径两者必须一致ROUND_TRIP_DEADLINE_SECONDS是整个闭环的有界时间预算MAX_LOOPBACK_REQUEST_BYTES把假上游读取的请求体限制在 16 KiB防止异常超大请求拖垮测试两个 schema keys 元组分别是进入 Provider 的请求顶层键与网关返回给调用方的响应顶层键。注意响应键中没有usage——假上游刻意返回一个不含用量字段的最小合法体。4.2 假上游_LoopbackOpenAI它用标准库http.server.ThreadingHTTPServer绑定到(127.0.0.1, 0)端口 0 表示随机空闲端口并在后台守护线程中serve_forever()。base_url属性据此动态生成return fhttp://127.0.0.1:{self._server.server_address[1]}/v1do_POST处理器的校验逻辑全部是结构性断言剥离查询串后路径必须等于GATEWAY_PATH否则报OracleFailure(provider adapter used an unexpected upstream path)记录事件upstream_request请求计数 1读取请求体校验0 Content-Length 16KiB解析为 JSON 对象只保留顶层 schema 键排序后并与EXPECTED_PROVIDER_REQUEST_KEYS比对任何变化都视为契约破坏回写一个固定的最小 chat.completion 响应id: loopback、object: chat.completion、choices含一条finish_reason: stop的空 assistant 消息记录事件upstream_response。同时它覆写了log_message为空操作避免BaseHTTPRequestHandler把请求明细打印到 stdout 污染证据任何异常都会被捕获并转成 500 响应同时存入_handler_failure并在上下文管理器__exit__中重新抛出——保证假上游自身的失败绝不被静默吞掉。4.3 注入方式只覆盖依赖不动生产构造测试没有去替换llm_gateway.main.app里的路由而是利用 FastAPI 的app.dependency_overridesregistry ProviderRegistry( { openai: OpenAICompatibleChatCompletionProvider( base_urlupstream.base_url, http_clientprovider_http_client, ), } ) app.dependency_overrides[dependencies.get_provider_registry] lambda: registry这里的关键在于生产环境里get_provider_registry是一个lru_cache(maxsize1)的缓存依赖见 backend/llm_gateway/routers/dependencies.py真实构造的OpenAICompatibleChatCompletionProvider()默认指向https://api.openai.com/v1可用OPENAI_BASE_URL覆盖。而测试通过 override 把它替换成指向回环假上游的实例于是网关路由/v1/chat/completions见 backend/llm_gateway/routers/openai_compatible.py 的Depends(get_provider_registry)在本次请求中拿到的就是假 Provider生产构造路径完全不被触碰正如 README 所述Normal provider construction and routing remain unchanged outside this test process.测试结束的finally块会恢复之前的dependency_overrides并调用registry.aclose()释放 httpx 客户端。4.4 双客户端隔离trust_envFalse的深意Oracle 里所有 httpx 客户端都经由_loopback_http_client()构造它强制httpx.AsyncClient(trust_envFalse, **kwargs)该 harness 同时拥有两个端点。如果继承开发者或 CI 环境的代理设置假 Provider 请求可能被路由到进程之外而不是 127.0.0.1所以每个 harness 自有的客户端都显式退出。也就是说trust_envFalse是为了保证代理隔离proxy isolation无论开发机或 CI 上是否设置了HTTP_PROXY/HTTPS_PROXY回环流量都只会在本进程内完成。调用侧客户端还使用了httpx.ASGITransport(appapp, raise_app_exceptionsTrue)让 httpx 直接在进程内驱动真实 ASGI 应用base_url 用http://gateway.test占位不启动任何真实 TCP 监听——真实网关进程对调用方而言就是一个库内对象。4.5 有界请求asyncio.timeout而非事后计时_post_with_deadline的注释非常直白对请求本身设界而不是在事后检查已耗时间。async with asyncio.timeout(deadline_seconds): return await client.post(path, jsonpayload, headersheaders)超时TimeoutError会直接转成OracleFailure(gateway loopback round trip exceeded its bounded deadline)。这保证即使网关或 Provider 死锁测试也会在 5 秒内给出确定性失败。4.6 证据环境与日志纪律run_oracle()用三个上下文管理器包裹整个执行_temporary_environment临时注入OMI_LLM_GATEWAY_SERVICE_TOKENloopback-test-token与OPENAI_API_KEYloopback-test-key这是网关服务鉴权与 Provider 鉴权所需的最小环境用完即恢复绝不污染外部环境_bounded_evidence_logging把llm_gateway.gateway.metrics日志器临时提升到CRITICAL保证 metrics 日志不会混入 stdout 证据流_LoopbackOpenAI上下文启动/关闭假上游。五、两次请求、四个事件完整执行剧本5.1 第一次请求合法闭环测试发送的请求体刻意使用空内容见_valid_gateway_requestreturn {model: omi:auto:chat-structured, messages: [{role: user, content: }]}注释点明空内容只用来证明路由契约同时从根上杜绝任何提示词被假上游留存。请求头为Authorization: Bearer loopback-test-token与X-Omi-Service-Caller: backend。断言链依次为响应状态类必须为2xx响应体必须是 JSON 对象且排序后的顶层键必须等于EXPECTED_GATEWAY_RESPONSE_KEYS此时假上游事件已累计为upstream_request、upstream_response网关侧追加gateway_response。5.2 第二次请求试图劫持上游随后 Oracle 用同一个客户端再次 POST但 payload 在合法请求之上多塞了一个upstream_url字段指向假上游自己的 base_urlpayload{**_valid_gateway_request(), upstream_url: upstream.base_url}断言链响应状态类必须为4xx_status_class把 400 归为4xx错误体中的error.code必须等于invalid_request关键upstream.request_count必须仍然为 1——即这个恶意字段在路由层就被拒绝从未触达 Provider 终端路径。这正是契约第 4 条的落地证明外部调用方无法选择或重定向上游目标。5.3 固定事件顺序最终断言假上游的完整事件序列必须精确等于[upstream_request, upstream_response, gateway_response, redirect_rejected]任何多出、缺失或乱序都会触发OracleFailure(gateway and provider terminal event ordering changed)——这锁死了网关→上游→网关的方向性防止某次重构意外引入先响应后请求之类的乱序。六、输出证据一份自描述的 JSON成功时 stdout 打印sort_keysTrue字段顺序稳定便于 Agent / 脚本解析{ bounded_round_trip: within_5_seconds, endpoint_path: /v1/chat/completions, error_classes: [invalid_request], event_order: [upstream_request, upstream_response, gateway_response, redirect_rejected], oracle: replay-llm-gateway-fake-upstream, provider_request_count: 1, request_schema_keys: [messages, model, reasoning_effort, stream], response_schema_keys: [choices, created, id, model, object], status_classes: [2xx, 4xx] }这份证据与 README 的Data safety白名单一一对应只包含端点路径形状、事件标签、请求/响应 schema 键、状态/错误类、请求计数与一个有界计时结果。七、数据安全结构性证据白名单README 的 Data safety 一节是这条测试的设计底线值得单独强调假上游只保留端点路径形状、事件标签、请求/响应 schema 键、状态/错误类、请求计数和有界计时结果。它从不记录或持久化提示词内容、补全结果、token 值、Provider 请求体、header 值、凭据或标识符。从源码可以印证这一承诺是如何被结构性地保证的假上游解析请求体后只取sorted(request_body)的键集合任何键值都不会被写入事件或证据见_LoopbackOpenAI.do_POST测试请求的content恒为空字符串从源头上无提示词可采集响应体是固定的loopback占位内容不含任何业务补全_bounded_evidence_logging与log_message覆写进一步杜绝日志侧泄漏临时环境变量在finally中恢复原值loopback-test-key这类假凭据绝不外泄。对含敏感数据的网关项目而言这种测试自身必须无痕的设计正是它可以在 CI 与本地反复执行的前提。八、与真实 Provider 适配器的对应关系这条 Oracle 之所以有意义是因为它驱动的确实是生产代码路径。对照 backend/llm_gateway/gateway/providers.py 中OpenAICompatibleChatCompletionProvider.create_chat_completion的实现它向{base_url}/chat/completions发起流式 POSTself._http_client.stream(...)正是假上游校验的GATEWAY_PATH它构造Authorization: Bearer {api_key}而测试注入的OPENAI_API_KEYloopback-test-key恰好满足_resolve_provider_api_key的os.getenv(api_key_env)分支非 BYOK 模式它对响应调用_validate_chat_completion_response_shape要求object chat.completion、非空id/model、choices非空且 message 的role assistant——假上游的最小响应体正好通过这些校验它在读取响应时使用_read_limited_response默认上限DEFAULT_MAX_RESPONSE_BYTES 2 * 1024 * 1024与 Oracle 侧MAX_LOOPBACK_REQUEST_BYTES的有界理念一脉相承。也就是说Oracle 断言的不是 Provider 的行为细节而是网关按约定把请求交给了 ProviderProvider 按约定把合法响应还给了网关这一条完整接线。九、使用场景与边界建议适合使用该 Oracle 的场景每次改动llm_gateway的路由、executor、Provider 构造、依赖注入或鉴权逻辑后作为回归防线快速验证接线未断在 CI 中对网关服务做零外部依赖、零敏感数据的最小冒烟作为开发期脚手架为后续更完整的捕获/回放capture/replay或 Provider 一致性套件打底。需要注意的边界README 已明确它不验证 SSE 流式行为、不验证 Provider 一致性、不是发布门禁它不提供捕获/回放能力也不覆盖 Phase 0B运行前置条件是make setup已生成backend/.venv证据 JSON 是结构性的不要指望从中得到性能或兼容性结论。从仓库结构看这套testing/replay_harness_llm_gateway_fake_upstream/目录README.mdoracle.pyrun.sh__init__.py设计为完全自包含README 说明契约与数据安全oracle.py 承载全部逻辑run.sh 封装运行前置。读者若要在自己的项目中复刻只需保持回环假上游 依赖覆盖 结构化证据白名单 固定事件序列这四个支柱即可获得一个廉价、确定性、无数据风险的网关接线测试。【免费下载链接】FriendAI that sees your screen, listens to your conversations and tells you what to do项目地址: https://gitcode.com/GitHub_Trending/fr/Friend创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考