企业官网建设流程全解析

Friend 项目 LLM 网关流式传输 Wire-Fidelity Oracle：回放 Harness 守护 OpenAI 兼容 SSE 协议

【免费下载链接】FriendAI that sees your screen, listens to your conversations and tells you what to do项目地址: https://gitcode.com/GitHub_Trending/fr/Friend

本文介绍 Friend 开源仓库（项目口号为"AI that sees your screen, listens to your conversations and tells you what to do"）中 LLM 网关（Omi LLM Gateway）的确定性结构协议测试——Replay Harness LLM Streaming Wire-Fidelity Oracle。它通过本地回环（loopback）假上游以刻意分片的 HTTP chunk 注入合成 SSE 帧，驱动真实的网关路由、解析器、执行器与 OpenAI 兼容流式 Provider，从而以结构证据守护流式终端的线级契约。读完本文，你将理解该 Oracle 的契约边界、运行方式、源码实现，以及它在不触碰任何真实 Provider 的前提下如何验证流式响应的帧序、终止标记与安全拒绝策略。

一、为什么需要"线级保真"Oracle

LLM 网关对外暴露 OpenAI 兼容的/v1/chat/completions流式端点，当请求携带"stream": true时，网关需要把上游 Provider 的 SSE（Server-Sent Events）字节流原样转发给客户端，并在过程中完成用量记账、终止状态观测与（可选）JIT 预算收据注入。这条链路上最容易悄悄退化的不是"能否返回结果"，而是线级协议形态：

• 上游将 SSE 帧拆散在多个 HTTP chunk 中时，增量读取是否仍然完整？
• 客户端收到的帧顺序是否仍是"若干非终止数据帧 → 一个终止 usage/finish 帧 → 恰好一个[DONE]"？
• 请求是否真的请求了流式与终止用量（stream_options.include_usage）？
• 恶意或误用请求注入upstream_url试图让网关转发到任意目标时，是否在进入 Provider 路径之前就被拒绝？

传统的"只要 200 且内容可解析"的测试无法覆盖上述形态。Wire-Fidelity Oracle 的设计目标，就是把上述结构性质固化为一项**确定性、可重复、咨询性（advisory）**的守护测试。

二、Oracle 的定位与运行方式

该 Oracle 位于 backend/testing/replay_harness_llm_streaming_wire_fidelity/README.md，生命周期标记为permanent（永久），属于确定性的结构协议测试。运行入口定义在仓库根目录的 package.json：

"test:replay-llm-streaming-wire-fidelity": "backend/testing/replay_harness_llm_streaming_wire_fidelity/run.sh"

执行命令：

npm run test:replay-llm-streaming-wire-fidelity

底层的 run.sh 会先检查backend/.venv/bin/python是否存在（不存在则提示先执行make setup），随后以PYTHONPATH=backend的方式运行 Oracle 模块：

PYTHONPATH=backend backend/.venv/bin/python -m testing.replay_harness_llm_streaming_wire_fidelity.oracle

Oracle 成功时向 stdout 打印一份结构化证据 JSON 并以退出码 0 结束；断言失败时打印Replay LLM streaming wire-fidelity oracle failed: <异常类型>并以退出码 1 结束（见 oracle.py）。

三、工作机制：回环假上游驱动真实网关全链路

这是本 Oracle 最核心的设计：测试只替换"上游是谁"，绝不替换被测的网关代码。它启动一个仅监听回环地址的 OpenAI 兼容假上游（ThreadingHTTPServer，绑定127.0.0.1），向其中注入合成的 SSE 帧；而请求路径上跑的完全是生产实现：

• FastAPI 网关应用（backend/llm_gateway/main.py）
• 路由解析器（resolver）与执行器（executor）
• OpenAICompatibleChatCompletionProvider.stream_chat_completion（backend/llm_gateway/gateway/providers.py）
• Starlette 的StreamingResponse

测试通过 FastAPI 的app.dependency_overrides注入两个测试专属依赖（oracle.py）：

app.dependency_overrides[dependencies.get_gateway_config] = _streaming_enabled_gateway_config app.dependency_overrides[dependencies.get_provider_registry] = lambda: registry

其中_streaming_enabled_gateway_config从生产配置加载器load_gateway_config(prod_mode=True)读取真实配置，然后仅为被测 lane（omi:auto:chat-structured）打开streaming能力（oracle.py）；Provider 注册表则以假上游地址构造一个OpenAICompatibleChatCompletionProvider。生产路由、Provider 构造、重试、Schema 与部署配置全部保持不变。

3.1 合成 SSE 与刻意分片

假上游会通过httpx流式 POST 收到网关转发的请求，并校验请求形态，然后返回三段合成 SSE（oracle.py）：

data: {"choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]} data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":0,"completion_tokens":0,"total_tokens":0}} data: [DONE]

第一帧是非终止的角色增量帧，第二帧携带finish_reason: "stop"与合成 usage 字段（为了让网关的终止用量处理路径得以执行），第三帧是[DONE]哨兵。两帧内部与帧边界处都没有任何真实用户内容。

关键在传输层：这三段合成 SSE 被拆成 5 个片段，切点落在 SSE 字段内部和帧边界之间（oracle.py），并且以Transfer-Encoding: chunked逐个写出发送：

for fragment in _SSE_FRAGMENTS: self.wfile.write(f"{len(fragment):X}\r\n".encode("ascii")) self.wfile.write(fragment) self.wfile.write(b"\r\n") self.wfile.flush() self.wfile.write(b"0\r\n\r\n")

也就是说，一个完整 SSE 帧可能被 HTTP chunk 边界从data:字段中间切开。这就要求网关侧的增量读取器（Provider 的response.aiter_bytes()+SSEEventDecoder）具备跨 chunk 组帧的能力，这正是该测试要证明的核心线级性质。

四、契约（Contract）：Oracle 只证明四件事

README 明确限定，Oracle 只证明以下四项声明：

1. Provider 请求选择流式与终止用量：上游收到的请求中stream必须为true，且stream_options.include_usage必须为true（oracle.py）。
2. 网关以text/event-stream响应：客户端响应头的content-type必须以text/event-stream开头（oracle.py）。
3. 分片上游 SSE 产生"非终止数据帧 → 终止 usage/finish 帧 → 恰好一个[DONE]"的客户端帧序：这是核心语义顺序断言（见下节）。
4. 入站upstream_url在 Provider 路径之前被拒绝，且拒绝发生在有界本地期限内，同时任何非回环 socket 出网都被禁止。

4.1 客户端帧序的语义断言

Oracle 使用网关自带的SSEEventDecoder（backend/llm_gateway/gateway/sse.py）对客户端收到的完整响应体做增量解码，只保留帧标签与 Schema 键集合，然后分类（oracle.py）：

• data为[DONE]→ 标记为done；
• 携带usage对象且choices[0].finish_reason非空 → 标记为terminal_usage_finish；
• 无 usage 且finish_reason为空 → 标记为nonterminal_data；
• 其余形态（如无choices的帧、无 finish reason 却带 usage 的帧、有 finish reason 却无 usage 的终止数据帧）一律判定为失败。

随后_assert_semantic_stream_order断言：整个响应中done恰好出现一次；帧数不少于 3；末尾两帧依次必须是terminal_usage_finish与done；其余帧必须全部是nonterminal_data（oracle.py）。由此，"分片上游 → 网关转发" 后客户端侧的帧序被严格钉死。

4.2 有界期限与非回环出网拒绝

整个往返（round trip）被限定在ROUND_TRIP_DEADLINE_SECONDS = 5.0秒内完成，超时即失败（oracle.py）。同时，Oracle 用unittest.mock.patch替换socket.socket.connect，任何目标主机不在{"127.0.0.1", "::1"}内的连接尝试都会立刻抛出OracleFailure("non-loopback socket egress was attempted")（oracle.py）。这保证该测试是**封闭（hermetic）**的：即使被测代码行为异常，也不会真的打到外网。

4.3 重定向拒绝验证

在流式往返成功后，Oracle 再发一个携带upstream_url字段的请求（oracle.py），断言：

• 响应状态类必须是4xx（网关不能接受入站上游目标）；
• 错误对象的code必须是invalid_request；
• 假上游收到的请求计数仍为 1——即该请求根本没有到达 Provider 流式路径就被网关拦截。

最后，Oracle 校验假上游记录的事件顺序严格等于["upstream_request", "upstream_fragmented_sse", "gateway_stream_response", "redirect_rejected"]（oracle.py），保证网关与 Provider 的交互时序没有漂移。

五、残余边界：Oracle 明确"不做什么"

README 用一节 "Residual boundary" 划清了职责边界，避免测试被误用。它不是：

• Provider 一致性测试（provider conformance）；
• 生产端点行为测试；
• 客户端兼容性测试；
• 流量捕获/回放（traffic capture/replay）；
• 发布门禁（release gate）；
• LC3 或时序资格认证；
• Phase 0B。

它既不改变、也不替代 Listen Pusher 或 Sync Cloud Tasks 的防护测试（gauntlets），并且从不发起任何真实 Provider 调用。换句话说，它是一台"结构体检仪"，只回答"线级协议形态是否还正确"这一个问题，业务正确性、延迟目标、真实 Provider 行为由其他测试体系负责。

六、数据安全：只保留白名单内的结构证据

由于假上游会解析请求体与 SSE 信封，Oracle 对数据留存做了严格限制（README 与 oracle.py 均有声明）：

• 保留并打印：endpoint/帧标签与顺序、Schema 键集合、状态/错误类别、一个时间桶（timing bucket）、请求计数；
• 绝不记录或持久化：prompt/completion/token 值、SSE payload 内容、请求/响应头、凭据、任何标识符、Provider 响应体。

实现上有多重保障：测试请求的消息内容刻意置空（{"role": "user", "content": ""}，oracle.py），合成 SSE 不含用户内容；同时_bounded_evidence_logging会把llm_gateway.gateway.metrics与httpx两个 logger 临时提升到CRITICAL，把执行期日志输出收敛在结构证据白名单内（oracle.py）；假上游的BaseHTTPRequestHandler.log_message也被覆写为空操作，避免打印请求细节（oracle.py）。

Oracle 成功时输出的证据 JSON 字段包括：oracle、endpoint_path、event_order、frame_order、provider_request_schema_keys、client_frame_schema_key_sets、status_classes、error_classes、provider_request_count、bounded_round_trip（oracle.py）。

七、被测流式路径如何与之配合（源码印证）

Oracle 断言的性质，恰好对应网关流式实现中的几处关键代码，两者互为印证：

• Provider 请求注入终止用量：网关在 backend/llm_gateway/routers/openai_compatible.py 的_request_stream_usage中，对openai、openrouter、perplexity三类 Provider 自动补写stream_options.include_usage = True——这正是契约第 1 条在上游侧看到include_usage的来源。
• 增量读取保留分片：OpenAICompatibleChatCompletionProvider.stream_chat_completion使用httpx.AsyncClient.stream(...).aiter_bytes()逐块产出字节（providers.py），因此上游 HTTP chunk 的刻意分片会原样进入网关的增量解码器。
• SSE 增量解码：SSEEventDecoder.feed()维护跨 chunk 的字节缓冲，统一\r\n/\r/\n换行并只在出现\n\n时切出一个完整帧，缓冲区上限为 1 MiB（backend/llm_gateway/gateway/sse.py）。Oracle 正是复用该解码器来分类客户端帧。
• 流式响应与终止观测：网关在 openai_compatible.py 通过_streaming_response返回StreamingResponse(media_type='text/event-stream')，并在_stream_with_terminal_metrics中观测[DONE]触发终止记账（observe_terminal），把eof_before_terminal_marker、invalid_sse_frame、transport_midstream等异常分类为终止错误——这些语义正是契约第 3 条"恰好一个[DONE]"背后的运行时逻辑。值得注意的实现细节：网关用原子分组正则(?>\r\n|\r|\n){2}作为 JIT 收据注入的帧边界（openai_compatible.py），避免回溯把多行 SSE 事件拆散。
• 错误响应的 OpenAI 分类：重定向/非法请求被GatewayInvalidRequestError抛出后，_error_response会组装{message, type, param, code}结构，其中code取GatewayErrorCode.INVALID_REQUEST的字符串值（openai_compatible.py），对应 Oracle 断言的invalid_request错误类别，状态码映射到400（4xx 类）。

八、如何阅读运行结果

运行npm run test:replay-llm-streaming-wire-fidelity后：

• 成功：stdout 输出一行排序后的证据 JSON（如{"oracle": "replay-llm-streaming-wire-fidelity", "event_order": ["upstream_request", "upstream_fragmented_sse", "gateway_stream_response", "redirect_rejected"], "frame_order": ["nonterminal_data", "terminal_usage_finish", "done"], "status_classes": ["2xx", "4xx"], ...}），退出码 0。
• 失败：stdout 打印Replay LLM streaming wire-fidelity oracle failed: <异常类型名>，退出码 1，失败类型对应源码中的各类OracleFailure（如gateway did not emit exactly one done marker、gateway accepted an inbound upstream target、non-loopback socket egress was attempted等），可直接定位协议形态漂移的位置。

由于测试是确定性的（合成帧固定、回环环境固定），同一份代码每次运行结果应当完全一致，因此它适合作为 CI 的常规守护项，也适合在改动 resolver、executor、SSE 解码、流式 Provider 或路由鉴权代码后本地先行验证。

九、总结

Friend 仓库的 Replay Harness LLM Streaming Wire-Fidelity Oracle 是一个设计克制、边界清晰的协议守护测试：用回环假上游 + 合成分片 SSE 驱动真实网关全链路，只断言"流式请求形态、text/event-stream响应、客户端帧序、upstream_url拒绝、非回环出网禁止"这组线级性质，同时通过白名单证据输出与空内容请求把数据安全做到极致。它以永久生命周期常驻在 backend/testing/replay_harness_llm_streaming_wire_fidelity/ 下，与 backend/llm_gateway 的流式实现构成一对可互相印证的"契约 + 实现"，是研究 OpenAI 兼容流式网关如何被系统化验证的很好范本。

【免费下载链接】FriendAI that sees your screen, listens to your conversations and tells you what to do项目地址: https://gitcode.com/GitHub_Trending/fr/Friend