✨ Add user-level isolation when select knowledge base

Author: ljy65535

backend/agents/create_agent_info.py

@@ -9,6 +9,7 @@
 from database.agent_db import search_agent_info_by_agent_id, search_tools_for_sub_agent, query_sub_agents, \
     query_or_create_main_agent_id
 from services.elasticsearch_service import ElasticSearchService
+from services.tenant_config_service import get_selected_knowledge_list
 from utils.config_utils import config_manager
 from utils.user_utils import get_user_info
 
@@ -45,13 +46,19 @@ async def create_agent_config(agent_id, tenant_id, user_id):
     system_prompt = agent_info.get("prompt")
 
     # special logic
-    for tool in tool_list:
-        if "KnowledgeBaseSearchTool" == tool.class_name:
-            knowledge_base_summary = "\n\n### 本地知识库信息 ###\n"
-            for knowledge_name in json.loads(config_manager.get_config("SELECTED_KB_NAMES", "[]")):
-                message = ElasticSearchService().get_summary(index_name=knowledge_name)
-                knowledge_base_summary += f"{knowledge_name}:{message['summary']}\n"
-            system_prompt += knowledge_base_summary
+    try:
+        for tool in tool_list:
+            if "KnowledgeBaseSearchTool" == tool.class_name:
+                knowledge_base_summary = "\n\n### 本地知识库信息 ###\n"
+
+                knowledge_info_list = get_selected_knowledge_list(tenant_id=tenant_id, user_id=user_id)
+                for knowledge_info in knowledge_info_list:
+                    knowledge_name = knowledge_info.get("index_name")
+                    message = ElasticSearchService().get_summary(index_name=knowledge_name)
+                    knowledge_base_summary += f"{knowledge_name}:{message['summary']}\n"
+                system_prompt += knowledge_base_summary
+    except Exception as e:
+        logger.error(f"add knowledge base summary to system prompt failed, error: {e}")
 
     agent_config = AgentConfig(
         name="" if agent_info["name"] is None else agent_info["name"],
@@ -82,7 +89,9 @@ async def create_tool_config_list(agent_id, tenant_id, user_id):
 
         # special logic for knowledge base search tool
         if tool_config.class_name == "KnowledgeBaseSearchTool":
-            tool_config.metadata = {"index_names": json.loads(config_manager.get_config("SELECTED_KB_NAMES", "[]"))}
+            knowledge_info_list = get_selected_knowledge_list(tenant_id=tenant_id, user_id=user_id)
+            index_names = [knowledge_info.get("index_name") for knowledge_info in knowledge_info_list]
+            tool_config.metadata = {"index_names": index_names}
         tool_config_list.append(tool_config)
     return tool_config_list
 

backend/apps/base_app.py

@@ -16,6 +16,7 @@
 from .tool_config_app import router as tool_config_router
 from .prompt_app import router as prompt_router
 from .knowledge_summary_app import router as summary_router
+from .tenant_config_app import router as tenant_config_router
 
 app = FastAPI(root_path="/api")
 
@@ -40,6 +41,7 @@
 app.include_router(tool_config_router)
 app.include_router(summary_router)
 app.include_router(prompt_router)
+app.include_router(tenant_config_router)
 
 # Global exception handler for HTTP exceptions
 @app.exception_handler(HTTPException)

backend/apps/config_sync_app.py

@@ -52,14 +52,6 @@ async def save_config(config: GlobalConfig):
                 env_key = f"{model_prefix}_DIMENSION"
                 env_config[env_key] = safe_value(model_config.get("dimension"))
 
-        # Process knowledge base configuration - use key names directly without prefix, store lists in JSON format
-        for key, value in config_dict.get("data", {}).items():
-            env_key = get_env_key(key)
-            if isinstance(value, list):
-                env_config[env_key] = safe_list(value)
-            else:
-                env_config[env_key] = safe_value(value)
-
         # Batch update environment variables
         for key, value in env_config.items():
             config_manager.set_config(key, value)
@@ -165,11 +157,6 @@ async def load_config():
                         "modelUrl": config_manager.get_config("TTS_MODEL_URL", "")
                     }
                 }
