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

Author: Phinease

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

@@ -13,14 +13,18 @@
 from typing import Dict, List, Optional, Tuple
 
 import numpy as np
-import yaml
 from jinja2 import Template, StrictUndefined
 from nexent.vector_database.base import VectorDatabaseCore
 from sklearn.cluster import KMeans
 from sklearn.metrics import silhouette_score
 from sklearn.metrics.pairwise import cosine_similarity
 
 from consts.const import LANGUAGE
+from utils.prompt_template_utils import (
+    get_document_summary_prompt_template,
+    get_cluster_summary_reduce_prompt_template,
+    get_cluster_summary_agent_prompt_template
+)
 
 logger = logging.getLogger("document_vector_utils")
 
@@ -548,14 +552,8 @@ def summarize_document(document_content: str, filename: str, language: str = LAN
         Document summary text
     """
     try:
-        # Select prompt file based on language
-        if language == LANGUAGE["ZH"]:
-            prompt_path = 'backend/prompts/document_summary_agent_zh.yaml'
-        else:
-            prompt_path = 'backend/prompts/document_summary_agent.yaml'
-        
-        with open(prompt_path, 'r', encoding='utf-8') as f:
-            prompts = yaml.safe_load(f)
+        # Get prompt template from prompt_template_utils
+        prompts = get_document_summary_prompt_template(language)
 
         system_prompt = prompts.get('system_prompt', '')
         user_prompt_template = prompts.get('user_prompt', '')
@@ -626,14 +624,8 @@ def summarize_cluster(document_summaries: List[str], language: str = LANGUAGE["Z
         Cluster summary text
     """
     try:
-        # Select prompt file based on language
-        if language == LANGUAGE["ZH"]:
-            prompt_path = 'backend/prompts/cluster_summary_reduce_zh.yaml'
-        else:
-            prompt_path = 'backend/prompts/cluster_summary_reduce.yaml'
-        
-        with open(prompt_path, 'r', encoding='utf-8') as f:
-            prompts = yaml.safe_load(f)
+        # Get prompt template from prompt_template_utils
+        prompts = get_cluster_summary_reduce_prompt_template(language)
 
         system_prompt = prompts.get('system_prompt', '')
         user_prompt_template = prompts.get('user_prompt', '')
@@ -938,9 +930,8 @@ def summarize_cluster_legacy(cluster_content: str, language: str = LANGUAGE["ZH"
         Cluster summary text
     """
     try:
-        prompt_path = 'backend/prompts/cluster_summary_agent.yaml'
-        with open(prompt_path, 'r', encoding='utf-8') as f:
-            prompts = yaml.safe_load(f)
+        # Get prompt template from prompt_template_utils
+        prompts = get_cluster_summary_agent_prompt_template(language)
 
         system_prompt = prompts.get('system_prompt', '')
         user_prompt_template = prompts.get('user_prompt', '')

backend/utils/prompt_template_utils.py

@@ -21,6 +21,9 @@ def get_prompt_template(template_type: str, language: str = LANGUAGE["ZH"], **kw
             - 'analyze_file': File analysis template
             - 'generate_title': Title generation template
             - 'file_processing_messages': File processing messages template
+            - 'document_summary': Document summary template (Map stage)
+            - 'cluster_summary_reduce': Cluster summary reduce template (Reduce stage)
+            - 'cluster_summary_agent': Cluster summary agent template (legacy)
         language: Language code ('zh' or 'en')
         **kwargs: Additional parameters, for agent type need to pass is_manager parameter
 
@@ -61,6 +64,18 @@ def get_prompt_template(template_type: str, language: str = LANGUAGE["ZH"], **kw
         'file_processing_messages': {
             LANGUAGE["ZH"]: 'backend/prompts/utils/file_processing_messages.yaml',
             LANGUAGE["EN"]: 'backend/prompts/utils/file_processing_messages_en.yaml'
+        },
+        'document_summary': {
+            LANGUAGE["ZH"]: 'backend/prompts/document_summary_agent_zh.yaml',
+            LANGUAGE["EN"]: 'backend/prompts/document_summary_agent.yaml'
+        },
+        'cluster_summary_reduce': {
+            LANGUAGE["ZH"]: 'backend/prompts/cluster_summary_reduce_zh.yaml',
+            LANGUAGE["EN"]: 'backend/prompts/cluster_summary_reduce.yaml'
+        },
+        'cluster_summary_agent': {
+            LANGUAGE["ZH"]: 'backend/prompts/cluster_summary_agent.yaml',
+            LANGUAGE["EN"]: 'backend/prompts/cluster_summary_agent.yaml'
         }
     }
 
@@ -164,3 +179,42 @@ def get_file_processing_messages_template(language: str = 'zh') -> Dict[str, Any
         dict: Loaded file processing messages configuration
     """
     return get_prompt_template('file_processing_messages', language)
+
+
+def get_document_summary_prompt_template(language: str = LANGUAGE["ZH"]) -> Dict[str, Any]:
+    """
+    Get document summary prompt template (Map stage)
+
+    Args:
+        language: Language code ('zh' or 'en')
+
+    Returns:
+        dict: Loaded document summary prompt template configuration
+    """
+    return get_prompt_template('document_summary', language)
+
+
+def get_cluster_summary_reduce_prompt_template(language: str = LANGUAGE["ZH"]) -> Dict[str, Any]:
+    """
+    Get cluster summary reduce prompt template (Reduce stage)
+
+    Args:
+        language: Language code ('zh' or 'en')
+
+    Returns:
+        dict: Loaded cluster summary reduce prompt template configuration
+    """
+    return get_prompt_template('cluster_summary_reduce', language)
+
+
+def get_cluster_summary_agent_prompt_template(language: str = LANGUAGE["ZH"]) -> Dict[str, Any]:
+    """
+    Get cluster summary agent prompt template (legacy)
+
+    Args:
+        language: Language code ('zh' or 'en')
+
+    Returns:
+        dict: Loaded cluster summary agent prompt template configuration
+    """
+    return get_prompt_template('cluster_summary_agent', language)