From 2fcc0a4d17bf47ad9d13da5b7e278b7ef771b875 Mon Sep 17 00:00:00 2001 From: Sunsh1neY <1909164966@qq.com> Date: Thu, 3 Sep 2026 22:18:28 +0800 Subject: [PATCH] =?UTF-8?q?feat(chat):=20=E6=B5=81=E5=BC=8F=E8=BD=AC?= =?UTF-8?q?=E5=8F=91=E5=B8=A6=E5=B7=A5=E5=85=B7=E8=AF=B7=E6=B1=82=E7=9A=84?= =?UTF-8?q?=E6=AD=A3=E6=96=87=EF=BC=8Ctool=5Fcall=20=E5=9B=B4=E6=A0=8F?= =?UTF-8?q?=E7=BC=93=E5=86=B2=E8=A7=A3=E6=9E=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 此前带工具的流式请求被整体降级为非流式：generate() 整段生成后再包装成 SSE 一次性发出，agent 类客户端（ZCode/dsh 等）每轮都带工具定义，因此 永远看不到流式效果。 新增工具流式路径：正文 delta 实时转发；检测到 ```tool_call 围栏开始 标记（含跨 delta 边界的情况）后停止转发、转入缓冲，流结束后解析为 OpenAI 流式 tool_calls delta 事件（id/name/arguments），finish_reason 正确置为 tool_calls。 附带修复：围栏解析失败（生成中断导致围栏未闭合、JSON 不完整等）时， 原实现会在 parse_tool_calls 中静默丢弃整个工具调用——模型的调用意图 凭空消失，agent 反复重试形成死循环。现在缓冲的原始文本会作为正文补发 给客户端，不再有内容凭空丢失。 - 纯文本回复：完全实时流式（仅末尾 11 字符延迟到流结束防围栏标记跨界） - 工具调用回合：正文部分流式，仅 tool_call JSON 部分等流结束后解析 - 现有 18 个测试全部通过，新增 3 个测试覆盖工具解析/残缺回退/纯文本 - 与 #92（流中断优雅收尾）配合体验最佳；无硬依赖——重试耗尽时本路径 的异常分支会记录 Stream error 并结束，与既有降级路径行为一致 --- gemini_web2api/server.py | 78 ++++++++++++++++++++++++++++++++ tests/test_modular_sync.py | 92 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 170 insertions(+) diff --git a/gemini_web2api/server.py b/gemini_web2api/server.py index 3fc35b6..d56a0ae 100644 --- a/gemini_web2api/server.py +++ b/gemini_web2api/server.py @@ -13,6 +13,9 @@ from .multimodal import detect_image_mime, fetch_image_bytes, upload_image from . import __version__ +# Fence marker the model is instructed to wrap tool calls in (see tools.py). +TOOL_CALL_MARKER = "```tool_call" + def _usage(prompt: str, text: str) -> dict: p = len(prompt) // 4 @@ -192,6 +195,7 @@ def _handle_chat(self, body: bytes): return stream = req.get("stream", False) + log(f"Chat completions: stream={stream}, tools={len(tools) if tools else 0}, model={model_name}") cid = f"chatcmpl-{uuid.uuid4().hex[:12]}" try: file_refs = _upload_images(images) @@ -231,6 +235,80 @@ def _handle_chat(self, body: bytes): log(f"Stream error: {e}") return + if stream: + # Tools present + streaming: forward prose deltas in real time, + # but hold back ```tool_call fenced blocks (including a fence start + # straddling delta boundaries) so they can be parsed into OpenAI + # tool_calls at end of stream instead of leaking into chat text. + self._start_sse() + first_chunk = { + "id": cid, + "object": "chat.completion.chunk", + "created": int(time.time()), + "model": model_name, + "choices": [{ + "index": 0, + "delta": {"role": "assistant"}, + "finish_reason": None, + }], + } + self.wfile.write(f"data: {json.dumps(first_chunk)}\n\n".encode()) + self.wfile.flush() + + def send_delta(content=None, tool_calls=None, finish=None): + delta = {} + if content is not None: + delta["content"] = content + if tool_calls: + delta["tool_calls"] = tool_calls + chunk = {"id": cid, "object": "chat.completion.chunk", "created": int(time.time()), + "model": model_name, "choices": [{"index": 0, "delta": delta, "finish_reason": finish}]} + self.wfile.write(f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n".encode()) + self.wfile.flush() + + full_text = "" + emitted = 0 + finish = "stop" + try: + for delta in generate_stream(prompt, model_id, think_mode, file_refs, extra_fields): + full_text += delta + marker_pos = full_text.find(TOOL_CALL_MARKER) + # Without a marker, hold back the last len(marker)-1 chars so a + # fence start split across deltas is not forwarded prematurely. + limit = marker_pos if marker_pos != -1 else len(full_text) - len(TOOL_CALL_MARKER) + 1 + if limit > emitted: + send_delta(content=full_text[emitted:limit]) + emitted = limit + clean, tool_calls = parse_tool_calls(full_text) + if tool_calls: + log(f"Chat tool-fenced streaming: parsed {len(tool_calls)} tool call(s)") + if len(clean) > emitted: + send_delta(content=clean[emitted:]) + for i, tc in enumerate(tool_calls): + send_delta(tool_calls=[{ + "index": i, + "id": tc["id"], + "type": "function", + "function": { + "name": tc["function"]["name"], + "arguments": tc["function"]["arguments"], + }, + }]) + finish = "tool_calls" + else: + # No (or malformed) tool call: forward whatever is still + # buffered, raw, so nothing disappears silently. + if len(full_text) > emitted: + send_delta(content=full_text[emitted:]) + send_delta(finish=finish) + self.wfile.write(b"data: [DONE]\n\n") + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError): + pass + except Exception as e: + log(f"Stream error: {e}") + return + try: text = generate(prompt, model_id, think_mode, file_refs, extra_fields) except Exception as e: diff --git a/tests/test_modular_sync.py b/tests/test_modular_sync.py index 7479e92..6a5c3dc 100644 --- a/tests/test_modular_sync.py +++ b/tests/test_modular_sync.py @@ -440,6 +440,98 @@ def test_responses_function_call_stream_has_complete_event_sequence( self.assertEqual(events[4][1]["arguments"], '{"city":"Shanghai"}') self.assertEqual(events[-1][1]["response"]["output"][0]["name"], "get_weather") + def _stream_chunks(self, body): + return [ + json.loads(line[len("data: "):]) + for line in body.splitlines() + if line.startswith("data: {") + ] + + @staticmethod + def _tools_payload(): + return [{ + "type": "function", + "function": {"name": "write_file", "description": "write a file", "parameters": {}}, + }] + + @mock.patch("gemini_web2api.server.generate_stream") + def test_chat_stream_with_tools_parses_tool_call_deltas(self, generate_stream): + generate_stream.return_value = iter([ + "Creating the file.\n", + "```tool_call\n{\"name\": \"write_file\", \"arguments\": {\"path\": \"a.txt\"}}\n```", + ]) + + status, _, body = self.post_json( + "/v1/chat/completions", + { + "model": "gemini-3.6-flash", + "messages": [{"role": "user", "content": "go"}], + "stream": True, + "tools": self._tools_payload(), + }, + ) + + self.assertEqual(status, 200) + chunks = self._stream_chunks(body) + contents = "".join(c["choices"][0]["delta"].get("content", "") for c in chunks) + self.assertEqual(contents, "Creating the file.\n") + self.assertNotIn("tool_call", contents) + tc_deltas = [c for c in chunks if c["choices"][0]["delta"].get("tool_calls")] + self.assertEqual(len(tc_deltas), 1) + call = tc_deltas[0]["choices"][0]["delta"]["tool_calls"][0] + self.assertTrue(call["id"].startswith("call_")) + self.assertEqual(call["type"], "function") + self.assertEqual(call["function"]["name"], "write_file") + self.assertEqual(json.loads(call["function"]["arguments"]), {"path": "a.txt"}) + finishes = [c["choices"][0]["finish_reason"] for c in chunks if c["choices"][0]["finish_reason"]] + self.assertEqual(finishes, ["tool_calls"]) + self.assertTrue(body.endswith("data: [DONE]\n\n")) + + @mock.patch("gemini_web2api.server.generate_stream") + def test_chat_stream_malformed_tool_call_falls_back_to_text(self, generate_stream): + raw = '```tool_call\n{"name": "write_file", "arguments": {"path":' + generate_stream.return_value = iter([raw]) + + status, _, body = self.post_json( + "/v1/chat/completions", + { + "model": "gemini-3.6-flash", + "messages": [{"role": "user", "content": "go"}], + "stream": True, + "tools": self._tools_payload(), + }, + ) + + self.assertEqual(status, 200) + chunks = self._stream_chunks(body) + contents = "".join(c["choices"][0]["delta"].get("content", "") for c in chunks) + self.assertEqual(contents, raw) + finishes = [c["choices"][0]["finish_reason"] for c in chunks if c["choices"][0]["finish_reason"]] + self.assertEqual(finishes, ["stop"]) + self.assertTrue(body.endswith("data: [DONE]\n\n")) + + @mock.patch("gemini_web2api.server.generate_stream") + def test_chat_stream_with_tools_plain_text_fully_streamed(self, generate_stream): + generate_stream.return_value = iter(["Hello", " there friend"]) + + status, _, body = self.post_json( + "/v1/chat/completions", + { + "model": "gemini-3.6-flash", + "messages": [{"role": "user", "content": "go"}], + "stream": True, + "tools": self._tools_payload(), + }, + ) + + self.assertEqual(status, 200) + chunks = self._stream_chunks(body) + contents = "".join(c["choices"][0]["delta"].get("content", "") for c in chunks) + self.assertEqual(contents, "Hello there friend") + finishes = [c["choices"][0]["finish_reason"] for c in chunks if c["choices"][0]["finish_reason"]] + self.assertEqual(finishes, ["stop"]) + self.assertTrue(body.endswith("data: [DONE]\n\n")) + if __name__ == "__main__": unittest.main()