Refactor _chat_stream_events in api_server.py to standardize SSE response format by removing additional status codes and consolidating output into a unified JSON structure. Update logging to enhance clarity on streaming events and improve error handling. Adjust documentation to reflect changes in event data structure.

api_server.py

@@ -659,7 +659,9 @@ async def _run_generate(
 async def _chat_stream_events(request: NLChatRequest) -> AsyncIterator[bytes]:
     if not request.message or not request.message.strip():
         lang = _normalize_lang_code(request.lang_code)
-        yield _sse_data({"code": 400, "msg": _localized_empty_input_reply(lang), "data": None})
+        yield _sse_data(
+            {"stage": "sql_gen", "stream_kind": "content", "content": _localized_empty_input_reply(lang)}
+        )
         return
     text = request.message.strip()
     orch = get_orchestrator()
@@ -696,16 +698,13 @@ async def _chat_stream_events(request: NLChatRequest) -> AsyncIterator[bytes]:
             len(reply),
             reply[:500] + ("…" if len(reply) > 500 else ""),
         )
-        yield _sse_data({"stage": "orchestrator", "stream_kind": "content", "content": "CHAT"})
-        async for pkt in _sse_stream_text_chunks("chat", reply):
+        async for pkt in _sse_stream_text_chunks("sql_gen", reply):
             yield pkt
         data_dict = _conversation_nl_dict(reply)
         data_dict["llm"] = _effective_llm_route_from_request(request)
-        yield _sse_data({"code": 200, "msg": "success", "data": data_dict})
         await _append_session_if_needed(request, text, data_dict)
         return
 
-    yield _sse_data({"stage": "orchestrator", "stream_kind": "content", "content": "DATA_QUERY"})
     dialect = resolve_sql_dialect(os.getenv("TEXT2SQL_DIALECT", "sqlserver"))
     dc_ctx = (dialog_block or "").strip() or None
     top_k = 20
@@ -759,7 +758,7 @@ async def _chat_stream_events(request: NLChatRequest) -> AsyncIterator[bytes]:
             result = holder["result"]
     except Exception as e:
         logger.error(f"[API/stream] 生成异常: {e}", exc_info=True)
-        yield _sse_data({"code": 500, "msg": str(e), "data": None})
+        yield _sse_data({"stage": "sql_gen", "stream_kind": "content", "content": str(e)})
         return
 
     sql_out = (result.sql or "").strip()
@@ -792,8 +791,6 @@ async def _chat_stream_events(request: NLChatRequest) -> AsyncIterator[bytes]:
 
     data_dict = _nl_dict_from_generation(result)
     data_dict["llm"] = _effective_llm_route_from_request(request)
-    msg = "success" if result.valid else "partial"
-    yield _sse_data({"code": 200, "msg": msg, "data": data_dict})
     await _append_session_if_needed(request, text, data_dict)
 
 
@@ -919,8 +916,8 @@ async def nl_chat(request: NLChatRequest):
 async def nl_chat_stream(request: NLChatRequest):
     """
     自然语言对话流式接口（SSE）。
-    事件体为 JSON：分片 `data: {"stage","stream_kind","content"}`（例：sql_gen 时
-    `stream_kind` 为 `content`）或结束包 `{code, msg, data}`。
+    事件体为 JSON：仅发送分片 `data: {"stage":"sql_gen","stream_kind":"content","content":"..."}`，
+    流结束表示完成；不再发送额外的结束包。
     """
     return StreamingResponse(
         _chat_stream_events(request),