✨ Now supports add/delete/edit chunks

Author: Phinease

backend/apps/vectordatabase_app.py

@@ -5,7 +5,7 @@
 from fastapi import APIRouter, Body, Depends, Header, HTTPException, Path, Query
 from fastapi.responses import JSONResponse
 
-from consts.model import HybridSearchRequest, IndexingResponse
+from consts.model import ChunkCreateRequest, ChunkUpdateRequest, HybridSearchRequest, IndexingResponse
 from nexent.vector_database.base import VectorDatabaseCore
 from services.vectordatabase_service import (
     ElasticSearchService,
@@ -228,6 +228,99 @@ def get_index_chunks(
             status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=f"Error getting chunks: {error_msg}")
 
 
+@router.post("/{index_name}/chunk")
+def create_chunk(
+        index_name: str = Path(..., description="Name of the index"),
+        payload: ChunkCreateRequest = Body(..., description="Chunk data"),
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
+        authorization: Optional[str] = Header(None),
+):
+    """Create a manual chunk."""
+    try:
+        user_id, _ = get_current_user_id(authorization)
+        result = ElasticSearchService.create_chunk(
+            index_name=index_name,
+            chunk_request=payload,
+            vdb_core=vdb_core,
+            user_id=user_id,
+        )
+        return JSONResponse(status_code=HTTPStatus.OK, content=result)
+    except Exception as exc:
+        logger.error(
+            "Error creating chunk for index %s: %s", index_name, exc, exc_info=True
+        )
+        raise HTTPException(
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(exc)
+        )
+
+
+@router.put("/{index_name}/chunk/{chunk_id}")
+def update_chunk(
+        index_name: str = Path(..., description="Name of the index"),
+        chunk_id: str = Path(..., description="Chunk identifier"),
+        payload: ChunkUpdateRequest = Body(...,
+                                           description="Chunk update payload"),
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
+        authorization: Optional[str] = Header(None),
+):
+    """Update an existing chunk."""
+    try:
+        user_id, _ = get_current_user_id(authorization)
+        result = ElasticSearchService.update_chunk(
+            index_name=index_name,
+            chunk_id=chunk_id,
+            chunk_request=payload,
+            vdb_core=vdb_core,
+            user_id=user_id,
+        )
+        return JSONResponse(status_code=HTTPStatus.OK, content=result)
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST, detail=str(exc))
+    except Exception as exc:
+        logger.error(
+            "Error updating chunk %s for index %s: %s",
+            chunk_id,
+            index_name,
+            exc,
+            exc_info=True,
+        )
+        raise HTTPException(
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(exc)
+        )
+
+
+@router.delete("/{index_name}/chunk/{chunk_id}")
+def delete_chunk(
+        index_name: str = Path(..., description="Name of the index"),
+        chunk_id: str = Path(..., description="Chunk identifier"),
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
+        authorization: Optional[str] = Header(None),
+):
+    """Delete a chunk."""
+    try:
+        get_current_user_id(authorization)
+        result = ElasticSearchService.delete_chunk(
+            index_name=index_name,
+            chunk_id=chunk_id,
+            vdb_core=vdb_core,
+        )
+        return JSONResponse(status_code=HTTPStatus.OK, content=result)
+    except ValueError as exc:
+        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail=str(exc))
+    except Exception as exc:
+        logger.error(
+            "Error deleting chunk %s for index %s: %s",
+            chunk_id,
+            index_name,
+            exc,
+            exc_info=True,
+        )
+        raise HTTPException(
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(exc)
+        )
+
+
 @router.post("/search/hybrid")
 async def hybrid_search(
         payload: HybridSearchRequest,

backend/consts/model.py

@@ -175,6 +175,31 @@ class IndexingResponse(BaseModel):
     total_submitted: int
 
 
+class ChunkCreateRequest(BaseModel):
+    """Request payload for manual chunk creation."""
+
+    content: str = Field(..., min_length=1, description="Chunk content")
+    title: Optional[str] = Field(None, description="Optional chunk title")
+    filename: Optional[str] = Field(None, description="Associated file name")
+    path_or_url: Optional[str] = Field(None, description="Source path or URL")
+    chunk_id: Optional[str] = Field(
+        None, description="Explicit chunk identifier")
+    metadata: Dict[str, Any] = Field(
+        default_factory=dict, description="Additional chunk metadata")
+
+
+class ChunkUpdateRequest(BaseModel):
+    """Request payload for chunk updates."""
+
+    content: Optional[str] = Field(None, description="Updated chunk content")
+    title: Optional[str] = Field(None, description="Updated chunk title")
+    filename: Optional[str] = Field(None, description="Updated file name")
+    path_or_url: Optional[str] = Field(
+        None, description="Updated source path or URL")
+    metadata: Dict[str, Any] = Field(
+        default_factory=dict, description="Additional metadata updates")
+
+
 class HybridSearchRequest(BaseModel):
     """Request payload for hybrid knowledge-base searches."""
     query: str = Field(..., min_length=1,

backend/services/vectordatabase_service.py

@@ -13,6 +13,7 @@
 import logging
 import os
 import time
+import uuid
 from datetime import datetime, timezone
 from typing import Any, Dict, Generator, List, Optional
 
@@ -24,6 +25,7 @@
 from nexent.vector_database.elasticsearch_core import ElasticSearchCore
 
 from consts.const import ES_API_KEY, ES_HOST, LANGUAGE, VectorDatabaseType
+from consts.model import ChunkCreateRequest, ChunkUpdateRequest
 from database.attachment_db import delete_file
 from database.knowledge_db import (
     create_knowledge_record,
@@ -35,8 +37,17 @@
 from utils.config_utils import tenant_config_manager, get_model_name_from_config
 from utils.file_management_utils import get_all_files_status, get_file_size
 
-ALLOWED_CHUNK_FIELDS = {"filename",
-                        "path_or_url", "content", "create_time", "id"}
+ALLOWED_CHUNK_FIELDS = {
+    "id",
+    "title",
+    "filename",
+    "path_or_url",
+    "content",
+    "create_time",
+    "language",
+    "author",
+    "date",
+}
 
 # Configure logging
 logger = logging.getLogger("vectordatabase_service")
@@ -997,6 +1008,105 @@ def get_index_chunks(
             logger.error(error_msg)
             raise Exception(error_msg)
 
+    @staticmethod
+    def create_chunk(
+        index_name: str,
+        chunk_request: ChunkCreateRequest,
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
+        user_id: Optional[str] = None,
+    ):
+        """
+        Create a manual chunk entry in the specified index.
+        """
+        try:
+            chunk_payload = ElasticSearchService._build_chunk_payload(
+                base_fields={
+                    "id": chunk_request.chunk_id or ElasticSearchService._generate_chunk_id(),
+                    "title": chunk_request.title,
+                    "filename": chunk_request.filename,
+                    "path_or_url": chunk_request.path_or_url,
+                    "content": chunk_request.content,
+                    "created_by": user_id,
+                },
+                metadata=chunk_request.metadata,
+                ensure_create_time=True,
+            )
+            result = vdb_core.create_chunk(index_name, chunk_payload)
+            return {
+                "status": "success",
+                "message": f"Chunk {result.get('id')} created successfully",
+                "chunk_id": result.get("id"),
+            }
+        except Exception as exc:
+            logger.error("Error creating chunk in index %s: %s",
+                         index_name, exc, exc_info=True)
+            raise Exception(f"Error creating chunk: {exc}")
+
+    @staticmethod
+    def update_chunk(
+        index_name: str,
+        chunk_id: str,
+        chunk_request: ChunkUpdateRequest,
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
+        user_id: Optional[str] = None,
+    ):
+        """
+        Update a chunk document.
+        """
+        try:
+            update_fields = chunk_request.dict(
+                exclude_unset=True, exclude={"metadata"})
+            metadata = chunk_request.metadata or {}
+            update_payload = ElasticSearchService._build_chunk_payload(
+                base_fields={
+                    **update_fields,
+                    "updated_by": user_id,
+                    "update_time": datetime.utcnow().strftime(
+                        "%Y-%m-%dT%H:%M:%S"),
+                },
+                metadata=metadata,
+                ensure_create_time=False,
+            )
+
+            if not update_payload:
+                raise ValueError("No update fields supplied.")
+
+            result = vdb_core.update_chunk(
+                index_name, chunk_id, update_payload)
+            return {
+                "status": "success",
+                "message": f"Chunk {result.get('id')} updated successfully",
+                "chunk_id": result.get("id"),
+            }
+        except Exception as exc:
+            logger.error("Error updating chunk %s in index %s: %s",
+                         chunk_id, index_name, exc, exc_info=True)
+            raise Exception(f"Error updating chunk: {exc}")
+
+    @staticmethod
+    def delete_chunk(
+        index_name: str,
+        chunk_id: str,
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
+    ):
+        """
+        Delete a chunk document by id.
+        """
+        try:
+            deleted = vdb_core.delete_chunk(index_name, chunk_id)
+            if not deleted:
+                raise ValueError(
+                    f"Chunk {chunk_id} not found in index {index_name}")
+            return {
+                "status": "success",
+                "message": f"Chunk {chunk_id} deleted successfully",
+                "chunk_id": chunk_id,
+            }
+        except Exception as exc:
+            logger.error("Error deleting chunk %s in index %s: %s",
+                         chunk_id, index_name, exc, exc_info=True)
+            raise Exception(f"Error deleting chunk: {exc}")
+
     @staticmethod
     def search_hybrid(
             *,
@@ -1058,4 +1168,32 @@ def search_hybrid(
                 f"Hybrid search failed for indices {index_names}: {exc}",
                 exc_info=True,
             )
-            raise Exception(f"Error executing hybrid search: {str(exc)}")
+            raise Exception(f"Error executing hybrid search: {str(exc)}")
+
+    @staticmethod
+    def _generate_chunk_id() -> str:
+        """Generate a deterministic chunk id."""
+        return f"chunk_{uuid.uuid4().hex}"
+
+    @staticmethod
+    def _build_chunk_payload(
+        base_fields: Dict[str, Any],
+        metadata: Optional[Dict[str, Any]],
+        ensure_create_time: bool = True,
+    ) -> Dict[str, Any]:
+        """
+        Merge and sanitize chunk payload fields.
+        """
+        payload = {
+            key: value for key, value in (base_fields or {}).items() if value is not None
+        }
+        if metadata:
+            for key, value in metadata.items():
+                if value is not None:
+                    payload[key] = value
+
+        if ensure_create_time and "create_time" not in payload:
+            payload["create_time"] = datetime.utcnow().strftime(
+                "%Y-%m-%dT%H:%M:%S")
+
+        return payload