chat 流式优化:LLM streaming + 前端 SSE 全事件对接 + bug 修复

app/main.py

@@ -533,11 +533,18 @@ def _get_harness_llm():
         if not api_key:
             return None
 
+        # streaming=True so that astream_events(version="v2") receives
+        # per-token on_chat_model_stream events.  Without it, ChatOpenAI's
+        # _agenerate produces the whole answer in one shot, so the SSE
+        # streamer emits a single `chunk` frame with the full text and the
+        # frontend renders it non-incrementally.  stream_usage=True keeps
+        # token usage flowing in stream mode so usage tracking still works.
         return ChatOpenAI(
             api_key=api_key,
             base_url=settings.openai_base_url or None,
             model=settings.llm_model,
             temperature=0,
+            streaming=True,
             stream_usage=True,
         )
     except Exception as e:

app/routers/chat.py

@@ -137,6 +137,7 @@ async def ask_question_agent_stream(
         db=db,
         background_tasks=background_tasks,
         agent_harness=getattr(http_request.app.state, "agent_harness", None),
+        api_key_manager=getattr(http_request.app.state, "api_key_manager", None),
         usage_writer=getattr(http_request.app.state, "usage_writer", None),
     )
     return StreamingResponse(generator, media_type="text/event-stream")

app/services/chat/agent_sse.py

@@ -16,12 +16,11 @@
 from __future__ import annotations
 
 import json
-import logging
 from typing import Any, AsyncIterator, Optional
 
-from app.services.chat.sse import sse_event
+from loguru import logger
 
-logger = logging.getLogger(__name__)
+from app.services.chat.sse import sse_event
 
 _PREVIEW_LIMIT = 200
 
@@ -84,6 +83,11 @@ def __init__(self) -> None:
         self._step_no = 0
         self._tool_runs: dict[str, dict[str, Any]] = {}
         self._root_run_name: str = ""
+        # Error tracking: when stream() swallows an exception, these let the
+        # orchestrator fail_turn instead of finalize_turn (which would persist
+        # a partial answer as a successful message).
+        self.had_error: bool = False
+        self.error_message: str = ""
 
     async def stream(
         self,
@@ -93,15 +97,19 @@ async def stream(
     ) -> AsyncIterator[str]:
         """Yield SSE frames; mutate ``self.full_content`` / ``self.sources``."""
         self._root_run_name = run_config.get("run_name", "LangGraph")
+        event_counts: dict[str, int] = {}
         try:
             async for event in agent_graph.astream_events(
                 input_state, config=run_config, version="v2"
             ):
                 kind = event.get("event", "")
+                event_counts[kind] = event_counts.get(kind, 0) + 1
                 frame: Optional[str] = None
 
                 if kind == "on_chat_model_stream":
                     frame = self._handle_token(event)
+                elif kind == "on_chat_model_start":
+                    frame = self._handle_model_start(event)
                 elif kind == "on_tool_start":
                     frame = self._handle_tool_start(event)
                 elif kind == "on_tool_end":
@@ -112,12 +120,37 @@ async def stream(
                 if frame is not None:
                     yield frame
 
+            logger.info(
+                "[SSE_STREAMER] event_counts={} content_chars={} token_events={}",
+                event_counts,
+                len(self.full_content),
+                event_counts.get("on_chat_model_stream", 0),
+            )
             yield sse_event({"type": "sources", "sources": self.sources[:5]})
             yield sse_event({"type": "done"})
         except Exception as exc:
             logger.exception("Agent SSE stream failed")
+            self.had_error = True
+            self.error_message = str(exc)
             yield sse_event({"type": "error", "message": str(exc)})
 
+    def _handle_model_start(self, event: dict[str, Any]) -> Optional[str]:
+        """Reset accumulated content when a new LLM call begins mid-stream.
+
+        A second ``on_chat_model_start`` after content has already been
+        accumulated means the graph is re-running the LLM (e.g. a retryable
+        "peer closed connection" error triggered error_node -> agent). Without
+        a reset, the retry's tokens append to the partial first attempt,
+        producing garbled half+duplicate content both streamed to the client
+        and persisted. Emit a ``reset`` frame so the client clears its bubble,
+        and zero ``full_content`` so only the retried (complete) answer is
+        persisted.
+        """
+        if self.full_content:
+            self.full_content = ""
+            return sse_event({"type": "reset"})
+        return None
+
     def _capture_root_output(self, event: dict[str, Any]) -> None:
         """Extract token usage from the root graph's final state.
 
@@ -144,7 +177,7 @@ def _capture_root_output(self, event: dict[str, Any]) -> None:
 
         if self.total_tokens > 0:
             logger.info(
-                "[SSE_STREAMER] root chain end: tokens=%s (prompt=%s, completion=%s, calls=%s)",
+                "[SSE_STREAMER] root chain end: tokens={} (prompt={}, completion={}, calls={})",
                 self.total_tokens, self.prompt_tokens,
                 self.completion_tokens, self.llm_calls,
             )

app/services/chat/orchestrator.py

@@ -383,6 +383,7 @@ async def ask_agent_stream(
     db: AsyncSession,
     background_tasks: BackgroundTasks,
     agent_harness: Any,
+    api_key_manager: Any = None,
     usage_writer: Any = None,
 ) -> AsyncIterator[str]:
     """Handle POST /chat/ask/agent/stream — token-level SSE via AgentHarness."""
@@ -395,7 +396,8 @@ async def ask_agent_stream(
     )
 
     try:
-        credential_id, provider, model = _resolve_usage_metadata(uid, None)
+        await _preload_user_credentials(api_key_manager, uid, db)
+        credential_id, provider, model = _resolve_usage_metadata(uid, api_key_manager)
         agent_name, agent_graph, input_state, run_config = await agent_stream_setup(
             request,
             uid=uid,
@@ -484,6 +486,16 @@ async def _stream_agent_events(
     # render a spurious error after a successful answer.
     try:
         latency_ms = int((time.time() - start_time) * 1000)
+        if streamer.had_error:
+            # The streamer already emitted an `error` frame to the client.
+            # Mark the turn failed instead of finalize_turn, which would
+            # persist a partial answer as a successful message.
+            await fail_turn(
+                db,
+                assistant_msg_id=ctx.assistant_msg_id,
+                error=streamer.error_message or "stream failed",
+            )
+            return
         total_tokens = usage_callback.total_tokens
         logger.info(
             f"[CHAT_ORCH] stream done: tokens={total_tokens} "

frontend/app/globals.css

@@ -1886,6 +1886,20 @@ html.dark * {
   padding-top: 4px;
 }
 
+/* ---- Route badge (which agent handled the turn) ---- */
+.msg-route-badge {
+  align-self: flex-start;
+  display: inline-flex;
+  align-items: center;
+  gap: 4px;
+  padding: 2px 8px;
+  border-radius: 999px;
+  background: var(--gemini-hover, rgba(0, 0, 0, 0.04));
+  color: var(--gemini-text-tertiary, #5f6368);
+  font-size: 11px;
+  font-weight: 500;
+}
+
 /* ---- Reasoning toggle (Gemini chip) ---- */
 .msg-reasoning-toggle {
   align-self: flex-start;