From bcb9a205fa394bb7360add1187c0e50603e3c944 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E8=BE=85=E5=85=83?= <2401926342@qq.com> Date: Fri, 17 Apr 2026 09:58:08 +0800 Subject: [PATCH] Refactor SSE streaming behavior in `api_server.py` to optimize data transmission by reducing redundant ApiEnvelope in dual mode. Update documentation to clarify streaming shapes and their respective behaviors, ensuring better alignment with client expectations. Enhance handling of stage deltas and improve overall response structure for SQL generation phases. --- api_server.py | 33 +++++++++++---------------------- 1 file changed, 11 insertions(+), 22 deletions(-) diff --git a/api_server.py b/api_server.py index f718875..1736769 100644 --- a/api_server.py +++ b/api_server.py @@ -245,10 +245,10 @@ def _sse_data(obj: Dict[str, Any]) -> bytes: _DEFAULT_SSE_CHUNK_CHARS = int(os.getenv("SSE_STREAM_CHUNK_CHARS", "64")) # 流式输出形态： -# - dual（默认）：同时发送 UI 所需的 {stage, stream_kind, content} delta， -# 并同步发送 ApiEnvelope（data 内携带逐步增长的字段），兼顾 chatStore 打字效果与“data 内流式”。 -# - envelope：仅 ApiEnvelope（适合非本仓库前端/自定义消费端；本仓库 Vue 侧会看起来不“打字”）。 -# - delta：仅 delta + 末尾 envelope（旧行为） +# - dual（默认）：发送 {stage, stream_kind, content} delta（与 delta 一致）；DATA_QUERY 生成阶段不再 +# 每条 delta 重复一份完整 ApiEnvelope，避免与前端拼接重复、体积 O(n²)。需要“仅 envelope 增长”时用 envelope。 +# - envelope：仅 ApiEnvelope，data.stream_narrative 逐步增长（无 stage delta；适合只解析 envelope 的客户端）。 +# - delta：仅 delta + 末尾 envelope（与 dual 在 SQL 流式段落行为一致）。 _SSE_STREAM_SHAPE = (os.getenv("SSE_STREAM_SHAPE", "dual") or "dual").strip().lower() @@ -835,6 +835,7 @@ async def _chat_stream_events(request: NLChatRequest) -> AsyncIterator[bytes]: loop.call_soon_threadsafe(chunk_queue.put_nowait, None) gen_task = asyncio.create_task(asyncio.to_thread(_run_generate_sync)) + # 仅 envelope 模式需要累积全文；dual/delta 只发 stage 增量，结束包再带完整 data stream_acc = "" while True: piece = await chunk_queue.get() @@ -885,14 +886,8 @@ async def _chat_stream_events(request: NLChatRequest) -> AsyncIterator[bytes]: stream_acc += chunk async for pkt in _yield_data_query_stream_narrative_envelope(stream_acc): yield pkt - elif _sse_stream_shape_dual(): - stream_acc += chunk - yield _sse_data( - {"stage": "sql_gen", "stream_kind": "content", "content": chunk} - ) - async for pkt in _yield_data_query_stream_narrative_envelope(stream_acc): - yield pkt else: + # dual / delta：仅 stage 增量，避免每条再套一层完整 ApiEnvelope yield _sse_data( {"stage": "sql_gen", "stream_kind": "content", "content": chunk} ) @@ -902,11 +897,6 @@ async def _chat_stream_events(request: NLChatRequest) -> AsyncIterator[bytes]: stream_acc += carry async for pkt in _yield_data_query_stream_narrative_envelope(stream_acc): yield pkt - elif _sse_stream_shape_dual(): - stream_acc += carry - yield _sse_data({"stage": "sql_gen", "stream_kind": "content", "content": carry}) - async for pkt in _yield_data_query_stream_narrative_envelope(stream_acc): - yield pkt else: yield _sse_data({"stage": "sql_gen", "stream_kind": "content", "content": carry}) carry = "" @@ -1122,14 +1112,13 @@ async def nl_chat_stream(request: NLChatRequest): """ 自然语言对话流式接口（SSE）。 - 默认（`SSE_STREAM_SHAPE=dual`）同时发送两类事件（适配本仓库前端 chatStore 的增量渲染）： - - UI 增量：`data: {"stage":"sql_gen","stream_kind":"content","content":"..."}`（用于打字/拼接 sqlGenHtml） - - 结构化增量：`data: {"code":200,"msg":"streaming","data":{...}}`（`DATA_QUERY` 时增长 `data.stream_narrative`） - - 结束：再发送最终 `{"code":200,"msg":"success|partial","data":{...}}`（含真实 SQL 等） + 默认（`SSE_STREAM_SHAPE=dual`）在 DATA_QUERY SQL 生成阶段只推送 stage delta（与 `delta` 一致）； + 寒暄等仍会按 dual 发送 delta + 增长的 `data.branch_result.answer`。结束包为完整 + `{"code":200,"msg":"success|partial","data":{...}}`。 其它模式： - - `SSE_STREAM_SHAPE=envelope`：仅 ApiEnvelope（自定义消费端可用；本仓库 UI 可能不显示“打字”） - - `SSE_STREAM_SHAPE=delta`：仅 delta + 末尾 envelope（旧行为） + - `SSE_STREAM_SHAPE=envelope`：无 stage delta，`data.stream_narrative` 在 envelope 内逐步增长 + - `SSE_STREAM_SHAPE=delta`：与当前 dual 在 SQL 流式段行为相同 """ return StreamingResponse( _chat_stream_events(request),