-            },
-            "data": {
-                "selectedKbNames": json.loads(config_manager.get_config("SELECTED_KB_NAMES", "[]")),
-                "selectedKbModels": json.loads(config_manager.get_config("SELECTED_KB_MODELS", "[]")),
-                "selectedKbSources": json.loads(config_manager.get_config("SELECTED_KB_SOURCES", "[]"))
             }
         }
 

backend/apps/knowledge_app.py

@@ -0,0 +1,61 @@
+from typing import Optional
+
+from fastapi import HTTPException, Query, Body, Path, Depends, APIRouter, Header
+from consts.model import ChangeSummaryRequest
+from fastapi.responses import StreamingResponse
+from nexent.vector_database.elasticsearch_core import ElasticSearchCore
+from services.elasticsearch_service import ElasticSearchService, get_es_core
+from database.utils import get_current_user_id
+router = APIRouter(prefix="/summary")
+
+@router.post("/{index_name}/auto_summary")
+async def auto_summary(
+            index_name: str = Path(..., description="Name of the index to get documents from"),
+            batch_size: int = Query(1000, description="Number of documents to retrieve per batch"),
+            es_core: ElasticSearchCore = Depends(get_es_core),
+            authorization: Optional[str] = Header(None)
+    ):
+    """Summary Elasticsearch index_name by model"""
+    try:
+        user_id = get_current_user_id(authorization)
+        service = ElasticSearchService()
+
+        return await service.summary_index_name(
+            index_name=index_name,
+            batch_size=batch_size,
+            es_core=es_core,
+            user_id=user_id
+        )
+    except Exception as e:
+        return StreamingResponse(
+            f"data: {{\"status\": \"error\", \"message\": \"知识库摘要生成失败: {e}\"}}\n\n",
+            media_type="text/event-stream",
+            status_code=500
+        )
+
+
+@router.post("/{index_name}/summary")
+def change_summary(
+            index_name: str = Path(..., description="Name of the index to get documents from"),
+            change_summary_request: ChangeSummaryRequest = Body(None, description="knowledge base summary"),
+            authorization: Optional[str] = Header(None)
+    ):
+    """Summary Elasticsearch index_name by user"""
+    try:
+        user_id = get_current_user_id(authorization)
+        summary_result = change_summary_request.summary_result
+        return ElasticSearchService().change_summary(index_name=index_name,summary_result=summary_result,user_id=user_id)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"知识库摘要更新失败: {str(e)}")
+
+
+@router.get("/{index_name}/summary")
+def get_summary(
+            index_name: str = Path(..., description="Name of the index to get documents from"),
+    ):
+    """Get Elasticsearch index_name Summary"""
+    try:
+        # 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)}")

backend/apps/tenant_config_app.py

@@ -0,0 +1,53 @@
+from fastapi import APIRouter, Depends, Header, Query, Path, Body
+from typing import Optional, List
+from services.tenant_config_service import get_selected_knowledge_list, update_selected_knowledge
+from fastapi.responses import JSONResponse
+
+from utils.user_utils import get_user_info
+
+router = APIRouter(prefix="/tenant_config")
+
+@router.get("/load_knowledge_list")
+def load_knowledge_list(
+    authorization: Optional[str] = Header(None)
+):
+    try:
+        user_id, tenant_id = get_user_info()
+        selected_knowledge_info = get_selected_knowledge_list(tenant_id=tenant_id, user_id=user_id)
+        content = {"selectedKbNames": [item["index_name"] for item in selected_knowledge_info],
+                   "selectedKbModels": [item["knowledge_embedding_model"] for item in selected_knowledge_info],
+                   "selectedKbSources": [item["knowledge_sources"] for item in selected_knowledge_info]}
+
+        return JSONResponse(
+            status_code=200,
+            content={"content": content, "status": "success"}
+        )
+    except Exception as e:
+        return JSONResponse(
+            status_code=400,
+            content={"message": f"Failed to load configuration: {str(e)}", "status": "error"}
+        )
+
+@router.post("/update_knowledge_list")
+def update_knowledge_list(
+    authorization: Optional[str] = Header(None),
+    knowledge_list: List[str] = Body(None)
+):
+    try:
+        user_id, tenant_id = get_user_info()
+        result = update_selected_knowledge(tenant_id=tenant_id, user_id=user_id, index_name_list=knowledge_list)
+        if result:
+            return JSONResponse(
+                status_code=200,
+                content={"message": "update success", "status": "success"}
+            )
+        else:
+            return JSONResponse(
+                status_code=400,
+                content={"message": "update failed", "status": "error"}
+        )
+    except Exception as e:
+        return JSONResponse(
+            status_code=400,
+            content={"message": f"Failed to update configuration: {str(e)}", "status": "error"}
+        )

