From 62da8b3c58ea1a42c4ee2c6db67b03cb64ad07b5 Mon Sep 17 00:00:00 2001 From: xiaopeng <1509442308@qq.com> Date: Wed, 27 May 2026 11:33:32 +0800 Subject: [PATCH] fix: async document processing in background threads, improve chat UX MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix knowledge_base_service calling async process_document synchronously in background threads — use asyncio.new_event_loop().run_until_complete() instead of bare call that returned unawaited coroutine - Same fix for _update_document and reindex_document - Replace ModeSelector with auto-detect: RAG mode when knowledge bases selected - Convert regenerateMessage to streaming (was synchronous API call) - Show "知识库检索模式" indicator above input when KB selected - Remove loadSessions side-effect from streaming onComplete - Fix chat page loading state to avoid flash - Improve forum page spacing and sizing Co-Authored-By: Claude Opus 4.7 --- .../src/services/knowledge_base_service.py | 20 +- web/src/app/(main)/chat/page.tsx | 7 +- web/src/app/(main)/forum/page.tsx | 210 +++++++++--------- web/src/components/chat/chat-interface.tsx | 15 +- web/src/store/chat.ts | 155 +++++++++---- 5 files changed, 250 insertions(+), 157 deletions(-) diff --git a/backend/src/services/knowledge_base_service.py b/backend/src/services/knowledge_base_service.py index 23f1e28..0c6bb06 100644 --- a/backend/src/services/knowledge_base_service.py +++ b/backend/src/services/knowledge_base_service.py @@ -161,14 +161,16 @@ class KnowledgeBaseService: # 处理文档（向量化）- 在后台异步处理，不阻塞主流程 try: - # 使用同步方法，但不等待完成（在后台处理） import threading + import asyncio as _asyncio def process_in_background(): try: - self.document_service.process_document(document.id) + loop = _asyncio.new_event_loop() + loop.run_until_complete(self.document_service.process_document(document.id)) + loop.close() except Exception as e: print(f"后台处理文档 {document.id} 失败: {e}") - + thread = threading.Thread(target=process_in_background, daemon=True) thread.start() except Exception as e: @@ -198,12 +200,15 @@ class KnowledgeBaseService: # 重新处理文档 - 在后台异步处理 try: import threading + import asyncio as _asyncio def process_in_background(): try: - self.document_service.process_document(document.id) + loop = _asyncio.new_event_loop() + loop.run_until_complete(self.document_service.process_document(document.id)) + loop.close() except Exception as e: print(f"后台处理文档 {document.id} 失败: {e}") - + thread = threading.Thread(target=process_in_background, daemon=True) thread.start() except Exception as e: @@ -340,7 +345,10 @@ class KnowledgeBaseService: self.db.query(DocumentChunk).filter(DocumentChunk.document_id == document_id).delete() # 重新处理文档 - success = self.document_service.process_document(document_id) + import asyncio as _asyncio + success = _asyncio.get_event_loop().run_until_complete( + self.document_service.process_document(document_id) + ) if success: return {"success": True, "message": "文档重新索引成功"} diff --git a/web/src/app/(main)/chat/page.tsx b/web/src/app/(main)/chat/page.tsx index e313653..14286b2 100644 --- a/web/src/app/(main)/chat/page.tsx +++ b/web/src/app/(main)/chat/page.tsx @@ -14,9 +14,10 @@ import { PanelLeftClose, PanelLeftOpen } from "lucide-react"; export default function ChatPage() { const router = useRouter(); const { isAuthenticated, user, isLoading: authLoading } = useAuthStore(); - const { loadSessions, isLoading: chatLoading } = useChatStore(); + const { loadSessions, sessions } = useChatStore(); const [isInitialized, setIsInitialized] = useState(false); const [sidebarOpen, setSidebarOpen] = useState(true); + const [initialLoading, setInitialLoading] = useState(true); useEffect(() => { if (!authLoading && !isAuthenticated) { @@ -25,12 +26,12 @@ export default function ChatPage() { } if (isAuthenticated && !isInitialized) { - loadSessions(); + loadSessions().then(() => setInitialLoading(false)); setIsInitialized(true); } }, [isAuthenticated, authLoading, isInitialized, router, loadSessions]); - if (authLoading || chatLoading) { + if (authLoading || initialLoading) { return (
diff --git a/web/src/app/(main)/forum/page.tsx b/web/src/app/(main)/forum/page.tsx index 33b64f6..a67e972 100644 --- a/web/src/app/(main)/forum/page.tsx +++ b/web/src/app/(main)/forum/page.tsx @@ -25,11 +25,11 @@ const CATEGORY_ORDER = ["公告", "通知", "学习", "讨论", "课程", "反 function getCategoryIcon(name: string) { if (name.includes("公告") || name.includes("通知")) - return ; + return ; if (name.includes("学习") || name.includes("讨论") || name.includes("课程")) - return ; + return ; if (name.includes("反馈") || name.includes("使用")) - return ; + return ; return CATEGORY_ICONS.default; } @@ -106,110 +106,112 @@ export default function ForumHomePage() { }, []); return ( -
- {isLoading ? ( -
- - 正在加载社区... -
- ) : error ? ( -
-

{error}

-
- ) : ( -
- {[...categories] - .sort((a, b) => { - const getOrder = (name: string) => { - for (let i = 0; i < CATEGORY_ORDER.length; i++) { - if (name.includes(CATEGORY_ORDER[i])) return i; - } - return CATEGORY_ORDER.length; - }; - return getOrder(a.name) - getOrder(b.name); - }) - .map((category) => { - const posts = categoryPosts[category.id] || []; - return ( -
- {/* Category header */} -
-
- - {getCategoryIcon(category.name)} - -
-

- {category.name} -

- {category.description && ( -

- {category.description} -

- )} -
-
-
- - {category.post_count} 条讨论 - - - 查看全部 - - -
-
- - {/* Posts list */} -
- {posts.length > 0 ? ( - posts.map((post) => ( - -
- -
-
-

- {post.title} +

+
+ {isLoading ? ( +
+ + 正在加载社区... +
+ ) : error ? ( +
+

{error}

+
+ ) : ( +
+ {[...categories] + .sort((a, b) => { + const getOrder = (name: string) => { + for (let i = 0; i < CATEGORY_ORDER.length; i++) { + if (name.includes(CATEGORY_ORDER[i])) return i; + } + return CATEGORY_ORDER.length; + }; + return getOrder(a.name) - getOrder(b.name); + }) + .map((category) => { + const posts = categoryPosts[category.id] || []; + return ( +
+ {/* Category header */} +
+
+ + {getCategoryIcon(category.name)} + +
+

+ {category.name} +

+ {category.description && ( +

+ {category.description}

-
- {post.author_name} - - - {formatRelativeTime(post.created_at)} - - - - {post.reply_count} - -
-
- - - )) - ) : ( -
- 暂无讨论，成为第一个发帖的人 + )} +
- )} +
+ + {category.post_count} 条讨论 + + + 查看全部 + + +
+
+ + {/* Posts list */} +
+ {posts.length > 0 ? ( + posts.map((post) => ( + +
+ +
+
+

+ {post.title} +

+
+ {post.author_name} + + + {formatRelativeTime(post.created_at)} + + + + {post.reply_count} + +
+
+ + + )) + ) : ( +
+ 暂无讨论，成为第一个发帖的人 +
+ )} +
-
- ); - })} -
- )} + ); + })} +
+ )} +
); } diff --git a/web/src/components/chat/chat-interface.tsx b/web/src/components/chat/chat-interface.tsx index f46cca5..9c970bc 100644 --- a/web/src/components/chat/chat-interface.tsx +++ b/web/src/components/chat/chat-interface.tsx @@ -4,10 +4,9 @@ import { useState, useRef, useEffect } from "react"; import { useChatStore } from "@/store/chat"; import { Button } from "@/components/ui/button"; import { Input } from "@/components/ui/input"; -import { Send, Square, Bot, ChevronDown, Settings2 } from "lucide-react"; +import { Send, Square, Bot, ChevronDown, Settings2, Database } from "lucide-react"; import MessageList from "./message-list"; import QuickQuestions from "./quick-questions"; -import ModeSelector, { ChatMode } from "./mode-selector"; import ModelSelector from "./model-selector"; import KnowledgeSelector, { KnowledgeBase } from "./knowledge-selector"; import { cn } from "@/lib/utils"; @@ -16,7 +15,6 @@ import { knowledgeBaseAPI } from "@/lib/api"; export default function ChatInterface() { const [inputMessage, setInputMessage] = useState(""); const [isComposing, setIsComposing] = useState(false); - const [chatMode, setChatMode] = useState("normal"); const [selectedModel, setSelectedModel] = useState("deepseek-ai/DeepSeek-V3"); const [selectedKnowledgeBases, setSelectedKnowledgeBases] = useState([]); const [systemKnowledgeBases, setSystemKnowledgeBases] = useState([]); @@ -72,7 +70,8 @@ export default function ChatInterface() { await selectSession(newSession.id); } } - await streamMessage(message, chatMode, selectedKnowledgeBases, selectedModel); + const mode = selectedKnowledgeBases.length > 0 ? "rag" : "normal"; + await streamMessage(message, mode, selectedKnowledgeBases, selectedModel); }; const handleKeyPress = (e: React.KeyboardEvent) => { @@ -124,7 +123,6 @@ export default function ChatInterface() { {/* Settings panel — collapsible */} {showSettings && (
-
+ {selectedKnowledgeBases.length > 0 && ( +
+ + 知识库检索模式 +
+ )} 0 && "pt-4", "bg-muted/40 border border-border/30", "placeholder:text-muted-foreground/60", "focus:outline-none focus:ring-2 focus:ring-primary/20 focus:border-primary/30", diff --git a/web/src/store/chat.ts b/web/src/store/chat.ts index fdb3495..ce4db14 100644 --- a/web/src/store/chat.ts +++ b/web/src/store/chat.ts @@ -54,14 +54,12 @@ export const useChatStore = create((set, get) => ({ // 加载会话列表 loadSessions: async () => { - set({ isLoading: true, error: null }); - try { const sessions = await chatAPI.getSessions(); - set({ sessions, isLoading: false }); + set({ sessions }); } catch (error) { const errorMessage = error instanceof Error ? error.message : "加载会话失败"; - set({ error: errorMessage, isLoading: false }); + set({ error: errorMessage }); } }, @@ -320,7 +318,7 @@ export const useChatStore = create((set, get) => ({ }); } }, - (sessionId: number, messageId?: number, userMessageId?: number) => { + (_sessionId: number, messageId?: number, userMessageId?: number) => { if (messageId) { set((state) => ({ messages: state.messages.map(msg => { @@ -332,15 +330,9 @@ export const useChatStore = create((set, get) => ({ abortController: null, })); } else { - const { currentSession } = get(); - if (currentSession) { - get().loadMessages(currentSession.id); - } + // 前端已通过流式获得完整内容，不需要从服务端重新加载 set({ isStreaming: false, abortController: null }); } - - // 刷新会话列表以确保新会话显示在顶部 - get().loadSessions(); }, (error: string) => { // 流式错误 @@ -414,36 +406,121 @@ export const useChatStore = create((set, get) => ({ } }, - // 重新生成消息（支持 assistant 消息 ID） + // 重新生成消息（流式） regenerateMessage: async (messageId: number, model?: string) => { + const { currentSession, messages } = get(); + if (!currentSession) return; + + const messageIndex = messages.findIndex(msg => msg.id === messageId); + if (messageIndex === -1) return; + + // 找到该 assistant 消息之前的 user 消息内容 + const userContent = messages + .slice(0, messageIndex) + .reverse() + .find(msg => msg.role === "user")?.content; + if (!userContent) return; + + // 删除该 assistant 消息及之后的所有消息，替换为流式占位符 + const trimmed = messages.slice(0, messageIndex); + const assistantPlaceholder: ChatMessage = { + id: Date.now() + 1, + role: "assistant", + content: "", + created_at: new Date().toISOString(), + }; + + const abortController = new AbortController(); + set({ + messages: [...trimmed, assistantPlaceholder], + isStreaming: true, + error: null, + abortController, + }); + + // 判断模式：检查之前的消息中是否有知识库相关内容 + const mode = "normal"; + + let thinkingSteps: any[] = []; + try { - set({ isLoading: true, error: null }); - const response = await chatAPI.regenerateMessage(messageId, model); - - // 找到该消息，删除它及之后的所有消息 - const messageIndex = get().messages.findIndex(msg => msg.id === messageId); - if (messageIndex !== -1) { - set((state) => ({ - messages: state.messages.slice(0, messageIndex), - })); - } - - // 添加新的助手回复 - const newMessage: ChatMessage = { - id: response.new_message_id, - role: "assistant", - content: response.content, - created_at: new Date().toISOString(), - }; - - set((state) => ({ - messages: [...state.messages, newMessage], - isLoading: false, - })); + await chatAPI.streamMessage( + userContent, + currentSession.id, + mode, + undefined, // knowledgeBaseIds + abortController.signal, + (chunk: string) => { + // 处理 chunk + try { + const data = JSON.parse(chunk); + if (data.type === 'thinking') { + thinkingSteps.push({ + stage: data.stage, + message: data.message, + doc_count: data.doc_count, + time: data.time + }); + requestAnimationFrame(() => { + set((state) => ({ + messages: state.messages.map(msg => + msg.id === assistantPlaceholder.id + ? { ...msg, thinking: [...thinkingSteps] } + : msg + ), + })); + }); + } else if (data.type === 'chunk') { + requestAnimationFrame(() => { + set((state) => ({ + messages: state.messages.map(msg => + msg.id === assistantPlaceholder.id + ? { ...msg, content: msg.content + data.content } + : msg + ), + })); + }); + } + } catch { + // 纯文本 chunk + requestAnimationFrame(() => { + set((state) => ({ + messages: state.messages.map(msg => + msg.id === assistantPlaceholder.id + ? { ...msg, content: msg.content + chunk } + : msg + ), + })); + }); + } + }, + (_sessionId: number, messageId?: number) => { + // onComplete: 只替换 assistant 占位符 ID，不重新加载 + if (messageId) { + set((state) => ({ + messages: state.messages.map(msg => + msg.id === assistantPlaceholder.id + ? { ...msg, id: messageId } + : msg + ), + isStreaming: false, + abortController: null, + })); + } else { + set({ isStreaming: false, abortController: null }); + } + }, + (error: string) => { + set({ error, isStreaming: false, abortController: null }); + }, + undefined, // onStatus + undefined, // onThinking + undefined, // onThinkingContent + model + ); } catch (error) { - console.error("重新生成消息失败:", error); - const errorMessage = error instanceof Error ? error.message : "重新生成消息失败"; - set({ error: errorMessage, isLoading: false }); + const errorMessage = error instanceof Error ? error.message : "重新生成失败"; + set({ error: errorMessage, isStreaming: false, abortController: null }); } },