From e5319c5138c276a4046963b766e44d8124863c29 Mon Sep 17 00:00:00 2001 From: H2Rhan <2373723338@qq.com> Date: Wed, 9 Sep 2026 19:54:51 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20v1.7=20=E5=90=8C=E5=8D=8F=E8=AE=AE?= =?UTF-8?q?=E7=9B=B4=E9=80=9A=E6=B5=81=E5=BC=8F=20+=20=E6=9C=AA=E5=AE=9E?= =?UTF-8?q?=E7=8E=B0=E6=B5=81=E5=BC=8F=E6=96=B9=E5=90=91=E6=98=BE=E5=BC=8F?= =?UTF-8?q?=20501?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 直通流式（chat←chat / anthropic←anthropic）：tee_lines 逐行透传字节不动；旁路收集流尾汇总（anthropic 复用 AnthropicToChatStream，chat 新增 ChatStreamCollector）；落库/埋点仍走 finalize_turn 单一路径 - 隐藏 bug 修复：stream:true 落到未实现方向此前被 post_json 当 JSON 解析炸成 502，现显式 501（附已支持方向清单） - 客户端中途断开：已收部分照常落账 - mock：/chat/completions 发 Chat 形状 chunk（role 首帧 + finish + usage + [DONE]） - 测试：TestChatStreamCollector 3 项单测 + 冒烟第 7 组 7 项（透传文本拼合 / 逐块到达 ~31ms / 501）；56 单测 + 25 冒烟全过 - 文档：LIMITATIONS #1 范围更新；README/REVIEW/capability-matrix/CHANGELOG 同步 --- CHANGELOG.md | 8 ++++ README.md | 8 ++-- docs/LIMITATIONS.md | 2 +- docs/REVIEW.md | 13 ++++--- docs/capability-matrix.md | 2 +- src/gateway/server.py | 80 ++++++++++++++++++++++++++++++++++++--- src/gateway/sse.py | 65 +++++++++++++++++++++++++++++++ tests/test_offline.py | 48 ++++++++++++++++++++++- tools/mock_backend.py | 21 ++++++++++ tools/smoke_e2e.py | 48 +++++++++++++++++++++++ 10 files changed, 276 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2b2852d..c82f7f8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -68,3 +68,11 @@ - mock：Anthropic 路径发完整官方事件序列（message_start 带 usage 与缓存字段 → content_block_* → message_delta → message_stop） - 测试：`TestSseConversion` 7 项单测（分帧容错 / 即时下发 / 参数缓冲 / stop_reason 映射 / usage 合并 / 错误帧）；冒烟第 6 组 7 项（**实测首尾帧间隔 30.7ms，证明逐块到达非整读补吐** + 流式轮照常进埋点）。53 项单测 + 18 项冒烟全过 - 文档：LIMITATIONS #1 标「部分闭环」（其余方向与 dropped 可见性写明）；README SSE 行与测试口径同步；REVIEW 网关文件清单补 sse.py + +## 2026-09-09 · v1.7 同协议直通流式 + 未实现方向显式 501（本提交） + +- **直通流式**（chat←chat / anthropic←anthropic）：`tee_lines` 三通逐行透传（字节不动、即时 flush），旁路收集器攒流尾汇总——anthropic 方向复用 `AnthropicToChatStream`（帧丢弃只取 usage/文本），chat 方向新增 `ChatStreamCollector`（工具槽位拼接、脏数据容错）；落库/埋点仍走 `finalize_turn` 单一路径 +- **隐藏 bug 修复**：此前**任何** stream:true 请求落到未实现方向，`post_json` 会把 SSE 当 JSON 解析、炸成语焉不详的 502；现改为显式 **501**（附已支持方向清单，提示改用 stream:false） +- 客户端中途断开：已收到的部分照常落账（费用已实际发生） +- mock：`/chat/completions` 路径发 Chat 形状 chunk（role 首帧 + finish + usage + [DONE]） +- 测试：`TestChatStreamCollector` 3 项单测；冒烟第 7 组 7 项（两个透传方向文本拼合 / 逐块到达 ~31ms / [DONE] / 501）。56 项单测 + 25 项冒烟全过 diff --git a/README.md b/README.md index f502204..0f2d431 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ OpenAI Chat / OpenAI Response ↔ Anthropic 的协议转换层（网关/代理 + | 状态层（SQLite 持久化）+ 可切换 Session 配置 | ✅ v1.1 | `src/state/` + `config/session.json`，重启不丢（`TestStateLayer`） | | 三对 adapter（字段映射） | ✅ 离线 | `src/adapters/`，录制样例可测（`TestAdapterRoundTrip`） | | 网关转发 | ✅ 离线（mock）/ Groq 真实链路已验证 | `src/gateway/`；E29 真实端点对账 | -| SSE 流式 | ◐ **v1.6 部分逐块**（chat 客户端 ← anthropic 上游） | `src/gateway/sse.py` 逐块转换（冒烟实测 30.7ms 逐块到达）；其余方向整读透传 → `docs/LIMITATIONS.md` #1；E21 证明流式无损命中率 | +| SSE 流式 | ◐ **v1.7 三方向逐块**（chat←anthropic 转换 + 同协议透传） | `src/gateway/sse.py`；未实现方向显式 501 → `docs/LIMITATIONS.md` #1；E21 证明流式无损命中率 | | 记忆注入（幂等去重 + memory_cap） | ✅ v1.1 新增 | `inject_memories` + `TestGatewayPolicies`；管理口 `/v1/admin/session/meta` | | 工具调用全链路（含 ID 双向映射） | ✅ **v1.4** | 跨协议参数/签名不丢（`TestCrossProtocolToolArgs`/`TestThinkingSignature`）；`src/state/idmap.py` 会话级映射（`TestToolIdMap`） | | 状态层重放（结构化块） | ✅ **v1.4** | thinking（signature/redacted）与 tool_use 入历史（`TestAssistantFromUpstream`） | @@ -39,8 +39,8 @@ OpenAI Chat / OpenAI Response ↔ Anthropic 的协议转换层（网关/代理 + ```bash # 零外部依赖（仅标准库），Python 3.11+ -python -m unittest tests.test_offline -v # 53 项离线单测：IR 往返 / adapter / 状态层 / 预热 / 三轮自查回归 / 多轮链 E2E / ID 映射 / property-based / 惰性淘汰 / SSE 转换 -python tools/smoke_e2e.py # 18 项端到端冒烟：真起 mock+网关子进程，6 组链路断言（含 SSE 逐块流式） +python -m unittest tests.test_offline -v # 56 项离线单测：IR 往返 / adapter / 状态层 / 预热 / 三轮自查回归 / 多轮链 E2E / ID 映射 / property-based / 惰性淘汰 / SSE 转换与透传 +python tools/smoke_e2e.py # 25 项端到端冒烟：真起 mock+网关子进程，7 组链路断言（含 SSE 转换/透传/501） python tools/mock_backend.py # 起 mock backend（127.0.0.1:9100，按端点返回三种协议形状） python -m src.gateway.server # 起网关（转发到 mock） python experiments/run_experiments.py --dry-run # 实验框架 dry-run（不真实调 API） @@ -95,7 +95,7 @@ src/ ir/ IR 数据模型（L0/L1/L2） adapters/ chat / response / anthropic 三对 adapter state/ 会话表 + TTL + session 配置 + build_prefix - gateway/ HTTP 转发 + SSE 逐块转换（chat←anthropic 已实现 sse.py，其余方向整读透传见 docs/LIMITATIONS.md #1） + gateway/ HTTP 转发 + SSE 流式（chat←anthropic 逐块转换 + 同协议透传；未实现方向 501，见 docs/LIMITATIONS.md #1） observability/ 命中率埋点（5 项暴露） warmup/ max_tokens:0 预热 webui/ 弹网页本地仪表盘（安全骨架 + 记忆编排界面） diff --git a/docs/LIMITATIONS.md b/docs/LIMITATIONS.md index 894cf00..666eef0 100644 --- a/docs/LIMITATIONS.md +++ b/docs/LIMITATIONS.md @@ -8,7 +8,7 @@ ### #1 SSE 逐块流式转换（v1.6 部分闭环：chat 客户端 ← anthropic 上游） -> ◐ **v1.6 部分闭环**（2026-09-09）：`openai_chat` 客户端 ← `anthropic` 上游方向已实现**逐块流式**——`src/gateway/sse.py` 把 Anthropic 事件逐块翻成 Chat chunk 即时下发（冒烟实测首尾帧间隔 30.7ms，证明非整读补吐）；text_delta 即时下发、工具参数按既定折损缓冲到 `content_block_stop`、usage 流尾合并后照常进 5 项埋点、历史落库复用 `assistant_from_upstream`（流式与非流式同一路径）。回归测试 `TestSseConversion` 7 项 + 冒烟第 6 组 7 项。**仍为整读透传的方向**：chat←chat、response 源、anthropic 客户端；流式轮的 dropped 清单对客户端不可见（SSE 帧无其位置，埋点中 degradation 仍如实记录）。以下原文留档，未闭环范围以四要素为准。 +> ◐ **v1.6 部分闭环**（2026-09-09）：`openai_chat` 客户端 ← `anthropic` 上游方向已实现**逐块流式**——`src/gateway/sse.py` 把 Anthropic 事件逐块翻成 Chat chunk 即时下发（冒烟实测首尾帧间隔 30.7ms，证明非整读补吐）；text_delta 即时下发、工具参数按既定折损缓冲到 `content_block_stop`、usage 流尾合并后照常进 5 项埋点、历史落库复用 `assistant_from_upstream`（流式与非流式同一路径）。回归测试 `TestSseConversion` 7 项 + 冒烟第 6 组 7 项。**v1.7 补充**（2026-09-09）：同协议直通流式（chat←chat / anthropic←anthropic）已闭环——字节原样透传 + 旁路收集流尾汇总（`tee_lines` / `ChatStreamCollector`）；**未实现的流式方向（response 源三维寻址、anthropic←chat 等）改为显式 501**（此前是 `post_json` 把 SSE 当 JSON 解析炸成语焉不详的 502）。**仍未闭环**：response 源、anthropic←chat 跨协议方向的流式转换；流式轮的 dropped 清单对客户端不可见（埋点中 degradation 仍如实记录）。以下原文留档，未闭环范围以四要素为准。 - **影响半径**：流式体验与首 token 延迟（TTFB）。功能正确性不受影响——响应内容完整、usage 归一与埋点照常工作。 - **触发条件**：客户端请求 `stream:true` 经过网关。当前网关 `post_json` 整读响应后一次性返回；跨协议的事件模型转换（Chat `delta` / Responses 三维寻址 / Anthropic 块生命周期）未实现。 diff --git a/docs/REVIEW.md b/docs/REVIEW.md index 025353a..5fd6275 100644 --- a/docs/REVIEW.md +++ b/docs/REVIEW.md @@ -1,7 +1,7 @@ # 代码评审材料（Code Review Pack） > 版本 v1.2（2026-09-02）｜ src 1493 行 + tools/experiments/tests 1065 行（纯标准库，零第三方依赖，Python 3.11+） -> 测试：**53 项单测全过 + 18 项端到端冒烟全过**（2026-09-09，含第三轮自查回归 + property-based 性质测试 + 惰性淘汰回归 + SSE 逐块转换）+ Groq 真实链路验证通过（2026-09-01） +> 测试：**56 项单测全过 + 25 项端到端冒烟全过**（2026-09-09，含第三轮自查回归 + property-based 性质测试 + 惰性淘汰回归 + SSE 转换/透传）+ Groq 真实链路验证通过（2026-09-01） > 本版差异：**两轮代码自查发现 9 个真实 bug，全部修复并各配回归测试**（见第六节） ## 一、整体架构（IR 星型） @@ -66,7 +66,7 @@ do_POST（路由 /v1/{source}/to/{target}，未知协议 → 400） | 文件 | 行数 | 功能 | | ---- | --- | ---- | | `server.py` | 363 | 见下方「网关关键函数」 | -| `sse.py` | 150 | **v1.6 新增**：SSE 逐块流式转换（chat 客户端 ← anthropic 上游）。`parse_sse_lines` 行流分帧；`AnthropicToChatStream` 事件→Chat chunk：text_delta 即时下发、工具参数缓冲到 content_block_stop（既定折损）、usage 流尾合并、`synthetic_response()` 复用 `assistant_from_upstream` 落库 | +| `sse.py` | 240 | **v1.6/v1.7**：SSE 流式三件套。`parse_sse_lines` 行流分帧；`AnthropicToChatStream` 事件→Chat chunk 逐块转换（chat 客户端 ← anthropic 上游）：text_delta 即时下发、工具参数缓冲到 content_block_stop（既定折损）、usage 流尾合并、`synthetic_response()` 复用 `assistant_from_upstream` 落库；`tee_lines` + `ChatStreamCollector` 同协议直通透传（字节不动、旁路汇总） | **网关关键函数**： @@ -111,9 +111,9 @@ do_POST（路由 /v1/{source}/to/{target}，未知协议 → 400） | ---- | --- | ---- | | `mock_backend.py` | 194 | 离线顶替真实端点。**v1.2 按端点路径返回三种协议各自形状**（/v1/messages、/responses、/chat/completions），mock 与真实后端走同一条转发路径；有状态缓存模拟（前缀哈希记账）；支持 `max_tokens:0` 预热形态与 SSE | | `verify_cache.py` | 152 | **验站脚本**：唯一随机 run_id 前缀两步验证（预热看 creation → 复发看 read）；PASS/PARTIAL/FAIL；`max_tokens:0` 被拒自动降级 1 | -| `smoke_e2e.py` | 220 | 端到端冒烟——真起 mock + 网关两个子进程，跑 6 组 18 项断言（直通/跨协议/多轮链/预热/非法路由/**SSE 逐块流式**），退出码 0/1 | +| `smoke_e2e.py` | 220 | 端到端冒烟——真起 mock + 网关两个子进程，跑 7 组 25 项断言（直通/跨协议/多轮链/预热/非法路由/SSE 转换/**SSE 透传+501**），退出码 0/1 | -### tests/ —— 测试（53 项全过） +### tests/ —— 测试（56 项全过） | 测试类 | 覆盖 | | ------ | ---- | @@ -130,13 +130,14 @@ do_POST（路由 /v1/{source}/to/{target}，未知协议 → 400） | `TestAssistantFromUpstream` | **v1.4 新增**：状态层重放保留 thinking / tool_use 结构化块 | | `TestPropertyRoundTrip` | **v1.4 新增**：property-based 往返（440 随机用例、固定种子可复现）：断点恒在 [3,4] / 未知字段必降级 / 工具参数守恒 / signature 守恒 | | `TestSseConversion` | **v1.6 新增**（7 项）：SSE 分帧（注释行/半帧容错）、text_delta 即时下发、工具参数缓冲到块结束、stop_reason 映射、usage 流尾合并 + 落库复用、上游错误帧 | +| `TestChatStreamCollector` | **v1.7 新增**（3 项）：chat 直通流式旁路收集（文本+usage 汇总、工具槽位拼接、脏数据容错） | ## 四、已验证记录 | 验证 | 结果 | 日期 | | ---- | ---- | ---- | -| 53 项单测（`-W error::ResourceWarning` 下零警告，含第三轮自查回归 + property-based + 惰性淘汰 + SSE 转换） | ✅ 全过 | 2026-09-09 | -| 18 项端到端冒烟（子进程级，`tools/smoke_e2e.py`，含 SSE 逐块到达实测 30.7ms） | ✅ 全过 | 2026-09-09 | +| 56 项单测（`-W error::ResourceWarning` 下零警告，含第三轮自查回归 + property-based + 惰性淘汰 + SSE 转换/透传） | ✅ 全过 | 2026-09-09 | +| 25 项端到端冒烟（子进程级，`tools/smoke_e2e.py`，含 SSE 转换/透传逐块到达实测 ~31ms + 未实现方向 501） | ✅ 全过 | 2026-09-09 | | 多轮 previous_response_id 链路（0→2→4 条重放） | ✅ 修复后通过 | 2026-09-02 | | Groq 真实链路（Chat 直通 + Responses→Chat + usage 归一 + 埋点） | ✅ | 2026-09-01 | | 实验框架 dry-run | ✅ | 2026-09-02 | diff --git a/docs/capability-matrix.md b/docs/capability-matrix.md index 7f6535b..7b67f1f 100644 --- a/docs/capability-matrix.md +++ b/docs/capability-matrix.md @@ -48,6 +48,6 @@ | usage 归一 | ✅ | ✅ | ✅ | 离线已测 + Groq 真实链路验证 | | previous_response_id | — | ✅（状态层重放） | — | **已实现**：SQLite 会话表 + TTL + 多轮链 E2E 测试（0→2→4 重放累积） | | 缓存断点布局 | ✅ | ✅ | — | **已落地**：3 固定 + 1 滚动，exp7/E24 实测 | -| SSE 流式 | ◐ | ◐ | ◐ | **v1.6 部分逐块**：chat 客户端 ← anthropic 上游已逐块转换（`src/gateway/sse.py`，冒烟实测 30.7ms 逐块到达）；其余方向整读透传 → LIMITATIONS #1（E21 直连实验证明流式无损缓存命中率） | +| SSE 流式 | ◐ | ◐ | ◐ | **v1.7 三方向逐块**：chat←anthropic 逐块转换 + chat←chat / anthropic←anthropic 透传（冒烟实测 ~31ms 逐块到达）；response 源等未实现方向显式 501 → LIMITATIONS #1（E21 直连实验证明流式无损缓存命中率） | | max_tokens:0 预热 | — | — | ✅ | 字段处理已测 + 实测预热后 100% 命中 | | 多模态 / Realtime / Batches | ❌ | ❌ | ❌ | 方案 3.8 明确不做 | diff --git a/src/gateway/server.py b/src/gateway/server.py index 6cf58de..6822800 100644 --- a/src/gateway/server.py +++ b/src/gateway/server.py @@ -338,12 +338,25 @@ def do_POST(self): return self._json(400, {"error": {"message": werr, "type": "invalid_request_error"}}) - # SSE 逐块流式（LIMITATIONS #1 最小闭环）：仅 chat 客户端 ← anthropic 上游。 - # 其余方向保持整读透传。warmup+stream 已在上方被 validate_warmup 拦截， - # 走到这里的一定不是预热轮。 - if out.get("stream") and source == "openai_chat" and target == "anthropic": - return self._stream_from_anthropic(out, req, dropped, session, - replayed_hist) + # SSE 逐块流式（LIMITATIONS #1）： + # chat 客户端 ← anthropic 上游 —— 逐块转换（v1.6） + # 同协议直通（chat←chat / anthropic←anthropic）—— 字节透传 + 旁路收集（v1.7） + # 其余方向（response 源三维寻址 / anthropic←chat 等）未实现： + # 显式 501——否则 post_json 会把 SSE 当 JSON 解析，炸成语焉不详的 502。 + # warmup+stream 已在上方被 validate_warmup 拦截，走到这里的一定不是预热轮。 + if out.get("stream"): + if source == "openai_chat" and target == "anthropic": + return self._stream_from_anthropic(out, req, dropped, session, + replayed_hist) + if source == target and source in ("openai_chat", "anthropic"): + return self._stream_passthrough(out, req, dropped, session, + replayed_hist, target) + return self._json(501, {"error": { + "message": f"暂不支持 {source} -> {target} 方向的流式；请改用 stream:false", + "type": "invalid_request_error", + "supported_streaming": ["openai_chat->anthropic", + "openai_chat->openai_chat", + "anthropic->anthropic"]}}) # mock 与真实后端走同一套端点映射。tools/mock_backend.py 已按端点路径 # 返回对应协议形状，不再为 mock 特判——否则离线跑的和上线跑的不是同一条路径， @@ -428,6 +441,61 @@ def _stream_from_anthropic(self, out: dict, req, dropped, session, finalize_turn("anthropic", conv.synthetic_response(), req, dropped, session, replayed_hist, warmup=False) + def _stream_passthrough(self, out: dict, req, dropped, session, + replayed_hist: list, target: str): + """同协议直通流式（v1.7）：字节原样透传，旁路收集流尾汇总。 + + 与转换流式的差别：转发的是上游**原始行**（一个比特都不动）， + 收集器只负责把 usage / 文本攒下来供 finalize_turn 复用。 + """ + from .sse import (AnthropicToChatStream, ChatStreamCollector, + parse_sse_lines, tee_lines) + + path = BACKEND_PATH.get(target, self.path) + with _GATE: + try: + resp = open_stream(BACKEND_URL + path, out, target) + except urllib.error.HTTPError as e: + detail = e.read().decode("utf-8", "replace")[:500] + return self._json(int(e.code), { + "error": {"message": "backend error", "type": "upstream_error", + "status": int(e.code), "detail": detail}}) + except Exception as e: + return self._json(502, { + "error": {"message": f"backend unreachable: {e}", + "type": "upstream_error"}}) + + self.send_response(200) + self.send_header("Content-Type", "text/event-stream; charset=utf-8") + self.send_header("Cache-Control", "no-cache") + self.send_header("Connection", "close") + self.end_headers() + self.close_connection = True + + collector = (AnthropicToChatStream() if target == "anthropic" + else ChatStreamCollector()) + try: + if target == "anthropic": + # 复用转换器做汇总（其产出的 Chat 帧丢弃，只要 usage/文本） + for event, data in parse_sse_lines( + tee_lines(resp, self.wfile)): + collector.feed(event, data) + else: # openai_chat + for raw in tee_lines(resp, self.wfile): + line = (raw.decode("utf-8", "replace") + if isinstance(raw, (bytes, bytearray)) else raw) + line = line.strip() + if line.startswith("data:"): + collector.feed_data(line[len("data:"):]) + except (ConnectionError, BrokenPipeError, OSError): + # 客户端中途断开：用已收到的部分照常落账（费用已实际发生） + pass + finally: + resp.close() + + finalize_turn(target, collector.synthetic_response(), req, dropped, + session, replayed_hist, warmup=False) + def _usage_for(target: str, resp: dict) -> ir.IRUsage: u = resp.get("usage", {}) or {} diff --git a/src/gateway/sse.py b/src/gateway/sse.py index 48eb378..537c539 100644 --- a/src/gateway/sse.py +++ b/src/gateway/sse.py @@ -170,3 +170,68 @@ def synthetic_response(self) -> dict: content.append({"type": "tool_use", "id": tb["id"], "name": tb["name"], "input": tool_input}) return {"content": content, "usage": self.usage()} + + +def tee_lines(line_iter, wfile): + """透传三通：每行原样写给客户端（即时 flush），同时 yield 给旁路收集器。 + + 直通流式（同协议）不做任何转换——转发的是上游原始字节， + 旁路只负责把 usage / 文本攒下来供流尾埋点与历史落库。 + """ + for line in line_iter: + wfile.write(line if isinstance(line, (bytes, bytearray)) else line.encode("utf-8")) + wfile.flush() + yield line + + +class ChatStreamCollector: + """chat←chat 直通流式的流尾收集器（不转换，只汇总）。 + + Chat 流式响应默认不带 usage（除非客户端传 stream_options.include_usage）， + 收不到就是 0——如实记录，不编造。 + """ + + def __init__(self) -> None: + self._text_parts: list[str] = [] + self._usage: dict = {} + # index -> {"id","name","args":[str]}（Chat 流的工具参数也是分段 delta） + self._tool_slots: dict[int, dict] = {} + + def feed_data(self, data_str: str) -> None: + """喂一行 SSE data 内容（已去掉 'data:' 前缀）。""" + data_str = data_str.strip() + if not data_str or data_str == "[DONE]": + return + try: + d = json.loads(data_str) + except ValueError: + return + if isinstance(d.get("usage"), dict): + self._usage = d["usage"] + for ch in d.get("choices", []) or []: + delta = ch.get("delta", {}) or {} + if delta.get("content"): + self._text_parts.append(delta["content"]) + for tc in delta.get("tool_calls", []) or []: + i = tc.get("index", 0) + slot = self._tool_slots.setdefault(i, {"id": "", "name": "", "args": []}) + if tc.get("id"): + slot["id"] = tc["id"] + fn = tc.get("function", {}) or {} + if fn.get("name"): + slot["name"] = fn["name"] + if fn.get("arguments"): + slot["args"].append(fn["arguments"]) + + def usage(self) -> dict: + return dict(self._usage) + + def synthetic_response(self) -> dict: + """拼成 Chat 非流式响应形状，复用 `_usage_for` / `assistant_from_upstream`。""" + msg: dict = {"role": "assistant", "content": "".join(self._text_parts)} + if self._tool_slots: + msg["tool_calls"] = [ + {"id": s["id"], "type": "function", + "function": {"name": s["name"], "arguments": "".join(s["args"])}} + for _, s in sorted(self._tool_slots.items())] + return {"choices": [{"index": 0, "message": msg}], "usage": self._usage} diff --git a/tests/test_offline.py b/tests/test_offline.py index a49dcae..b06ee09 100644 --- a/tests/test_offline.py +++ b/tests/test_offline.py @@ -19,7 +19,8 @@ from src.adapters.chat import ChatAdapter, usage_from_chat from src.adapters.anthropic import AnthropicAdapter, usage_from_anthropic from src.adapters.response import ResponseAdapter -from src.gateway.sse import AnthropicToChatStream, parse_sse_lines +from src.gateway.sse import (AnthropicToChatStream, ChatStreamCollector, + parse_sse_lines) from src.ir import model as ir from src.state.session_config import (SessionConfig, build_prefix, load, FULL, LAST_BREAKPOINT, SLIDING_WINDOW) @@ -976,5 +977,50 @@ def test_upstream_error_frame(self): chunk["choices"][0]["delta"].get("content", "")) +class TestChatStreamCollector(unittest.TestCase): + """v1.7 直通流式：chat←chat 旁路收集器（不转换，只汇总）。""" + + def test_text_and_usage_collected(self): + c = ChatStreamCollector() + c.feed_data(json.dumps({"choices": [{"delta": {"role": "assistant", + "content": "你"}}]})) + c.feed_data(json.dumps({"choices": [{"delta": {"content": "好"}}]})) + c.feed_data(json.dumps({"choices": [{"delta": {}, + "finish_reason": "stop"}], + "usage": {"prompt_tokens": 100, + "completion_tokens": 7}})) + c.feed_data("[DONE]") + self.assertEqual(c.usage()["prompt_tokens"], 100) + synth = c.synthetic_response() + self.assertEqual(synth["choices"][0]["message"]["content"], "你好") + # 复用既有落库/归一路径 + reply = assistant_from_upstream("openai_chat", synth) + self.assertEqual(reply.blocks[0].text, "你好") + u = usage_from_chat(synth["usage"]) + self.assertEqual(u.output_tokens, 7) + + def test_tool_call_slots_merged(self): + """Chat 流的工具参数同样分段到达：按 index 攒槽、拼完整。""" + c = ChatStreamCollector() + c.feed_data(json.dumps({"choices": [{"delta": {"tool_calls": [ + {"index": 0, "id": "call_1", "type": "function", + "function": {"name": "get_weather", "arguments": "{\"ci"}}]}}]})) + c.feed_data(json.dumps({"choices": [{"delta": {"tool_calls": [ + {"index": 0, "function": {"arguments": "ty\":\"天津\"}"}}]}}]})) + synth = c.synthetic_response() + tc = synth["choices"][0]["message"]["tool_calls"][0] + self.assertEqual(tc["id"], "call_1") + self.assertEqual(tc["function"]["arguments"], '{"city":"天津"}') + + def test_garbage_and_empty_tolerated(self): + c = ChatStreamCollector() + c.feed_data("") # 空行 + c.feed_data("非JSON") # 容错跳过 + c.feed_data("[DONE]") + self.assertEqual(c.usage(), {}) + synth = c.synthetic_response() + self.assertEqual(synth["choices"][0]["message"]["content"], "") + + if __name__ == "__main__": unittest.main() diff --git a/tools/mock_backend.py b/tools/mock_backend.py index d377a36..1b7666a 100644 --- a/tools/mock_backend.py +++ b/tools/mock_backend.py @@ -190,6 +190,27 @@ def sse_lines(payload: dict, path: str): yield "event: message_stop\n" yield 'data: {"type":"message_stop"}\n\n' return + if "chat/completions" in path: + _, read = _cache_tick(payload) + chunks = ["mock ", "stream ", "reply"] + for i, c in enumerate(chunks): + delta = {"content": c} + if i == 0: + delta = {"role": "assistant", "content": c} + frame = {"id": "chatcmpl_mock_stream", "object": "chat.completion.chunk", + "created": int(time.time()), "model": payload.get("model", "mock"), + "choices": [{"index": 0, "delta": delta, "finish_reason": None}]} + yield f"data: {json.dumps(frame, ensure_ascii=False)}\n\n" + time.sleep(0.01) + fin = {"id": "chatcmpl_mock_stream", "object": "chat.completion.chunk", + "created": int(time.time()), "model": payload.get("model", "mock"), + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": _input_chars(payload), + "completion_tokens": 12, + "prompt_tokens_details": {"cached_tokens": read}}} + yield f"data: {json.dumps(fin, ensure_ascii=False)}\n\n" + yield "data: [DONE]\n\n" + return for c in ["mock ", "stream ", "reply"]: yield f"data: {json.dumps({'type': 'content_block_delta', 'delta': {'type': 'text_delta', 'text': c}})}\n\n" time.sleep(0.01) diff --git a/tools/smoke_e2e.py b/tools/smoke_e2e.py index 7b34bcb..6611e0a 100644 --- a/tools/smoke_e2e.py +++ b/tools/smoke_e2e.py @@ -207,6 +207,54 @@ def main() -> int: json.dumps(last, ensure_ascii=False)[:120]) except Exception as e: check("流式轮照常进埋点", False, f"{type(e).__name__}: {e}") + + print("\n7) SSE 直通流式（同协议透传，v1.7）+ 未实现方向显式 501") + # 7a anthropic -> anthropic：字节透传，客户端收到原样 Anthropic 事件 + fa, ta = post_stream("/v1/anthropic/to/anthropic", + {"model": "mock", "stream": True, "max_tokens": 64, + "messages": [{"role": "user", "content": "hi"}]}) + deltas = [] + for f in fa: + try: + d = json.loads(f) + except ValueError: + continue + if d.get("type") == "content_block_delta": + deltas.append(d.get("delta", {}).get("text", "")) + check("anthropic 透传文本拼合完整", + "".join(deltas) == "mock stream reply", repr("".join(deltas))) + check("anthropic 透传收到 message_stop", + any("message_stop" in f for f in fa)) + spread_a = (ta[-1] - ta[0]) if len(ta) >= 2 else 0 + check("anthropic 透传逐块到达", spread_a >= 0.015, + f"首尾间隔 {spread_a*1000:.1f}ms") + # 7b chat -> chat：字节透传，客户端收到原样 Chat chunk + fb, tb = post_stream("/v1/openai_chat/to/openai_chat", + {"model": "mock", "stream": True, + "messages": [{"role": "user", "content": "hi"}]}) + chunks_b = [] + for f in fb: + if f == "[DONE]": + continue + try: + chunks_b.append(json.loads(f)) + except ValueError: + pass + text_b = "".join(c["choices"][0]["delta"].get("content", "") + for c in chunks_b) + check("chat 透传文本拼合完整", text_b == "mock stream reply", repr(text_b)) + check("chat 透传以 [DONE] 收尾", bool(fb) and fb[-1] == "[DONE]") + spread_b = (tb[-1] - tb[0]) if len(tb) >= 2 else 0 + check("chat 透传逐块到达", spread_b >= 0.015, + f"首尾间隔 {spread_b*1000:.1f}ms") + # 7c 未实现的流式方向：显式 501（此前是 post_json 把 SSE 当 JSON 解析炸成 502） + try: + post("/v1/openai_response/to/anthropic", + {"model": "mock", "stream": True, + "input": [{"type": "message", "role": "user", "content": "hi"}]}) + check("未实现流式方向显式 501", False, "请求通过了") + except urllib.error.HTTPError as e: + check("未实现流式方向显式 501", e.code == 501, f"HTTP {e.code}") finally: stop(tmp)