backend/consts/model.py

@@ -84,16 +84,9 @@ class AppConfig(BaseModel):
     avatarUri: Optional[str] = None
 
 
-class KnowledgeBaseConfig(BaseModel):
-    selectedKbNames: List[str]
-    selectedKbModels: List[str]
-    selectedKbSources: List[str]
-
-
 class GlobalConfig(BaseModel):
     app: AppConfig
     models: ModelConfig
-    data: KnowledgeBaseConfig
 
 
 # Request models

backend/database/db_models.py

@@ -215,9 +215,31 @@ class KnowledgeRecord(TableBase):
     knowledge_id = Column(Integer, Sequence("knowledge_record_t_knowledge_id_seq", schema="nexent"), primary_key=True, nullable=False, doc="Knowledge base ID, unique primary key")
     index_name = Column(String(100), doc="Knowledge base name")
     knowledge_describe = Column(String(300), doc="Knowledge base description")
+    knowledge_embedding_model = Column(String(300), doc="Knowledge base embedding model")
+    knowledge_sources = Column(String(300), doc="Knowledge base sources")
     tenant_id = Column(String(100), doc="Tenant ID")
     delete_flag = Column(String(1), default="N", doc="Knowledge base status. Currently defaults to 1, if knowledge base status is 0, then this knowledge base is unavailable")
     create_time = Column(TIMESTAMP(timezone=False), server_default=func.now(), doc="Creation time, audit field")
     update_time = Column(TIMESTAMP(timezone=False), server_default=func.now(), doc="Update date, audit field")
     updated_by = Column(String(100), doc="ID of the last updater, audit field")
-    created_by = Column(String(100), doc="ID of the creator, audit field")
+    created_by = Column(String(100), doc="ID of the creator, audit field")
+
+
+class TenantConfig(TableBase):
+    """
+    Tenant configuration information table
+    """
+    __tablename__ = "tenant_config_t"
+    __table_args__ = {"schema": SCHEMA}
+
+    tenant_config_id = Column(Integer, Sequence("tenant_config_t_tenant_config_id_seq", schema=SCHEMA), primary_key=True, nullable=False, doc="ID")
+    tenant_id = Column(String(100), doc="Tenant ID")
+    user_id = Column(String(100), doc="User ID")
+    value_type = Column(String(100), doc=" the data type of config_value, optional values: single/multi", default="single")
+    config_key = Column(String(100), doc="the key of the config")
+    config_value = Column(String(100), doc="the value of the config")
+    create_time = Column(TIMESTAMP(timezone=False), server_default=func.now(), doc="Creation time")
+    update_time = Column(TIMESTAMP(timezone=False), server_default=func.now(), doc="Update time")
+    created_by = Column(String(100), doc="Creator")
+    updated_by = Column(String(100), doc="Updater")
+    delete_flag = Column(String(1), default="N", doc="Whether it is deleted. Optional values: Y/N")

backend/database/knowledge_db.py

@@ -1,4 +1,4 @@
-from typing import Dict, Any, Optional
+from typing import Dict, Any, Optional, List
 from sqlalchemy import func
 from sqlalchemy.exc import SQLAlchemyError
 
