♻️ Refactor: refactor elasticsearch core to vectordb core

♻️ Refactor: refactor elasticsearch core to vectordb core

Author: Jasonxia007

backend/agents/create_agent_info.py

@@ -9,7 +9,11 @@
 from nexent.core.agents.agent_model import AgentRunInfo, ModelConfig, AgentConfig, ToolConfig
 from nexent.memory.memory_service import search_memory_in_levels
 
-from services.elasticsearch_service import ElasticSearchService, elastic_core, get_embedding_model
+from services.vectordatabase_service import (
+    ElasticSearchService,
+    get_vector_db_core,
+    get_embedding_model,
+)
 from services.tenant_config_service import get_selected_knowledge_list
 from services.remote_mcp_service import get_remote_mcp_server_list
 from services.memory_config_service import build_memory_context
@@ -227,9 +231,11 @@ async def create_tool_config_list(agent_id, tenant_id, user_id):
                 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,
-                                    "es_core": elastic_core,
-                                    "embedding_model": get_embedding_model(tenant_id=tenant_id)}
+            tool_config.metadata = {
+                "index_names": index_names,
+                "vdb_core": get_vector_db_core(),
+                "embedding_model": get_embedding_model(tenant_id=tenant_id),
+            }
         tool_config_list.append(tool_config)
 
     return tool_config_list

backend/apps/config_app.py

@@ -6,7 +6,7 @@
 
 from apps.agent_app import agent_config_router as agent_router
 from apps.config_sync_app import router as config_sync_router
-from apps.elasticsearch_app import router as elasticsearch_router
+from apps.vectordatabase_app import router as vectordatabase_router
 from apps.file_management_app import file_management_config_router as file_manager_router
 from apps.image_app import router as proxy_router
 from apps.knowledge_summary_app import router as summary_router
@@ -41,7 +41,7 @@
 app.include_router(model_manager_router)
 app.include_router(config_sync_router)
 app.include_router(agent_router)
-app.include_router(elasticsearch_router)
+app.include_router(vectordatabase_router)
 app.include_router(voice_router)
 app.include_router(file_manager_router)
 app.include_router(proxy_router)

backend/apps/knowledge_summary_app.py

@@ -3,10 +3,10 @@
 
 from fastapi import APIRouter, Body, Depends, Header, HTTPException, Path, Query, Request
 from fastapi.responses import StreamingResponse
-from nexent.vector_database.elasticsearch_core import ElasticSearchCore
+from nexent.vector_database.base import VectorDatabaseCore
 
 from consts.model import ChangeSummaryRequest
-from services.elasticsearch_service import ElasticSearchService, get_es_core
+from services.vectordatabase_service import ElasticSearchService, get_vector_db_core
 from utils.auth_utils import get_current_user_id, get_current_user_info
 
 router = APIRouter(prefix="/summary")
@@ -22,7 +22,7 @@ async def auto_summary(
             1000, description="Number of documents to retrieve per batch"),
         model_id: Optional[int] = Query(
             None, description="Model ID to use for summary generation"),
-        es_core: ElasticSearchCore = Depends(get_es_core),
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
         authorization: Optional[str] = Header(None)
 ):
     """Summary Elasticsearch index_name by model"""
