✨Knowledge_base summary: frontend streaming output

Author: linsensen222

backend/apps/knowledge_summary_app.py

@@ -29,7 +29,7 @@ async def auto_summary(
         )
     except Exception as e:
         async def generate_error():
-            yield f"data: {json.dumps({'status': 'error', 'message': str(e)})}\n\n"
+            yield f"data: {{\"status\": \"error\", \"message\": \"知识库摘要生成失败: {e}\"}}\n\n"
         return StreamingResponse(
             generate_error(),
             media_type="text/event-stream",
@@ -51,7 +51,7 @@ def change_summary(
         # Try to list indices as a health check
         return ElasticSearchService().change_summary(index_name=index_name,summary_result=summary_result, es_core=es_core,user_id=user_id)
     except Exception as e:
-        raise HTTPException(status_code=500, detail=f"{str(e)}")
+        raise HTTPException(status_code=500, detail=f"知识库摘要更新失败: {str(e)}")
 
 
 @router.get("/{index_name}/summary")
@@ -63,4 +63,4 @@ def get_summary(
         # Try to list indices as a health check
         return ElasticSearchService().get_summary(index_name=index_name)
     except Exception as e:
-        raise HTTPException(status_code=500, detail=f"{str(e)}")
+        raise HTTPException(status_code=500, detail=f"获取知识库摘要失败: {str(e)}")

backend/services/elasticsearch_service.py

@@ -8,7 +8,8 @@
 3. Support for multiple search methods: exact search, semantic search, and hybrid search
 4. Health check interface
 """
-
+import asyncio
+import logging
 import os
 import time
 from typing import Optional
@@ -632,23 +633,24 @@ async def generate_summary():
                 token_join = []
                 try:
                     for new_token in generate_knowledge_summary_stream(keywords_for_summary):
-                        print(new_token)
-                        token_join.append(new_token)
-                        yield f"data: {{\"status\": \"success\", \"message\": \"Index {index_name} summary successfully\", \"summary\": \"{new_token}\"}}\n\n"
-
-                    if new_token == "END":
-                        model_output = "".join(token_join)
-                        # updata sql
-                        knowledge_record = get_knowledge_by_name(index_name)
-                        if knowledge_record:
-                            update_data = {
-                                "knowledge_describe": model_output,
-                                "updated_by": user_id,
-                            }
-                            update_knowledge_record(knowledge_record["knowledge_id"], update_data)
+                        if new_token == "END":
+                            model_output = "".join(token_join)
+                            knowledge_record = get_knowledge_by_name(index_name)
+                            if knowledge_record:
+                                update_data = {
+                                    "knowledge_describe": model_output,
+                                    "updated_by": user_id,
+                                }
+                                update_knowledge_record(knowledge_record["knowledge_id"], update_data)
+                            break
+                        else:
+                            token_join.append(new_token)
+                            logging.info(new_token)
+                            yield f"data: {{\"status\": \"success\", \"message\": \"{new_token}\"}}\n\n"
+                        await asyncio.sleep(0.1)
 
                 except Exception as e:
-                    yield f"data: {{\"status\": \"error\", \"message\": \"{str(e)}\"}}\n\n"
+                    yield f"data: {{\"status\": \"error\", \"message\": \"{e}\"}}\n\n"
 
             # Return the flow response
             return StreamingResponse(
@@ -778,10 +780,10 @@ def get_summary(self,
             knowledge_record = get_knowledge_by_name(index_name)
             if knowledge_record:
                 summary_result = knowledge_record["knowledge_describe"]
-                return {"status": "success", "message": f"Index {index_name} summary successfully", "summary": summary_result}
+                return {"status": "success", "message": f"索引 {index_name} 摘要获取成功", "summary": summary_result}
             raise HTTPException(
                 status_code=500,
-                detail=f"Failed to get index {index_name} summary"
+                detail=f"无法获取索引 {index_name} 的摘要"
             )
         except Exception as e:
-            raise HTTPException(status_code=500, detail=f"{str(e)}")
+            raise HTTPException(status_code=500, detail=f"获取摘要失败: {str(e)}")

frontend/app/setup/knowledgeBaseSetup/document/DocumentListLayout.tsx

@@ -124,7 +124,7 @@ const DocumentListLayout: React.FC<DocumentListLayoutProps> = ({
   const [summary, setSummary] = useState('');
   const [isSummarizing, setIsSummarizing] = useState(false);
   const [isSaving, setIsSaving] = useState(false);
-  const { summaryIndex } = useKnowledgeBaseContext();
+  const { } = useKnowledgeBaseContext();
 
   // Reset showDetail state when knowledge base name changes
   React.useEffect(() => {
@@ -155,14 +155,19 @@ const DocumentListLayout: React.FC<DocumentListLayoutProps> = ({
     }
 
     setIsSummarizing(true);
+    // 清空现有的摘要，准备流式更新
+    setSummary('');
+    
     try {
-      const result = await summaryIndex(knowledgeBaseName, 1000);
-      if (result) {
-        setSummary(result);
-        message.success('知识库总结完成');
-      } else {
-        message.warning('知识库总结为空');
-      }
+      // 使用带有进度回调的summaryIndex方法
+      await knowledgeBaseService.summaryIndex(
+        knowledgeBaseName, 
+        1000,
+        (newText) => {
+          setSummary(prev => prev + newText);
+        }
+      );
+      message.success('知识库总结完成');
     } catch (error) {
       message.error('获取知识库总结失败');
       console.error('获取知识库总结失败:', error);
@@ -281,35 +286,19 @@ const DocumentListLayout: React.FC<DocumentListLayoutProps> = ({
                 自动总结
               </Button>
             </div>
-            {isSummarizing ? (
-              <div style={{
+            <Input.TextArea
+              value={summary}
+              onChange={(e) => setSummary(e.target.value)}
+              style={{
                 flex: 1,
-                display: 'flex',
-                flexDirection: 'column',
-                justifyContent: 'center',
-                alignItems: 'center',
-                backgroundColor: '#fafafa',
-                borderRadius: '8px',
-                marginBottom: 20
-              }}>
-                <div className="animate-spin rounded-full h-12 w-12 border-b-2 border-blue-500 mb-4"></div>
-                <p style={{ fontSize: 16, color: '#666' }}>正在总结知识库内容，请稍候...</p>
-              </div>
-            ) : (
-              <Input.TextArea
-                value={summary}
-                onChange={(e) => setSummary(e.target.value)}
-                style={{
-                  flex: 1,
-                  minHeight: 0,
-                  marginBottom: 20,
-                  resize: 'none',
-                  fontSize: 18,
-                  lineHeight: 1.7,
-                  padding: 20
-                }}
-              />
-            )}
+                minHeight: 0,
+                marginBottom: 20,
+                resize: 'none',
+                fontSize: 18,
+                lineHeight: 1.7,
+                padding: 20
+              }}
+            />
             <div style={{ display: 'flex', gap: 12, justifyContent: 'flex-end' }}>
               <Button
                 type="primary"

frontend/services/knowledgeBaseService.ts

@@ -341,7 +341,7 @@ class KnowledgeBaseService {
   }
 
   // Summary index content
-  async summaryIndex(indexName: string, batchSize: number = 1000): Promise<string> {
+  async summaryIndex(indexName: string, batchSize: number = 1000, onProgress?: (text: string) => void): Promise<string> {
     try {
       const response = await fetch(API_ENDPOINTS.knowledgeBase.summary(indexName) + `?batch_size=${batchSize}`, {
         method: 'POST',
@@ -352,36 +352,50 @@ class KnowledgeBaseService {
         throw new Error(`HTTP error! status: ${response.status}`);
       }
 
-      // 创建 EventSource 来处理流式响应
-      const eventSource = new EventSource(API_ENDPOINTS.knowledgeBase.summary(indexName) + `?batch_size=${batchSize}`);
-      
-      return new Promise((resolve, reject) => {
-        let summary = '';
+      if (!response.body) {
+        throw new Error('Response body is null');
+      }
+
+      // 处理流式响应
+      const reader = response.body.getReader();
+      const decoder = new TextDecoder('utf-8');
+      let summary = '';
+
+      while (true) {
+        const { done, value } = await reader.read();
+        if (done) break;
+        
+        // 解码二进制数据为文本
+        const chunk = decoder.decode(value, { stream: true });
 
-        eventSource.onmessage = (event) => {
-          const data = JSON.parse(event.data);
-          
-          switch(data.status) {
-            case 'processing':
-              console.log('Processing:', data.message);
-              break;
-            case 'success':
-              summary = data.summary;
-              eventSource.close();
-              resolve(summary);
-              break;
-            case 'error':
-              eventSource.close();
-              reject(new Error(data.message));
-              break;
+        // 处理SSE格式的数据
+        const lines = chunk.split('\n\n');
+        for (const line of lines) {
+          if (line.trim().startsWith('data:')) {
+            try {
+              // 提取JSON数据
+              const jsonStr = line.substring(line.indexOf('{'));
+              const data = JSON.parse(jsonStr);
+              
+              if (data.status === 'success') {
+                // 累加消息部分到摘要
+                summary += data.message;
+                
+                // 如果提供了进度回调，则调用它
+                if (onProgress) {
+                  onProgress(data.message);
+                }
+              } else if (data.status === 'error') {
+                throw new Error(data.message);
+              }
+            } catch (e) {
+              console.error('解析SSE数据失败:', e, line);
+            }
           }
-        };
-
-        eventSource.onerror = (error) => {
-          eventSource.close();
-          reject(new Error('EventSource failed'));
-        };
-      });
+        }
+      }
+      
+      return summary;
     } catch (error) {
       console.error('Error summarizing index:', error);
       throw error;