@@ -27,6 +27,8 @@ def create_knowledge_record(query: Dict[str, Any]) -> int:
                 "knowledge_describe": query.get("knowledge_describe", ""),
                 "created_by": query.get("user_id"),
                 "updated_by": query.get("user_id"),
+                "knowledge_embedding_model": query.get("knowledge_embedding_model"),
+                "knowledge_sources": query.get("knowledge_sources", "elasticsearch")
             }
 
             # Create new record
@@ -132,3 +134,34 @@ def get_knowledge_record(query: Optional[Dict[str, Any]] = None) -> Dict[str, An
             return {}
     except SQLAlchemyError as e:
         raise e
+
+def get_knowledge_info_by_knowledge_ids(knowledge_ids: List[str]) -> List[Dict[str, Any]]:
+    try:
+        with get_db_session() as session:
+            result = session.query(KnowledgeRecord).filter(
+                KnowledgeRecord.knowledge_id.in_(knowledge_ids),
+                KnowledgeRecord.delete_flag != 'Y'
+            ).all()
+            knowledge_info = []
+            for item in result:
+                knowledge_info.append({
+                    "knowledge_id": item.knowledge_id,
+                    "index_name": item.index_name,
+                    "knowledge_embedding_model": item.knowledge_embedding_model,
+                    "knowledge_sources": item.knowledge_sources
+                })
+            return knowledge_info
+    except SQLAlchemyError as e:
+        raise e
+
+def get_knowledge_ids_by_index_names(index_names: List[str]) -> List[str]:
+    try:
+        with get_db_session() as session:
+            result = session.query(KnowledgeRecord.knowledge_id).filter(
+                KnowledgeRecord.index_name.in_(index_names),
+                KnowledgeRecord.delete_flag != 'Y'
+            ).all()
+            return [item.knowledge_id for item in result]
+    except SQLAlchemyError as e:
+        raise e
+    

backend/database/tenant_config_db.py

@@ -0,0 +1,53 @@
+from typing import Optional, Dict, List, Any
+
+from .client import db_client, get_db_session
+from .db_models import TenantConfig
+
+
+def get_tenant_config_info(tenant_id: str, user_id: str, select_key: str):
+    with get_db_session() as session:
+        result = session.query(TenantConfig).filter(TenantConfig.tenant_id == tenant_id,
+                                                    TenantConfig.user_id == user_id,
+                                                    TenantConfig.config_key == select_key,
+                                                    TenantConfig.delete_flag == "N").all()
+        record_info = []
+        for item in result:
+            record_info.append({
+                "config_value": item.config_value,
+                "tenant_config_id": item.tenant_config_id
+            })
+        return record_info
+
+
+def insert_config(insert_data: Dict[str, Any]):
+    with get_db_session() as session:
+        try:
+            session.add(TenantConfig(**insert_data))
+            session.commit()
+            return True
+        except Exception as e: 
+            session.rollback()
+            return False
+
+
+def delete_config_by_tenant_config_id(tenant_config_id: int):
+    with get_db_session() as session:
+        try:
+            session.query(TenantConfig).filter(TenantConfig.tenant_config_id == tenant_config_id,
+                                               TenantConfig.delete_flag == "N").update({"delete_flag": "Y"})
+            session.commit()
+            return True
+        except Exception as e:
+            session.rollback()
+            return False
+
+def update_config_by_tenant_config_id(tenant_config_id: int, update_value: str):
+    with get_db_session() as session:
+        try:
+            session.query(TenantConfig).filter(TenantConfig.tenant_config_id == tenant_config_id,
+                                               TenantConfig.delete_flag == "N").update({"config_value": update_value})
+            session.commit()
+            return True
+        except Exception as e:
+            session.rollback()
+            return False

backend/database/utils.py

@@ -5,4 +5,4 @@
 
 def get_current_user_id(authorization: Optional[str] = Header(None)) -> Optional[str]:
     if not authorization:
-        return None
+        return ""