@@ -34,15 +34,16 @@ async def auto_summary(
         return await service.summary_index_name(
             index_name=index_name,
             batch_size=batch_size,
-            es_core=es_core,
+            vdb_core=vdb_core,
             tenant_id=tenant_id,
             language=language,
             model_id=model_id
         )
     except Exception as e:
-        logger.error("Knowledge base summary generation failed", exc_info=True)
+        logger.error(
+            f"Knowledge base summary generation failed: {e}", exc_info=True)
         return StreamingResponse(
-            f"data: {{\"status\": \"error\", \"message\": \"Knowledge base summary generation failed due to an internal error.\"}}\n\n",
+            "data: {{\"status\": \"error\", \"message\": \"Knowledge base summary generation failed due to an internal error.\"}}\n\n",
             media_type="text/event-stream",
             status_code=500
         )

backend/apps/elasticsearch_app.py backend/apps/vectordatabase_app.pybackend/apps/elasticsearch_app.py renamed to backend/apps/vectordatabase_app.py

@@ -5,27 +5,31 @@
 from fastapi import APIRouter, Body, Depends, Header, HTTPException, Path, Query
 
 from consts.model import IndexingResponse
-from nexent.vector_database.elasticsearch_core import ElasticSearchCore
-from services.elasticsearch_service import ElasticSearchService, get_embedding_model, get_es_core, \
-    check_knowledge_base_exist_impl
+from nexent.vector_database.base import VectorDatabaseCore
+from services.vectordatabase_service import (
+    ElasticSearchService,
+    get_embedding_model,
+    get_vector_db_core,
+    check_knowledge_base_exist_impl,
+)
 from services.redis_service import get_redis_service
 from utils.auth_utils import get_current_user_id
 
 router = APIRouter(prefix="/indices")
 service = ElasticSearchService()
-logger = logging.getLogger("elasticsearch_app")
+logger = logging.getLogger("vectordatabase_app")
 
 
 @router.get("/check_exist/{index_name}")
 async def check_knowledge_base_exist(
         index_name: str = Path(..., description="Name of the index to check"),
-        es_core: ElasticSearchCore = Depends(get_es_core),
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
         authorization: Optional[str] = Header(None)
 ):
     """Check if a knowledge base name exists and in which scope."""
     try:
         user_id, tenant_id = get_current_user_id(authorization)
-        return check_knowledge_base_exist_impl(index_name=index_name, es_core=es_core, user_id=user_id, tenant_id=tenant_id)
+        return check_knowledge_base_exist_impl(index_name=index_name, vdb_core=vdb_core, user_id=user_id, tenant_id=tenant_id)
     except Exception as e:
         logger.error(
             f"Error checking knowledge base existence for '{index_name}': {str(e)}", exc_info=True)
@@ -38,13 +42,13 @@ def create_new_index(
         index_name: str = Path(..., description="Name of the index to create"),
         embedding_dim: Optional[int] = Query(
             None, description="Dimension of the embedding vectors"),
-        es_core: ElasticSearchCore = Depends(get_es_core),
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
         authorization: Optional[str] = Header(None)
 ):
     """Create a new vector index and store it in the knowledge table"""
     try:
         user_id, tenant_id = get_current_user_id(authorization)
-        return ElasticSearchService.create_index(index_name, embedding_dim, es_core, user_id, tenant_id)
+        return ElasticSearchService.create_index(index_name, embedding_dim, vdb_core, user_id, tenant_id)
     except Exception as e:
         raise HTTPException(
             status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=f"Error creating index: {str(e)}")
@@ -53,15 +57,15 @@ def create_new_index(
 @router.delete("/{index_name}")
 async def delete_index(
         index_name: str = Path(..., description="Name of the index to delete"),
-        es_core: ElasticSearchCore = Depends(get_es_core),
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
         authorization: Optional[str] = Header(None)
 ):
     """Delete an index and all its related data by calling the centralized service."""
     logger.debug(f"Received request to delete knowledge base: {index_name}")
     try:
         user_id, tenant_id = get_current_user_id(authorization)
         # Call the centralized full deletion service
-        result = await ElasticSearchService.full_delete_knowledge_base(index_name, es_core, user_id)
+        result = await ElasticSearchService.full_delete_knowledge_base(index_name, vdb_core, user_id)
         return result
     except Exception as e:
         logger.error(
@@ -75,13 +79,13 @@ def get_list_indices(
         pattern: str = Query("*", description="Pattern to match index names"),
         include_stats: bool = Query(
             False, description="Whether to include index stats"),
-        es_core: ElasticSearchCore = Depends(get_es_core),
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
         authorization: Optional[str] = Header(None),
 ):
     """List all user indices with optional stats"""
     try:
         user_id, tenant_id = get_current_user_id(authorization)
-        return ElasticSearchService.list_indices(pattern, include_stats, tenant_id, user_id, es_core)
+        return ElasticSearchService.list_indices(pattern, include_stats, tenant_id, user_id, vdb_core)
     except Exception as e:
         raise HTTPException(
             status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=f"Error get index: {str(e)}")
@@ -93,7 +97,7 @@ def create_index_documents(
         index_name: str = Path(..., description="Name of the index"),
         data: List[Dict[str, Any]
                    ] = Body(..., description="Document List to process"),
-        es_core: ElasticSearchCore = Depends(get_es_core),
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
         authorization: Optional[str] = Header(None)
 ):
     """
@@ -103,7 +107,7 @@ def create_index_documents(
     try:
         user_id, tenant_id = get_current_user_id(authorization)
         embedding_model = get_embedding_model(tenant_id)
-        return ElasticSearchService.index_documents(embedding_model, index_name, data, es_core)
+        return ElasticSearchService.index_documents(embedding_model, index_name, data, vdb_core)
     except Exception as e:
         error_msg = str(e)
         logger.error(f"Error indexing documents: {error_msg}")
@@ -114,11 +118,11 @@ def create_index_documents(
 @router.get("/{index_name}/files")
 async def get_index_files(
         index_name: str = Path(..., description="Name of the index"),
-        es_core: ElasticSearchCore = Depends(get_es_core)
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core)
 ):
     """Get all files from an index, including those that are not yet stored in ES"""
     try:
-        result = await ElasticSearchService.list_files(index_name, include_chunks=False, es_core=es_core)
+        result = await ElasticSearchService.list_files(index_name, include_chunks=False, vdb_core=vdb_core)
         # Transform result to match frontend expectations
         return {
             "status": "success",
@@ -136,13 +140,13 @@ def delete_documents(
         index_name: str = Path(..., description="Name of the index"),
         path_or_url: str = Query(...,
                                  description="Path or URL of documents to delete"),
-        es_core: ElasticSearchCore = Depends(get_es_core)
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core)
 ):
     """Delete documents by path or URL and clean up related Redis records"""
     try:
         # First delete the documents using existing service
         result = ElasticSearchService.delete_documents(
-            index_name, path_or_url, es_core)
+            index_name, path_or_url, vdb_core)
 
         # Then clean up Redis records related to this specific document
         try:
@@ -184,10 +188,10 @@ def delete_documents(
 
 # Health check
 @router.get("/health")
-def health_check(es_core: ElasticSearchCore = Depends(get_es_core)):
+def health_check(vdb_core: VectorDatabaseCore = Depends(get_vector_db_core)):
     """Check API and Elasticsearch health"""
     try:
         # Try to list indices as a health check
-        return ElasticSearchService.health_check(es_core)
+        return ElasticSearchService.health_check(vdb_core)
     except Exception as e:
         raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=f"{str(e)}")

backend/consts/const.py

@@ -1,4 +1,5 @@
 import os
+from enum import Enum
 from dotenv import load_dotenv
 
 # Load environment variables
@@ -10,6 +11,11 @@
     os.path.dirname(__file__)), 'assets', 'test.wav')
 
 
+# Vector database providers
+class VectorDatabaseType(str, Enum):
+    ELASTICSEARCH = "elasticsearch"
+
+
 # ModelEngine Configuration
 MODEL_ENGINE_HOST = os.getenv('MODEL_ENGINE_HOST')
 MODEL_ENGINE_APIKEY = os.getenv('MODEL_ENGINE_APIKEY')

backend/services/file_management_service.py

@@ -20,7 +20,7 @@
     list_files
 )
 from utils.attachment_utils import convert_image_to_text, convert_long_text_to_text
-from services.elasticsearch_service import ElasticSearchService, get_es_core
+from services.vectordatabase_service import ElasticSearchService, get_vector_db_core
 from utils.prompt_template_utils import get_file_processing_messages_template
 from utils.file_management_utils import save_upload_file
 
@@ -79,8 +79,8 @@ async def upload_files_impl(destination: str, file: List[UploadFile], folder: st
         # Resolve filename conflicts against existing KB documents by renaming (e.g., name -> name_1)
         if index_name:
             try:
-                es_core = get_es_core()
-                existing = await ElasticSearchService.list_files(index_name, include_chunks=False, es_core=es_core)
+                vdb_core = get_vector_db_core()
+                existing = await ElasticSearchService.list_files(index_name, include_chunks=False, vdb_core=vdb_core)
                 existing_files = existing.get(
                     "files", []) if isinstance(existing, dict) else []
                 # Prefer 'file' field; fall back to 'filename' if present

backend/services/model_management_service.py

@@ -1,5 +1,5 @@
 import logging
-from typing import List, Dict, Any, Optional
+from typing import List, Dict, Any
 
 from consts.const import LOCALHOST_IP, LOCALHOST_NAME, DOCKER_INTERNAL_HOST
 from consts.model import ModelConnectStatusEnum
@@ -25,7 +25,7 @@
     sort_models_by_id,
 )
 from utils.memory_utils import build_memory_config as build_memory_config_for_tenant
-from services.elasticsearch_service import get_es_core
+from services.vectordatabase_service import get_vector_db_core
 from nexent.memory.memory_service import clear_model_memories
 
 logger = logging.getLogger("model_management_service")
@@ -244,12 +244,12 @@ async def delete_model_for_tenant(user_id: str, tenant_id: str, display_name: st
 
             # Best-effort memory cleanup using the fetched variants
             try:
-                es_core = get_es_core()
+                vdb_core = get_vector_db_core()
                 base_memory_config = build_memory_config_for_tenant(tenant_id)
                 for t, m in models_by_type.items():
                     try:
                         await clear_model_memories(
-                            es_core=es_core,
+                            vdb_core=vdb_core,
                             model_repo=m.get("model_repo", ""),
                             model_name=m.get("model_name", ""),
                             embedding_dims=int(m.get("max_tokens") or 0),

backend/services/tool_configuration_service.py

@@ -20,10 +20,10 @@
     query_all_tools,
     query_tool_instances_by_id,
     update_tool_table_from_scan_tool_list,
-    search_last_tool_instance_by_tool_id
+    search_last_tool_instance_by_tool_id,
 )
 from database.user_tenant_db import get_all_tenant_ids
-from services.elasticsearch_service import get_embedding_model, elastic_core
+from services.vectordatabase_service import get_embedding_model, get_vector_db_core
 from services.tenant_config_service import get_selected_knowledge_list
 
 logger = logging.getLogger("tool_configuration_service")
@@ -605,11 +605,12 @@ def _validate_local_tool(
             index_names = [knowledge_info.get("index_name")
                            for knowledge_info in knowledge_info_list]
             embedding_model = get_embedding_model(tenant_id=tenant_id)
+            vdb_core = get_vector_db_core()
             params = {
                 **instantiation_params,
                 'index_names': index_names,
-                'es_core': elastic_core,
-                'embedding_model': embedding_model
+                'vdb_core': vdb_core,
+                'embedding_model': embedding_model,
             }
             tool_instance = tool_class(**params)
         else: