diff --git a/CHANGELOG.md b/CHANGELOG.md index 86156e6..2b2852d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -59,3 +59,12 @@ - **实验原始数据入库**：`data/experiments/` 19 个文件（34 组实测逐轮 jsonl + 汇总 csv）+ `data/README.md` 溯源表——实验文档每个数字可回查到具体文件具体行；`.gitignore` 改为只忽略运行时产物、放行精选数据目录 - LIMITATIONS #12 补标已闭环（property-based + lossy-fields 已于 PR #8 落地，原标记滞后） - README：目录补 data/、已知限制行标注 3 项已闭环、单测数 40 → 46；REVIEW.md 测试口径统一为 46 项 + +## 2026-09-09 · v1.6 SSE 逐块流式（LIMITATIONS #1 部分闭环，本提交） + +- **`src/gateway/sse.py`**：`openai_chat` 客户端 ← `anthropic` 上游方向的逐块流式转换。text_delta 即时翻成 Chat chunk 下发（首 token 延迟与直连一致）；`input_json_delta` 按既定折损缓冲到 `content_block_stop` 一次性发（非完整 JSON 片段无法流式透传）；usage 在 message_start/message_delta 分次到达、流尾合并照常进 5 项埋点；`synthetic_response()` 复用 `assistant_from_upstream`——流式与非流式落库/埋点同一条代码路径，不另造语义 +- 网关：抽 `_backend_headers` / `_backend_opener` / `open_stream`，do_POST 尾部逻辑收敛为 `finalize_turn()`（非流式与流式共用）；流式请求同样占 `_GATE` 并发名额 +- 设计验证：urllib 响应对象按行增量吐数据（实测逐行到达），无需引入第三方 SSE 库，保持零依赖 +- 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 diff --git a/README.md b/README.md index 99fd52c..f502204 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 流式 | ◐ 整读透传 | 逐块转换未实现 → `docs/LIMITATIONS.md` #1；E21 证明流式无损命中率 | +| SSE 流式 | ◐ **v1.6 部分逐块**（chat 客户端 ← anthropic 上游） | `src/gateway/sse.py` 逐块转换（冒烟实测 30.7ms 逐块到达）；其余方向整读透传 → `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 # 46 项离线单测：IR 往返 / adapter / 状态层 / 预热 / 三轮自查回归 / 多轮链 E2E / ID 映射 / property-based / 惰性淘汰 -python tools/smoke_e2e.py # 11 项端到端冒烟：真起 mock+网关子进程，5 组链路断言 +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 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 为整读透传，逐块转换见 docs/LIMITATIONS.md #1） + gateway/ HTTP 转发 + SSE 逐块转换（chat←anthropic 已实现 sse.py，其余方向整读透传见 docs/LIMITATIONS.md #1） observability/ 命中率埋点（5 项暴露） warmup/ max_tokens:0 预热 webui/ 弹网页本地仪表盘（安全骨架 + 记忆编排界面） diff --git a/docs/LIMITATIONS.md b/docs/LIMITATIONS.md index 2d45f9d..894cf00 100644 --- a/docs/LIMITATIONS.md +++ b/docs/LIMITATIONS.md @@ -6,7 +6,9 @@ ## 一、协议转换链路 -### #1 SSE 逐块流式转换未实现（整读透传） +### #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 仍如实记录）。以下原文留档，未闭环范围以四要素为准。 - **影响半径**：流式体验与首 token 延迟（TTFB）。功能正确性不受影响——响应内容完整、usage 归一与埋点照常工作。 - **触发条件**：客户端请求 `stream:true` 经过网关。当前网关 `post_json` 整读响应后一次性返回；跨协议的事件模型转换（Chat `delta` / Responses 三维寻址 / Anthropic 块生命周期）未实现。 diff --git a/docs/REVIEW.md b/docs/REVIEW.md index 635aa3b..025353a 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+） -> 测试：**46 项单测全过 + 11 项端到端冒烟全过**（2026-09-09，含第三轮自查回归 + property-based 性质测试 + 惰性淘汰回归）+ Groq 真实链路验证通过（2026-09-01） +> 测试：**53 项单测全过 + 18 项端到端冒烟全过**（2026-09-09，含第三轮自查回归 + property-based 性质测试 + 惰性淘汰回归 + SSE 逐块转换）+ Groq 真实链路验证通过（2026-09-01） > 本版差异：**两轮代码自查发现 9 个真实 bug，全部修复并各配回归测试**（见第六节） ## 一、整体架构（IR 星型） @@ -66,6 +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` 落库 | **网关关键函数**： @@ -110,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` | 164 | **v1.2 新增**：端到端冒烟——真起 mock + 网关两个子进程，跑 5 组 11 项断言（直通/跨协议/多轮链/预热/非法路由），退出码 0/1 | +| `smoke_e2e.py` | 220 | 端到端冒烟——真起 mock + 网关两个子进程，跑 6 组 18 项断言（直通/跨协议/多轮链/预热/非法路由/**SSE 逐块流式**），退出码 0/1 | -### tests/ —— 测试（46 项全过） +### tests/ —— 测试（53 项全过） | 测试类 | 覆盖 | | ------ | ---- | @@ -128,13 +129,14 @@ do_POST（路由 /v1/{source}/to/{target}，未知协议 → 400） | `TestToolIdMap` | **v1.4 新增**：工具 ID 双向映射（含 Anthropic→Chat→Anthropic 还原、并发分叉不串号） | | `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 流尾合并 + 落库复用、上游错误帧 | ## 四、已验证记录 | 验证 | 结果 | 日期 | | ---- | ---- | ---- | -| 46 项单测（`-W error::ResourceWarning` 下零警告，含第三轮自查回归 + property-based + 惰性淘汰） | ✅ 全过 | 2026-09-09 | -| 11 项端到端冒烟（子进程级，`tools/smoke_e2e.py`） | ✅ 全过 | 2026-09-02 | +| 53 项单测（`-W error::ResourceWarning` 下零警告，含第三轮自查回归 + property-based + 惰性淘汰 + SSE 转换） | ✅ 全过 | 2026-09-09 | +| 18 项端到端冒烟（子进程级，`tools/smoke_e2e.py`，含 SSE 逐块到达实测 30.7ms） | ✅ 全过 | 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 530a8e9..7f6535b 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 流式 | ◐ | ◐ | ◐ | **整读透传**：网关整读响应再返回，逐块流式转换未实现 → LIMITATIONS #1（E21 直连实验证明流式无损缓存命中率） | +| SSE 流式 | ◐ | ◐ | ◐ | **v1.6 部分逐块**：chat 客户端 ← anthropic 上游已逐块转换（`src/gateway/sse.py`，冒烟实测 30.7ms 逐块到达）；其余方向整读透传 → LIMITATIONS #1（E21 直连实验证明流式无损缓存命中率） | | max_tokens:0 预热 | — | — | ✅ | 字段处理已测 + 实测预热后 100% 命中 | | 多模态 / Realtime / Batches | ❌ | ❌ | ❌ | 方案 3.8 明确不做 | diff --git a/src/gateway/server.py b/src/gateway/server.py index 56e8063..6cf58de 100644 --- a/src/gateway/server.py +++ b/src/gateway/server.py @@ -196,8 +196,7 @@ def convert(source: str, target: str, payload: dict, headers: dict): return out, req, dropped, session, injected, replayed -def post_json(url: str, payload: dict, target: str = "") -> dict: - data = json.dumps(payload).encode("utf-8") +def _backend_headers(target: str = "") -> dict: headers = { "Content-Type": "application/json", # 部分真实网关（如 Cloudflare 前置）按 UA 拦 bot，带浏览器 UA 更稳 @@ -212,14 +211,79 @@ def post_json(url: str, payload: dict, target: str = "") -> dict: # Anthropic 原生端点常校验该头 headers["anthropic-version"] = os.environ.get("PB_ANTHROPIC_VERSION", "2023-06-01") - req = urllib.request.Request(url, data=data, headers=headers) + return headers + + +def _backend_opener(): # PB_DIRECT=1 时绕过系统代理（本地代理隧道对长 POST 可能 502） - opener = (urllib.request.build_opener(urllib.request.ProxyHandler({})) - if DIRECT else urllib.request.build_opener()) - with opener.open(req, timeout=60) as resp: + return (urllib.request.build_opener(urllib.request.ProxyHandler({})) + if DIRECT else urllib.request.build_opener()) + + +def post_json(url: str, payload: dict, target: str = "") -> dict: + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request(url, data=data, headers=_backend_headers(target)) + with _backend_opener().open(req, timeout=60) as resp: return json.loads(resp.read().decode("utf-8")) +def open_stream(url: str, payload: dict, target: str = ""): + """以流式方式打开上游连接，返回可**逐行迭代**的响应对象（调用方负责关闭）。 + + urllib 响应对象本身就是行迭代器，且行随上游到达即时吐出（已实测验证）， + 这是逐块流式转发的实现基础。 + """ + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request(url, data=data, headers=_backend_headers(target)) + return _backend_opener().open(req, timeout=60) + + +def finalize_turn(target: str, backend_resp: dict, req, dropped, session, + replayed_hist: list, warmup: bool) -> str | None: + """流尾共用逻辑（非流式响应与 SSE 流式同一条路径）： + usage 归一 + 命中率埋点 + 本轮历史落库 + 登记响应 id。 + + 流式场景传入的 backend_resp 是 sse.AnthropicToChatStream.synthetic_response() + 拼出的 Anthropic 非流式形状——两条路径语义一致，不另造埋点口径。 + """ + # 归一 usage + 命中率埋点（真实字段由真实 API 返回；mock 时为占位值） + usage = _usage_for(target, backend_resp) + # 重放代价只算真正重放的历史切片，不含本轮新消息 + replayed_chars = sum(len(b.text or "") + for m in replayed_hist for b in m.blocks) + # 按 extra 标记统计（而非按尾部切片），注入块位置变化时依然准确 + injected_chars = sum(len(b.text or "") for b in req.system + if b.extra.get("injected_memory")) + METRICS.record_turn(usage, injected=injected_chars, replayed=replayed_chars, + dropped=dropped.items, + degradation="explicit" if dropped else "none", + kind=WARMUP if warmup else NORMAL) + + # 有状态：登记响应供下一轮 previous_response_id 引用。 + # 预热轮不登记——它不产出可引用的对话轮，登记会污染响应链。 + rid = None + if session is not None and not warmup: + # 本轮对话必须落进会话历史，否则下一轮 previous_response_id 重放出的是空列表， + # 整个状态层等于空转（历史永远是空 -> 重放永远是空 -> 多轮链路第一轮就断）。 + # 只追加本轮新消息：重放切片已在历史里，再加一次会自我复制。 + new_turn = list(req.messages[len(replayed_hist):]) + reply = assistant_from_upstream(target, backend_resp) + if reply is not None: + new_turn.append(reply) + if new_turn: + # 落库前把 ID 翻回 canonical：new_turn 里的 ID 刚被 outgoing + # 翻成目标协议形式，reply 里的是上游新生成的——历史必须存 + # canonical，否则跨协议续轮时映射链会退化。 + idm = STORE.idmap + for m in new_turn: + for b in m.blocks: + if b.kind in (ir.TOOL_USE, ir.TOOL_RESULT) and b.tool_id: + b.tool_id = idm.incoming(session.key, b.tool_id, target) + STORE.append(session, new_turn) + rid = STORE.record_response(session) + return rid + + class Handler(BaseHTTPRequestHandler): def log_message(self, *a): # 静音 pass @@ -274,6 +338,13 @@ 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) + # mock 与真实后端走同一套端点映射。tools/mock_backend.py 已按端点路径 # 返回对应协议形状，不再为 mock 特判——否则离线跑的和上线跑的不是同一条路径， # 上线时才暴露的问题离线永远测不出来。 @@ -292,41 +363,8 @@ def do_POST(self): "error": {"message": f"backend unreachable: {e}", "type": "upstream_error"}}) - # 归一 usage + 命中率埋点（真实字段由真实 API 返回；mock 时为占位值） - usage = _usage_for(target, backend_resp) - # 重放代价只算真正重放的历史切片，不含本轮新消息 - replayed_chars = sum(len(b.text or "") - for m in replayed_hist for b in m.blocks) - # 按 extra 标记统计（而非按尾部切片），注入块位置变化时依然准确 - injected_chars = sum(len(b.text or "") for b in req.system - if b.extra.get("injected_memory")) - METRICS.record_turn(usage, injected=injected_chars, replayed=replayed_chars, - dropped=dropped.items, - degradation="explicit" if dropped else "none", - kind=WARMUP if warmup else NORMAL) - - # 有状态：登记响应供下一轮 previous_response_id 引用。 - # 预热轮不登记——它不产出可引用的对话轮，登记会污染响应链。 - rid = None - if session is not None and not warmup: - # 本轮对话必须落进会话历史，否则下一轮 previous_response_id 重放出的是空列表， - # 整个状态层等于空转（历史永远是空 -> 重放永远是空 -> 多轮链路第一轮就断）。 - # 只追加本轮新消息：重放切片已在历史里，再加一次会自我复制。 - new_turn = list(req.messages[len(replayed_hist):]) - reply = assistant_from_upstream(target, backend_resp) - if reply is not None: - new_turn.append(reply) - if new_turn: - # 落库前把 ID 翻回 canonical：new_turn 里的 ID 刚被 outgoing - # 翻成目标协议形式，reply 里的是上游新生成的——历史必须存 - # canonical，否则跨协议续轮时映射链会退化。 - idm = STORE.idmap - for m in new_turn: - for b in m.blocks: - if b.kind in (ir.TOOL_USE, ir.TOOL_RESULT) and b.tool_id: - b.tool_id = idm.incoming(session.key, b.tool_id, target) - STORE.append(session, new_turn) - rid = STORE.record_response(session) + rid = finalize_turn(target, backend_resp, req, dropped, session, + replayed_hist, warmup) body = dict(backend_resp) # 客户端说 Responses 协议时，下一轮会带 previous_response_id 回来。 @@ -342,6 +380,54 @@ def do_POST(self): "replayed_messages": len(replayed_hist)} self._json(200, body) + def _stream_from_anthropic(self, out: dict, req, dropped, session, + replayed_hist: list): + """SSE 逐块流式：Anthropic 上游事件 → Chat chunk 帧即时下发。 + + 与非流式的差别只在传输段：转换（to_ir/from_ir）、限流、usage 归一、 + 埋点、历史落库全部复用同一条路径（finalize_turn + synthetic_response）。 + """ + from .sse import AnthropicToChatStream, parse_sse_lines + + path = BACKEND_PATH["anthropic"] + with _GATE: # 限流：流式请求同样占并发名额 + try: + resp = open_stream(BACKEND_URL + path, out, "anthropic") + 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 + + conv = AnthropicToChatStream() + try: + for event, data in parse_sse_lines(resp): + for frame in conv.feed(event, data): + self.wfile.write(frame.encode("utf-8")) + self.wfile.flush() + except (ConnectionError, BrokenPipeError, OSError): + # 客户端中途断开：流尾落库/埋点照常（这轮 API 费用已实际发生） + pass + finally: + resp.close() + + # 流尾：与非流式同一套 usage 归一 + 埋点 + 历史落库。 + # dropped 清单对流式客户端不可见（SSE 帧里没有它的位置）—— + # 已写进 LIMITATIONS #1 的范围说明，埋点里 degradation 仍如实记录。 + finalize_turn("anthropic", conv.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 new file mode 100644 index 0000000..48eb378 --- /dev/null +++ b/src/gateway/sse.py @@ -0,0 +1,172 @@ +"""SSE 逐块流式转换（LIMITATIONS #1 最小闭环：openai_chat 客户端 ← Anthropic 上游）。 + +只有 source=openai_chat / target=anthropic 且 stream:true 的请求走这里， +其余方向保持整读透传（见 docs/LIMITATIONS.md #1 的范围说明）。 + +转换语义： +- Anthropic 事件按行增量解析（空行分帧），text_delta 立即翻成 Chat chunk 下发—— + 首 token 延迟与直连上游一致，这是本模块存在的意义； +- 工具参数（input_json_delta）是非完整 JSON 片段，缓冲到 content_block_stop + 一次性发出——长工具调用退化为非流式，是无法回避的折损（LIMITATIONS 已声明）； +- thinking_delta / signature_delta 在 Chat 协议里无对应概念，不落帧（客户端 + 只能看到正式文本——这是降级，不是丢失：thinking 本就不该出现在 Chat 响应里）； +- usage 在 message_start（input 侧 + 缓存字段）与 message_delta（output 侧） + 分次到达，流尾合并后照常进 5 项埋点——命中率北极星不因流式缺数据。 +""" +from __future__ import annotations + +import json +import time + +# Anthropic stop_reason → Chat finish_reason +_STOP_MAP = {"end_turn": "stop", "stop_sequence": "stop", "max_tokens": "length", + "tool_use": "tool_calls", "refusal": "stop"} + + +def parse_sse_lines(line_iter): + """把 SSE 行流解析成 (event, data) 事件对，逐事件 yield。 + + line_iter 是任何按行迭代的字节/字符串流（urllib 响应对象即可）。 + data 按 SSE 规范可多行拼接；能解析成 JSON 就给 dict，否则给原字符串。 + 流尾残留的半帧容错丢弃（正常结束的上游最后一定是 message_stop 完整帧）。 + """ + event, data_lines = None, [] + for raw in line_iter: + line = raw.decode("utf-8", "replace") if isinstance(raw, (bytes, bytearray)) else raw + line = line.rstrip("\r\n") + if not line: + if data_lines: + data = "\n".join(data_lines) + try: + data = json.loads(data) + except ValueError: + pass + yield event, data + event, data_lines = None, [] + continue + if line.startswith(":"): + continue # 注释 / 心跳行 + field, _, value = line.partition(":") + value = value.lstrip(" ") + if field == "event": + event = value + elif field == "data": + data_lines.append(value) + + +class AnthropicToChatStream: + """Anthropic SSE 事件 → Chat completion.chunk 帧（逐块、可即时下发）。 + + 用法：每个上游事件 feed() 一次，把返回的帧立即写给客户端； + 流结束后用 usage() / synthetic_response() 做埋点与历史落库。 + """ + + def __init__(self) -> None: + self.msg_id = "chatcmpl_stream" + self.model = "" + self.created = int(time.time()) + self._usage_in: dict = {} + self._output_tokens = 0 + self._text_parts: list[str] = [] + # index -> {"id","name","args":[str]}；input_json_delta 逐段累积 + self._tool_open: dict[int, dict] = {} + self._tools_done: list[dict] = [] + self._sent_role = False + + # -- 帧构造 ------------------------------------------------------------ + def _chunk(self, delta: dict, finish: str | None = None) -> str: + frame = {"id": self.msg_id, "object": "chat.completion.chunk", + "created": self.created, "model": self.model, + "choices": [{"index": 0, "delta": delta, "finish_reason": finish}]} + return "data: " + json.dumps(frame, ensure_ascii=False) + "\n\n" + + def _role_frame(self) -> str: + self._sent_role = True + return self._chunk({"role": "assistant"}) + + # -- 事件入口 ----------------------------------------------------------- + def feed(self, event: str | None, data) -> list[str]: + """喂一个 Anthropic 事件，返回要立即下发客户端的 SSE 帧（可为空列表）。""" + if not isinstance(data, dict): + return [] + etype = data.get("type", event or "") + out: list[str] = [] + if etype == "message_start": + msg = data.get("message", {}) or {} + self.msg_id = msg.get("id", self.msg_id) + self.model = msg.get("model", self.model) + self._usage_in = msg.get("usage", {}) or {} + out.append(self._role_frame()) + elif etype == "content_block_start": + idx = data.get("index", 0) + block = data.get("content_block", {}) or {} + if block.get("type") == "tool_use": + self._tool_open[idx] = {"id": block.get("id", ""), + "name": block.get("name", ""), "args": []} + elif etype == "content_block_delta": + idx = data.get("index", 0) + delta = data.get("delta", {}) or {} + dt = delta.get("type") + if dt == "text_delta": + text = delta.get("text", "") + self._text_parts.append(text) + if not self._sent_role: + out.append(self._role_frame()) + out.append(self._chunk({"content": text})) + elif dt == "input_json_delta": + tb = self._tool_open.get(idx) + if tb is not None: + tb["args"].append(delta.get("partial_json", "")) + # thinking_delta / signature_delta：Chat 无对应概念，不落帧（见模块 docstring） + elif etype == "content_block_stop": + idx = data.get("index", 0) + tb = self._tool_open.pop(idx, None) + if tb is not None: + self._tools_done.append(tb) + # 工具参数缓冲到块结束一次性发出（input_json_delta 非完整 JSON） + out.append(self._chunk({"tool_calls": [{ + "index": 0, "id": tb["id"], "type": "function", + "function": {"name": tb["name"], + "arguments": "".join(tb["args"])}}]})) + elif etype == "message_delta": + delta = data.get("delta", {}) or {} + usage = data.get("usage", {}) or {} + self._output_tokens = usage.get("output_tokens", self._output_tokens) + sr = delta.get("stop_reason") + if sr: + out.append(self._chunk({}, _STOP_MAP.get(sr, "stop"))) + elif etype == "message_stop": + out.append("data: [DONE]\n\n") + elif etype == "error": + err = data.get("error", {}) or {} + out.append(self._chunk( + {"content": f"[上游错误] {err.get('message', 'unknown')}"})) + # ping 等控制事件：不落帧 + return out + + # -- 流尾汇总（埋点 + 历史落库的原料） ------------------------------------ + def usage(self) -> dict: + """合并 message_start / message_delta 的 usage（保持 Anthropic 口径）。""" + u = dict(self._usage_in) + u["output_tokens"] = self._output_tokens + return u + + def synthetic_response(self) -> dict: + """把累计内容拼成 Anthropic 非流式响应形状——直接复用 + `assistant_from_upstream("anthropic", ...)` 与 `_usage_for`， + 流式与非流式的落库/埋点走同一条代码路径，不另造语义。""" + content: list[dict] = [] + text = "".join(self._text_parts) + if text: + content.append({"type": "text", "text": text}) + done = list(self._tools_done) + # 容错：流被截断时未正常 content_block_stop 的工具块也入史 + done += [self._tool_open[i] for i in sorted(self._tool_open)] + for tb in done: + try: + tool_input = json.loads("".join(tb["args"]) or "{}") + except ValueError: + tool_input = {} + content.append({"type": "tool_use", "id": tb["id"], + "name": tb["name"], "input": tool_input}) + return {"content": content, "usage": self.usage()} diff --git a/tests/test_offline.py b/tests/test_offline.py index 5041f71..a49dcae 100644 --- a/tests/test_offline.py +++ b/tests/test_offline.py @@ -19,6 +19,7 @@ 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.ir import model as ir from src.state.session_config import (SessionConfig, build_prefix, load, FULL, LAST_BREAKPOINT, SLIDING_WINDOW) @@ -833,5 +834,147 @@ def _count_breakpoints(payload: dict) -> int: return n +class TestSseConversion(unittest.TestCase): + """LIMITATIONS #1 最小闭环：Anthropic SSE → Chat chunk 逐块转换。""" + + @staticmethod + def _feed_seq(conv, events): + """按序喂事件，把每事件产出的帧全部收集。""" + frames = [] + for ev in events: + frames.extend(conv.feed(ev.get("type"), ev)) + return frames + + @staticmethod + def _parse_frame(frame: str): + assert frame.startswith("data: ") and frame.endswith("\n\n") + return json.loads(frame[len("data:"):].strip()) + + def test_parse_sse_lines_basic(self): + raw = ( + 'event: message_start\n' + 'data: {"type":"message_start","message":{"id":"msg_1"}}\n' + '\n' + ': heartbeat-comment\n' + 'data: {"type":"ping"}\n' + '\n' + 'data: 非JSON行\n' + '\n' + ).encode("utf-8") + events = list(parse_sse_lines(iter(raw.splitlines(keepends=True)))) + self.assertEqual(len(events), 3) + self.assertEqual(events[0], ("message_start", + {"type": "message_start", + "message": {"id": "msg_1"}})) + self.assertEqual(events[1][1], {"type": "ping"}) # 注释行被跳过 + self.assertEqual(events[2][1], "非JSON行") # 非 JSON 原样透传 + + def test_parse_sse_lines_partial_tail_dropped(self): + raw = b'data: {"type":"message_stop"}\n\ndata: {"type":"truncat' + events = list(parse_sse_lines(iter(raw.splitlines(keepends=True)))) + self.assertEqual(len(events), 1) # 尾部半帧容错丢弃，不崩 + + def test_text_delta_streams_immediately(self): + conv = AnthropicToChatStream() + frames = self._feed_seq(conv, [ + {"type": "message_start", "message": { + "id": "msg_x", "model": "claude-mock", + "usage": {"input_tokens": 100, "cache_read_input_tokens": 80}}}, + {"type": "content_block_delta", "index": 0, + "delta": {"type": "text_delta", "text": "你"}}, + {"type": "content_block_delta", "index": 0, + "delta": {"type": "text_delta", "text": "好"}}, + {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, + "usage": {"output_tokens": 7}}, + {"type": "message_stop"}, + ]) + # role + 2 文本帧 + finish + DONE：文本逐块即时下发，不等流尾 + self.assertEqual(len(frames), 5) + first = self._parse_frame(frames[0]) + self.assertEqual(first["choices"][0]["delta"], {"role": "assistant"}) + self.assertEqual(first["id"], "msg_x") + self.assertEqual(first["model"], "claude-mock") + self.assertEqual(first["object"], "chat.completion.chunk") + texts = [self._parse_frame(f)["choices"][0]["delta"].get("content", "") + for f in frames[1:3]] + self.assertEqual("".join(texts), "你好") + fin = self._parse_frame(frames[3]) + self.assertEqual(fin["choices"][0]["finish_reason"], "stop") + self.assertEqual(frames[4], "data: [DONE]\n\n") + + def test_tool_args_buffered_until_block_stop(self): + """input_json_delta 是非完整 JSON 片段：缓冲到 content_block_stop 一次性发。""" + conv = AnthropicToChatStream() + f1 = conv.feed("content_block_start", + {"type": "content_block_start", "index": 1, + "content_block": {"type": "tool_use", + "id": "toolu_1", "name": "get_weather"}}) + self.assertEqual(f1, []) # 工具块开始不落帧 + f2 = conv.feed("content_block_delta", + {"type": "content_block_delta", "index": 1, + "delta": {"type": "input_json_delta", + "partial_json": '{"city": "天'}}) + f3 = conv.feed("content_block_delta", + {"type": "content_block_delta", "index": 1, + "delta": {"type": "input_json_delta", + "partial_json": '津"}'}}) + self.assertEqual(f2 + f3, []) # 参数片段期间不落帧（无法回避的折损） + f4 = conv.feed("content_block_stop", + {"type": "content_block_stop", "index": 1}) + self.assertEqual(len(f4), 1) + chunk = self._parse_frame(f4[0]) + tc = chunk["choices"][0]["delta"]["tool_calls"][0] + self.assertEqual(tc["id"], "toolu_1") + self.assertEqual(tc["function"]["name"], "get_weather") + self.assertEqual(tc["function"]["arguments"], '{"city": "天津"}') + # 落库原料：工具参数被解析回 dict + synth = conv.synthetic_response() + tool_blocks = [b for b in synth["content"] if b["type"] == "tool_use"] + self.assertEqual(tool_blocks[0]["input"], {"city": "天津"}) + + def test_stop_reason_mapping(self): + for anthropic_reason, chat_reason in ( + ("end_turn", "stop"), ("max_tokens", "length"), + ("tool_use", "tool_calls")): + conv = AnthropicToChatStream() + frames = conv.feed("message_delta", + {"type": "message_delta", + "delta": {"stop_reason": anthropic_reason}, + "usage": {"output_tokens": 3}}) + chunk = self._parse_frame(frames[0]) + self.assertEqual(chunk["choices"][0]["finish_reason"], chat_reason) + + def test_usage_merge_and_history_reuse(self): + """usage 分次到达（start 带 input+缓存、delta 带 output），流尾合并； + synthetic_response 直接复用 assistant_from_upstream 落库。""" + conv = AnthropicToChatStream() + self._feed_seq(conv, [ + {"type": "message_start", "message": { + "id": "msg_u", "model": "m", + "usage": {"input_tokens": 5978, + "cache_creation_input_tokens": 5978, + "cache_read_input_tokens": 0}}}, + {"type": "content_block_delta", "index": 0, + "delta": {"type": "text_delta", "text": "答复"}}, + {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, + "usage": {"output_tokens": 12}}, + ]) + u = usage_from_anthropic(conv.usage()) + self.assertEqual(u.input_tokens, 5978) + self.assertEqual(u.cache_creation_input_tokens, 5978) + self.assertEqual(u.output_tokens, 12) + reply = assistant_from_upstream("anthropic", conv.synthetic_response()) + self.assertIsNotNone(reply) + self.assertEqual(reply.blocks[0].text, "答复") + + def test_upstream_error_frame(self): + conv = AnthropicToChatStream() + frames = conv.feed("error", {"type": "error", "error": { + "type": "overloaded_error", "message": "Overloaded"}}) + chunk = self._parse_frame(frames[0]) + self.assertIn("Overloaded", + chunk["choices"][0]["delta"].get("content", "")) + + if __name__ == "__main__": unittest.main() diff --git a/tools/mock_backend.py b/tools/mock_backend.py index a5e6a4e..d377a36 100644 --- a/tools/mock_backend.py +++ b/tools/mock_backend.py @@ -158,9 +158,39 @@ def payload_response(payload: dict, path: str) -> dict: return anthropic_response(payload) -def sse_lines(payload: dict): - chunks = ["mock ", "stream ", "reply"] - for c in chunks: +def sse_lines(payload: dict, path: str): + """流式响应。Anthropic 路径发**完整官方事件序列**（message_start → + content_block_* → message_delta → message_stop，带 usage 与缓存字段）， + 供网关 SSE 逐块转换的离线测试；其他路径发简化序列。""" + if path.rstrip("/").endswith("/v1/messages") or path.endswith("/messages"): + creation, read = _cache_tick(payload) + chunks = ["mock ", "stream ", "reply"] + msg_start = {"type": "message_start", "message": { + "id": "msg_mock_stream", "type": "message", "role": "assistant", + "model": payload.get("model", "mock"), "content": [], + "usage": {"input_tokens": _input_chars(payload), "output_tokens": 1, + "cache_creation_input_tokens": creation, + "cache_read_input_tokens": read}}} + yield "event: message_start\n" + yield f"data: {json.dumps(msg_start, ensure_ascii=False)}\n\n" + yield "event: content_block_start\n" + yield ('data: {"type":"content_block_start","index":0,' + '"content_block":{"type":"text","text":""}}\n\n') + for c in chunks: + delta = {"type": "content_block_delta", "index": 0, + "delta": {"type": "text_delta", "text": c}} + yield "event: content_block_delta\n" + yield f"data: {json.dumps(delta, ensure_ascii=False)}\n\n" + time.sleep(0.01) + yield "event: content_block_stop\n" + yield 'data: {"type":"content_block_stop","index":0}\n\n' + yield "event: message_delta\n" + yield ('data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},' + '"usage":{"output_tokens":12}}\n\n') + yield "event: message_stop\n" + yield 'data: {"type":"message_stop"}\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) yield 'data: {"type": "message_stop"}\n\n' @@ -177,7 +207,7 @@ def do_POST(self): self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.end_headers() - for line in sse_lines(payload): + for line in sse_lines(payload, self.path): self.wfile.write(line.encode("utf-8")) return body = json.dumps(payload_response(payload, self.path), diff --git a/tools/smoke_e2e.py b/tools/smoke_e2e.py index 235c227..7b34bcb 100644 --- a/tools/smoke_e2e.py +++ b/tools/smoke_e2e.py @@ -81,6 +81,25 @@ def post(path: str, payload: dict) -> dict: return json.loads(r.read().decode("utf-8")) +def post_stream(path: str, payload: dict): + """流式 POST：逐帧收 SSE，返回 (data 帧列表, 各帧到达时间列表)。 + + 到达时间用来证明网关是**逐块下发**而非整读后再吐（LIMITATIONS #1 的核心断言）。 + """ + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request(f"http://127.0.0.1:{GW_PORT}{path}", data=data, + headers={"Content-Type": "application/json"}) + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + frames, times = [], [] + with opener.open(req, timeout=20) as r: + for line in r: + line = line.decode("utf-8").strip() + if line.startswith("data:"): + frames.append(line[len("data:"):].strip()) + times.append(time.time()) + return frames, times + + CHECKS: list = [] @@ -150,6 +169,44 @@ def main() -> int: check("未知 target 回 400", e.code == 400, f"HTTP {e.code}") except Exception as e: # RemoteDisconnected 等 check("未知 target 回 400", False, f"{type(e).__name__}: {e}") + + print("\n6) SSE 逐块流式（openai_chat 客户端 ← anthropic 上游，LIMITATIONS #1）") + frames, times = post_stream("/v1/openai_chat/to/anthropic", + {"model": "mock", "stream": True, + "messages": [{"role": "user", "content": "hi"}]}) + chunks = [] + for f in frames: + if f == "[DONE]": + continue + try: + chunks.append(json.loads(f)) + except ValueError: + pass + check("收到多帧而非一整块", len(frames) >= 5, f"frames={len(frames)}") + check("首帧是 role 帧", + bool(chunks) and chunks[0]["choices"][0]["delta"].get("role") == "assistant", + frames[0][:80] if frames else "无帧") + text = "".join(c["choices"][0]["delta"].get("content", "") + for c in chunks) + check("delta 文本拼合完整", text == "mock stream reply", repr(text)) + check("有 finish_reason=stop 帧", + any(c["choices"][0].get("finish_reason") == "stop" for c in chunks)) + check("以 [DONE] 收尾", bool(frames) and frames[-1] == "[DONE]", + frames[-1][:40] if frames else "无帧") + # mock 每 chunk 间隔 10ms——若网关整读再吐，首尾帧会几乎同时到达 + spread = (times[-1] - times[0]) if len(times) >= 2 else 0 + check("帧是逐块到达的（非整读后补吐）", spread >= 0.015, + f"首尾间隔 {spread*1000:.1f}ms") + # 流尾落库与埋点：metrics 最后一行应是本轮流式请求 + # （TurnMetrics 口径为 5 项暴露，不含 output_tokens——断言 kind 与 input） + try: + with open(os.path.join(tmp, "m.jsonl"), encoding="utf-8") as fh: + last = json.loads(fh.readlines()[-1]) + check("流式轮照常进埋点", + last.get("kind") == "normal" and last.get("input_tokens", 0) > 0, + json.dumps(last, ensure_ascii=False)[:120]) + except Exception as e: + check("流式轮照常进埋点", False, f"{type(e).__name__}: {e}") finally: stop(tmp)