🐛 When the knowledge base is automatically summarized, other interfaces will be pended.

[Specification Details] 1. Start the thread pool to execute k-means and LLM calls in automatic summarization.
2. Concatenate prompt path to absolute path.

Author: YehongPan

backend/services/vectordatabase_service.py

@@ -741,6 +741,7 @@ async def summary_index_name(self,
             index_name: Name of the index to summarize
             batch_size: Number of documents to sample (default: 1000)
             vdb_core: VectorDatabaseCore instance
+            user_id: ID of the user delete the knowledge base
             tenant_id: ID of the tenant
             language: Language of the summary (default: 'zh')
             model_id: Model ID for LLM summarization
@@ -762,32 +763,45 @@ async def summary_index_name(self,
             # Use new Map-Reduce approach
             sample_count = min(batch_size // 5, 200)  # Sample reasonable number of documents
 
-            # Step 1: Get documents and calculate embeddings
-            document_samples, doc_embeddings = process_documents_for_clustering(
-                index_name=index_name,
-                vdb_core=vdb_core,
-                sample_doc_count=sample_count
-            )
-            
-            if not document_samples:
-                raise Exception("No documents found in index.")
-            
-            # Step 2: Cluster documents
-            clusters = kmeans_cluster_documents(doc_embeddings, k=None)
-            
-            # Step 3: Map-Reduce summarization
-            cluster_summaries = summarize_clusters_map_reduce(
-                document_samples=document_samples,
-                clusters=clusters,
-                language=language,
-                doc_max_words=100,
-                cluster_max_words=150,
-                model_id=model_id,
-                tenant_id=tenant_id
-            )
+            # Define a helper function to run all blocking operations in a thread pool
+            def _generate_summary_sync():
+                """Synchronous function that performs all blocking operations"""
+                # Step 1: Get documents and calculate embeddings
+                document_samples, doc_embeddings = process_documents_for_clustering(
+                    index_name=index_name,
+                    vdb_core=vdb_core,
+                    sample_doc_count=sample_count
+                )
+                
+                if not document_samples:
+                    raise Exception("No documents found in index.")
+                
+                # Step 2: Cluster documents (CPU-intensive operation)
+                clusters = kmeans_cluster_documents(doc_embeddings, k=None)
+                
+                # Step 3: Map-Reduce summarization (contains blocking LLM calls)
+                cluster_summaries = summarize_clusters_map_reduce(
+                    document_samples=document_samples,
+                    clusters=clusters,
+                    language=language,
+                    doc_max_words=100,
+                    cluster_max_words=150,
+                    model_id=model_id,
+                    tenant_id=tenant_id
+                )
+                
+                # Step 4: Merge into final summary
+                final_summary = merge_cluster_summaries(cluster_summaries)
+                return final_summary
 
-            # Step 4: Merge into final summary
-            final_summary = merge_cluster_summaries(cluster_summaries)
+            # Run blocking operations in a thread pool to avoid blocking the event loop
+            # Use get_running_loop() for better compatibility with modern asyncio
+            try:
+                loop = asyncio.get_running_loop()
+            except RuntimeError:
+                # Fallback for edge cases
+                loop = asyncio.get_event_loop()
+            final_summary = await loop.run_in_executor(None, _generate_summary_sync)
 
             # Stream the result
             async def generate_summary():

backend/utils/document_vector_utils.py

@@ -9,6 +9,7 @@
 4. Cluster summarization
 """
 import logging
+import os
 import random
 from typing import Dict, List, Optional, Tuple
 
@@ -25,6 +26,24 @@
 logger = logging.getLogger("document_vector_utils")
 
 
+def _get_prompt_absolute_path(relative_path: str) -> str:
+    """
+    Get absolute path for prompt files.
+    
+    Args:
+        relative_path: Relative path like 'backend/prompts/xxx.yaml'
+        
+    Returns:
+        Absolute path to the prompt file
+    """
+    # Get the directory of this file and construct absolute path
+    current_dir = os.path.dirname(os.path.abspath(__file__))
+    # Go up one level from utils to backend, then use the template path
+    backend_dir = os.path.dirname(current_dir)
+    absolute_path = os.path.join(backend_dir, relative_path.replace('backend/', ''))
+    return absolute_path
+
+
 def get_documents_from_es(index_name: str, vdb_core: VectorDatabaseCore, sample_doc_count: int = 200) -> Dict[str, Dict]:
     """
     Get document samples from Elasticsearch, aggregated by path_or_url
@@ -550,9 +569,9 @@ def summarize_document(document_content: str, filename: str, language: str = LAN
     try:
         # Select prompt file based on language
         if language == LANGUAGE["ZH"]:
-            prompt_path = 'backend/prompts/document_summary_agent_zh.yaml'
+            prompt_path = _get_prompt_absolute_path('backend/prompts/document_summary_agent_zh.yaml')
         else:
-            prompt_path = 'backend/prompts/document_summary_agent.yaml'
+            prompt_path = _get_prompt_absolute_path('backend/prompts/document_summary_agent.yaml')
 
         with open(prompt_path, 'r', encoding='utf-8') as f:
             prompts = yaml.safe_load(f)
@@ -628,9 +647,9 @@ def summarize_cluster(document_summaries: List[str], language: str = LANGUAGE["Z
     try:
         # Select prompt file based on language
         if language == LANGUAGE["ZH"]:
-            prompt_path = 'backend/prompts/cluster_summary_reduce_zh.yaml'
+            prompt_path = _get_prompt_absolute_path('backend/prompts/cluster_summary_reduce_zh.yaml')
         else:
-            prompt_path = 'backend/prompts/cluster_summary_reduce.yaml'
+            prompt_path = _get_prompt_absolute_path('backend/prompts/cluster_summary_reduce.yaml')
 
         with open(prompt_path, 'r', encoding='utf-8') as f:
             prompts = yaml.safe_load(f)
@@ -938,7 +957,7 @@ def summarize_cluster_legacy(cluster_content: str, language: str = LANGUAGE["ZH"
         Cluster summary text
     """
     try:
-        prompt_path = 'backend/prompts/cluster_summary_agent.yaml'
+        prompt_path = _get_prompt_absolute_path('backend/prompts/cluster_summary_agent.yaml')
         with open(prompt_path, 'r', encoding='utf-8') as f:
             prompts = yaml.safe_load(f)