From b317fa4b8e572decbe5f31877921e322271b4a99 Mon Sep 17 00:00:00 2001 From: zjr Date: Thu, 23 Jul 2026 11:27:13 +0800 Subject: [PATCH 1/2] =?UTF-8?q?[chat]=20fix:=20=E5=90=AF=E7=94=A8=20LLM=20?= =?UTF-8?q?streaming=20=E5=B9=B6=E4=BF=AE=E5=A4=8D=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E9=87=8D=E8=AF=95/credential/=E5=A4=B1=E8=B4=A5=E8=90=BD?= =?UTF-8?q?=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - main: _get_harness_llm 加 streaming=True，使 astream_events 逐 token 产出 on_chat_model_stream；否则 ChatOpenAI._agenerate 一次性产出整段答案， SSE streamer 只发一个 chunk 帧含全文，前端无法增量渲染 - agent_sse: 监听 on_chat_model_start，重试时（如 peer closed）发 reset 帧 并清空 full_content，避免半截+重复内容流式与落库 - agent_sse/orchestrator: streamer 标记 had_error/error_message，post-stream 据此走 fail_turn 而非 finalize_turn，失败回合不再当成功消息入库（且不重复 error 帧） - orchestrator/chat: ask_agent_stream 补齐 api_key_manager 预加载与 _resolve_usage_metadata(uid, api_key_manager)，用量不再误归系统默认 key --- app/main.py | 7 ++++++ app/routers/chat.py | 1 + app/services/chat/agent_sse.py | 41 ++++++++++++++++++++++++++++--- app/services/chat/orchestrator.py | 14 ++++++++++- 4 files changed, 58 insertions(+), 5 deletions(-) diff --git a/app/main.py b/app/main.py index a701758..805752d 100644 --- a/app/main.py +++ b/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: diff --git a/app/routers/chat.py b/app/routers/chat.py index 43577ba..ab6a03a 100644 --- a/app/routers/chat.py +++ b/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") diff --git a/app/services/chat/agent_sse.py b/app/services/chat/agent_sse.py index 859e3b1..0cb7fe9 100644 --- a/app/services/chat/agent_sse.py +++ b/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, ) diff --git a/app/services/chat/orchestrator.py b/app/services/chat/orchestrator.py index 69825ab..0c7933f 100644 --- a/app/services/chat/orchestrator.py +++ b/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} " From d2f427a1e7c2de972e2f4cacdf47c1ea6c779076 Mon Sep 17 00:00:00 2001 From: zjr Date: Thu, 23 Jul 2026 11:27:33 +0800 Subject: [PATCH 2/2] =?UTF-8?q?[frontend]=20feat:=20=E5=AE=8C=E6=95=B4?= =?UTF-8?q?=E5=AF=B9=E6=8E=A5=20agent=20=E6=B5=81=E5=BC=8F=20SSE=20?= =?UTF-8?q?=E5=B9=B6=E4=BC=98=E5=8C=96=E6=B5=81=E5=BC=8F=E6=B8=B2=E6=9F=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - chat-stream: 补齐 route/step/reset 事件处理（原仅 chunk/sources/error/done） - api: askStream 切到 /chat/ask/agent/stream 获取 route 帧 - ChatPanel/ChatDockPanel: 接 onStep（按 step 号合并 tool_start/end）/onRoute/onReset - ChatMessage: 包 React.memo 避免每 token 重渲染全部消息；流式中用纯文本 渲染，完成后再切 ReactMarkdown；route 标签 + 思考过程流式自动展开 - types: ChatMessageData 加 agent 字段；globals.css 加 route badge 样式 - 删除无引用的根 components/ChatPanel.tsx 死代码 --- frontend/app/globals.css | 14 + frontend/components/ChatPanel.tsx | 529 --------------------- frontend/components/chat/ChatContent.tsx | 1 + frontend/components/chat/ChatDockPanel.tsx | 47 ++ frontend/components/chat/ChatMessage.tsx | 33 +- frontend/components/chat/ChatPanel.tsx | 50 ++ frontend/components/chat/types.ts | 2 + frontend/lib/api.ts | 10 +- frontend/lib/chat-stream.ts | 21 + 9 files changed, 174 insertions(+), 533 deletions(-) delete mode 100644 frontend/components/ChatPanel.tsx diff --git a/frontend/app/globals.css b/frontend/app/globals.css index ddad159..d08eca2 100644 --- a/frontend/app/globals.css +++ b/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; diff --git a/frontend/components/ChatPanel.tsx b/frontend/components/ChatPanel.tsx deleted file mode 100644 index db151ef..0000000 --- a/frontend/components/ChatPanel.tsx +++ /dev/null @@ -1,529 +0,0 @@ -"use client"; - -import { useState, useRef, useEffect } from "react"; -import ReactMarkdown from "react-markdown"; -import remarkGfm from "remark-gfm"; -import { nanoid } from "nanoid"; -import { Loader2, AlertCircle, Trash2 } from "lucide-react"; -import { cn } from "@/lib/utils"; -import { sanitizeError } from "@/lib/error-utils"; -import { - chatApi, - knowledgeApi, - KnowledgeStats, - ChatMessage, - ChatRequestPayload, -} from "@/lib/api"; -import { Skeleton } from "@/components/ui/skeleton"; -import { useDockContext } from "@/lib/dock-context"; -import { useAppStore, LocalChatMessage } from "@/stores/app-store"; - -interface Props { - isOpen?: boolean; - onClose?: () => void; -} - -// 合并消息：按 id 或 clientId 去重，后端数据优先，按时间正序排列 -function mergeMessages( - existing: LocalChatMessage[], - incoming: LocalChatMessage[] -): LocalChatMessage[] { - const map = new Map(); - - for (const m of existing) { - const key = m.msg_id || m.clientId!!; - map.set(key, m); - } - - for (const m of incoming) { - const key = m.msg_id || m.clientId!!; - const existingMsg = map.get(key); - map.set(key, existingMsg ? { ...existingMsg, ...m } : m); - } - - return Array.from(map.values()).sort( - (a, b) => new Date(a.created_at).getTime() - new Date(b.created_at).getTime() - ); -} - -export default function ChatPanel({ isOpen, onClose }: Props) { - const { sessionId, workspacePages, activeChatSessionId: chatSessionId } = useDockContext(); - const statsKey = useAppStore((s) => s.statsKey); - const messages = useAppStore((s) => s.chatMessages); - const setChatMessages = useAppStore((s) => s.setChatMessages); - const clearChatMessages = useAppStore((s) => s.clearChatMessages); - - const [folderIds] = useState([]); - const [input, setInput] = useState(""); - const [loading, setLoading] = useState(false); - const [stats, setStats] = useState(null); - const [showReasoning, setShowReasoning] = useState>(new Set()); - - const [isHistoryLoading, setIsHistoryLoading] = useState(false); - const [hasMore, setHasMore] = useState(false); - - const endRef = useRef(null); - - // 加载知识库统计 - useEffect(() => { - knowledgeApi.getStats().then(setStats).catch(() => {}); - }, [statsKey]); - - // 自动滚动到底部 - useEffect(() => { - endRef.current?.scrollIntoView({ behavior: "smooth" }); - }, [messages]); - - // 当 chatSessionId 变化时加载历史消息 - useEffect(() => { - if (!chatSessionId) { - clearChatMessages(); - return; - } - - setIsHistoryLoading(true); - chatApi - .getHistory(chatSessionId) - .then((history) => { - setChatMessages(history.messages); - setHasMore(history.has_more); - }) - .catch((e) => console.error("加载历史失败", e)) - .finally(() => setIsHistoryLoading(false)); - }, [chatSessionId]); - - const toggleReasoning = (msgId: string) => { - setShowReasoning((prev) => { - const next = new Set(prev); - next.has(msgId) ? next.delete(msgId) : next.add(msgId); - return next; - }); - }; - - const handleClear = () => { - if (chatSessionId) { - chatApi.clearHistory(chatSessionId).catch(() => {}); - } - clearChatMessages(); - }; - - const send = async () => { - if (!input.trim() || loading) return; - if (!chatSessionId) { - console.error("chatSessionId 未初始化"); - return; - } - - const q = input.trim(); - setInput(""); - - const clientId = `client-${nanoid()}`; - const assistantClientId = `client-${nanoid()}`; - const now = new Date().toISOString(); - - // 乐观更新：先显示用户消息 + assistant 占位 - const optimisticUser: LocalChatMessage = { - msg_id: "", - clientId, - chat_session_id: chatSessionId, - role: "user", - content: q, - status: "completed", - created_at: now, - }; - const optimisticAssistant: LocalChatMessage = { - msg_id: "", - clientId: assistantClientId, - chat_session_id: chatSessionId, - role: "assistant", - content: "", - status: "pending", - created_at: now, - }; - - setChatMessages((prev) => mergeMessages(prev, [optimisticUser, optimisticAssistant])); - setLoading(true); - - const payload: ChatRequestPayload = { - question: q, - session_id: sessionId ?? undefined, - chat_session_id: chatSessionId, - folder_ids: folderIds, - workspace_pages: workspacePages, - }; - - try { - const stream = await chatApi.askStream(payload); - const reader = stream.getReader(); - const decoder = new TextDecoder("utf-8"); - let done = false; - let sseBuffer = ""; - let contentBuffer = ""; - - while (!done) { - const { value, done: doneReading } = await reader.read(); - done = doneReading; - if (!value) continue; - - sseBuffer += decoder.decode(value, { stream: !done }); - const events = sseBuffer.split("\n\n"); - sseBuffer = events.pop() || ""; - - for (const event of events) { - const lines = event.split("\n"); - let dataLine = ""; - for (const line of lines) { - if (line.startsWith("data:")) { - dataLine = line.slice(5).trim(); - } - } - if (!dataLine) continue; - - try { - const payload = JSON.parse(dataLine); - if (payload.type === "chunk") { - contentBuffer += payload.content || ""; - setChatMessages((prev) => - prev.map((m) => - m.clientId === assistantClientId - ? { ...m, content: contentBuffer } - : m - ) - ); - } else if (payload.type === "sources") { - const sources = Array.isArray(payload.sources) ? payload.sources : []; - setChatMessages((prev) => - prev.map((m) => - m.clientId === assistantClientId ? { ...m, sources } : m - ) - ); - } else if (payload.type === "error") { - setChatMessages((prev) => - prev.map((m) => - m.clientId === assistantClientId - ? { - ...m, - status: "failed" as const, - error: payload.message || "流式生成失败", - } - : m - ) - ); - } - // type === "done" 无需处理 - } catch { - // 忽略单行解析失败，继续处理后续事件 - } - } - } - } catch (e) { - // SSE 失败，尝试降级为非流式 - try { - const res = await chatApi.ask(payload); - setChatMessages((prev) => - prev.map((m) => - m.clientId === assistantClientId - ? { - ...m, - content: res.answer, - sources: res.sources, - status: "completed" as const, - } - : m - ) - ); - } catch (err) { - setChatMessages((prev) => - prev.map((m) => - m.clientId === assistantClientId - ? { - ...m, - status: "failed" as const, - error: sanitizeError(err), - } - : m - ) - ); - } - } - - setLoading(false); - }; - - if (!isOpen) return null; - - return ( -
-
-
-
对话工作台
- {stats && stats.total_videos > 0 && ( -
已收录 {stats.total_videos} 个视频
- )} -
- {messages.length > 0 && ( - - )} -
- -
-
- {isHistoryLoading && messages.length === 0 && ( -
- - - -
- )} - - {messages.length === 0 && !isHistoryLoading ? ( -
-
-
检索就绪
-

- 把收藏夹变成可提问的知识库 -

-
-
- {[ - "总结收藏夹里最有价值的内容", - "有哪些适合快速复习的系列？", - "列出与某个主题相关的视频并给出关键点", - "按主题整理我的收藏夹内容", - "用一句话概括每个视频的重点", - "推荐3个最适合入门的学习视频", - ].map((q, i) => ( - - ))} -
-
- ) : ( -
- {messages.map((m) => ( -
-
- {/* Pending 状态 */} - {m.role === "assistant" && m.status === "pending" && !m.content ? ( -
-
- - AI 思考中... -
-
- - -
-
- ) : ( - - {m.content || " "} - - )} - - {/* Failed 状态 */} - {m.status === "failed" && ( -
- - - {m.error || "回答生成失败"} - -
- )} - - {/* 推理过程 */} - {m.reasoningSteps && m.reasoningSteps.length > 0 && ( -
- - {showReasoning.has( - m.msg_id || m.clientId!! - ) && ( -
- {m.reasoningSteps.map((step, i) => ( -
-
- Step {step.step}: {step.action} -
-
- Query: - {step.query} -
-
- {step.reasoning} -
- {step.verdict && ( -
- Verdict: {step.verdict} -
- )} - {step.recall_score != null && ( -
- Recall: {step.recall_score.toFixed(3)} -
- )} - {step.sources.length > 0 && ( -
- {step.sources.map((s, j) => ( - - {s.title} - - ))} -
- )} -
- ))} -
- )} -
- )} - - {/* 来源链接 */} - {m.sources && m.sources.length > 0 && ( -
- {m.sources.map((s, i) => ( - - {s.title} - - ))} -
- )} -
-
- ))} -
-
- )} -
-
- -
-
- setInput(e.target.value)} - onKeyDown={(e) => e.key === "Enter" && !e.shiftKey && send()} - placeholder="输入问题..." - className="input" - /> - -
-
-
- ); -} diff --git a/frontend/components/chat/ChatContent.tsx b/frontend/components/chat/ChatContent.tsx index ff44abd..d7f6880 100644 --- a/frontend/components/chat/ChatContent.tsx +++ b/frontend/components/chat/ChatContent.tsx @@ -75,6 +75,7 @@ export default function ChatContent({ content={message.content} sources={message.sources} reasoningSteps={message.reasoningSteps} + agent={message.agent} status={message.status} error={message.error} timestamp={message.timestamp} diff --git a/frontend/components/chat/ChatDockPanel.tsx b/frontend/components/chat/ChatDockPanel.tsx index 8cf49cc..fa169c0 100644 --- a/frontend/components/chat/ChatDockPanel.tsx +++ b/frontend/components/chat/ChatDockPanel.tsx @@ -125,6 +125,53 @@ export default function ChatDockPanel({ isOpen, onClose }: ChatDockPanelProps) { ) ); }, + onRoute: (agent) => { + setMessages((prev) => + prev.map((m) => + m.id === assistantMsgId ? { ...m, agent } : m + ) + ); + }, + onReset: () => { + setMessages((prev) => + prev.map((m) => + m.id === assistantMsgId ? { ...m, content: "" } : m + ) + ); + }, + onStep: (step) => { + setMessages((prev) => + prev.map((m) => { + if (m.id !== assistantMsgId) return m; + const existing = m.reasoningSteps ?? []; + const idx = existing.findIndex((r) => r.step === step.step); + let next; + if (idx >= 0) { + // Merge: tool_start provides query, tool_end provides sources. + next = [...existing]; + next[idx] = { + ...next[idx], + action: step.action || next[idx].action, + query: step.query || next[idx].query, + reasoning: step.reasoning || next[idx].reasoning, + sources: step.sources?.length ? step.sources : next[idx].sources, + }; + } else { + next = [ + ...existing, + { + step: step.step, + action: step.action, + query: step.query, + reasoning: step.reasoning, + sources: step.sources ?? [], + }, + ]; + } + return { ...m, reasoningSteps: next }; + }) + ); + }, onError: (message) => { setMessages((prev) => prev.map((m) => diff --git a/frontend/components/chat/ChatMessage.tsx b/frontend/components/chat/ChatMessage.tsx index 075fc28..cd21185 100644 --- a/frontend/components/chat/ChatMessage.tsx +++ b/frontend/components/chat/ChatMessage.tsx @@ -1,6 +1,6 @@ "use client"; -import { useState } from "react"; +import { useState, useEffect, useRef, memo } from "react"; import ReactMarkdown from "react-markdown"; import remarkGfm from "remark-gfm"; import { @@ -36,6 +36,7 @@ interface ChatMessageProps { content: string; sources?: Source[] | null; reasoningSteps?: ReasoningStep[] | null; + agent?: string; status?: "pending" | "completed" | "failed"; error?: string; timestamp?: string; @@ -52,11 +53,12 @@ function domainOf(url?: string): string { } } -export default function ChatMessage({ +function ChatMessage({ role, content, sources, reasoningSteps, + agent, status = "completed", error, }: ChatMessageProps) { @@ -65,6 +67,15 @@ export default function ChatMessage({ const safeReasoningSteps = Array.isArray(reasoningSteps) ? reasoningSteps : []; const [showReasoning, setShowReasoning] = useState(false); + // Track manual toggle so auto-expand doesn't override the user's choice: + // while streaming (pending) with steps arriving, auto-expand; once the + // user toggles, respect their state. + const userToggledRef = useRef(false); + useEffect(() => { + if (!userToggledRef.current && safeReasoningSteps.length > 0 && status === "pending") { + setShowReasoning(true); + } + }, [safeReasoningSteps.length, status]); const [copied, setCopied] = useState(false); const [feedback, setFeedback] = useState<"up" | "down" | null>(null); @@ -101,11 +112,21 @@ export default function ChatMessage({
+ {/* Route badge - which agent handled this turn */} + {agent && ( +
+
+ )} {/* Reasoning toggle — Gemini style chip, above content */} {safeReasoningSteps.length > 0 && (
+ ) : isPending ? ( + // Streaming: render plain text to avoid re-parsing markdown on + // every token (the main cause of janky/non-incremental rendering). +
{content}
) : ( {content || ""} @@ -282,3 +307,5 @@ export default function ChatMessage({ ); } + +export default memo(ChatMessage); diff --git a/frontend/components/chat/ChatPanel.tsx b/frontend/components/chat/ChatPanel.tsx index 8fb6c56..a8e64e7 100644 --- a/frontend/components/chat/ChatPanel.tsx +++ b/frontend/components/chat/ChatPanel.tsx @@ -207,6 +207,56 @@ export default function ChatPanel() { ), })); }, + onRoute: (agent) => { + updateActiveSession((s) => ({ + ...s, + messages: s.messages.map((m) => + m.id === assistantMsgId ? { ...m, agent } : m + ), + })); + }, + onReset: () => { + updateActiveSession((s) => ({ + ...s, + messages: s.messages.map((m) => + m.id === assistantMsgId ? { ...m, content: "" } : m + ), + })); + }, + onStep: (step) => { + updateActiveSession((s) => ({ + ...s, + messages: s.messages.map((m) => { + if (m.id !== assistantMsgId) return m; + const existing = m.reasoningSteps ?? []; + const idx = existing.findIndex((r) => r.step === step.step); + let next; + if (idx >= 0) { + // Merge: tool_start provides query, tool_end provides sources. + next = [...existing]; + next[idx] = { + ...next[idx], + action: step.action || next[idx].action, + query: step.query || next[idx].query, + reasoning: step.reasoning || next[idx].reasoning, + sources: step.sources?.length ? step.sources : next[idx].sources, + }; + } else { + next = [ + ...existing, + { + step: step.step, + action: step.action, + query: step.query, + reasoning: step.reasoning, + sources: step.sources ?? [], + }, + ]; + } + return { ...m, reasoningSteps: next }; + }), + })); + }, onError: (message) => { updateActiveSession((s) => ({ ...s, diff --git a/frontend/components/chat/types.ts b/frontend/components/chat/types.ts index 907fc75..d534666 100644 --- a/frontend/components/chat/types.ts +++ b/frontend/components/chat/types.ts @@ -20,6 +20,8 @@ export interface ChatMessageData { content: string; sources?: ChatSource[]; reasoningSteps?: ReasoningStep[]; + // Agent name routed to by AgentOrchestrator (from the `route` SSE frame). + agent?: string; status: MessageStatus; error?: string; timestamp: string; diff --git a/frontend/lib/api.ts b/frontend/lib/api.ts index b4d78d7..ae7a003 100644 --- a/frontend/lib/api.ts +++ b/frontend/lib/api.ts @@ -590,7 +590,15 @@ export const chatApi = { // === 新增：流式接口（替代裸调 fetch）=== askStream: async (payload: ChatRequestPayload): Promise> => { - const res = await fetch(`${API_BASE_URL}/chat/ask/stream`, { + // Dev bypass: when API_BASE_URL is "" (browser + no NEXT_PUBLIC_API_URL), + // requests route through Next.js rewrites whose dev proxy buffers SSE, + // collapsing token-by-token streaming into one bulk delivery. Hit the + // backend directly. Prod sets NEXT_PUBLIC_API_URL (nginx), so this + // localhost fallback only applies in dev. + const streamBase = + API_BASE_URL || + (typeof window !== "undefined" ? "http://localhost:8000" : "http://backend:8000"); + const res = await fetch(`${streamBase}/chat/ask/agent/stream`, { method: "POST", headers: { "Content-Type": "application/json", ...getAuthHeaders() }, body: JSON.stringify(payload), diff --git a/frontend/lib/chat-stream.ts b/frontend/lib/chat-stream.ts index 7d5c9e9..a07de1e 100644 --- a/frontend/lib/chat-stream.ts +++ b/frontend/lib/chat-stream.ts @@ -8,11 +8,25 @@ export interface ChatSource { bvid?: string; } +// One retrieval/tool step emitted by the agent stream. +// Mirrors the backend `step` SSE frame payload (agent_sse.py). +export interface StreamStep { + step: number; + action: string; + query?: string; + reasoning?: string; + sources?: ChatSource[]; + content_preview?: string; +} + export interface StreamCallbacks { onChunk: (accumulated: string, delta: string) => void; onSources?: (sources: ChatSource[]) => void; onError?: (message: string) => void; onComplete?: () => void; + onStep?: (step: StreamStep) => void; + onRoute?: (agent: string) => void; + onReset?: () => void; } export interface StreamRequestParams { @@ -60,6 +74,13 @@ export async function streamChat( callbacks.onChunk(accumulated, delta); } else if (data.type === "sources") { callbacks.onSources?.(Array.isArray(data.sources) ? data.sources : []); + } else if (data.type === "step") { + callbacks.onStep?.(data.step as StreamStep); + } else if (data.type === "route") { + callbacks.onRoute?.(data.agent as string); + } else if (data.type === "reset") { + accumulated = ""; + callbacks.onReset?.(); } else if (data.type === "error") { callbacks.onError?.(data.message || data.error || "请求失败"); } else if (data.type === "done") {