✨ Now document process has a progress bar

Author: Phinease

backend/agents/create_agent_info.py

@@ -15,7 +15,7 @@
     get_vector_db_core,
     get_embedding_model,
 )
-from services.tenant_config_service import get_selected_knowledge_list
+from services.tenant_config_service import get_selected_knowledge_list, build_knowledge_name_mapping
 from services.remote_mcp_service import get_remote_mcp_server_list
 from services.memory_config_service import build_memory_context
 from services.image_service import get_vlm_model
@@ -241,6 +241,7 @@ async def create_tool_config_list(agent_id, tenant_id, user_id):
                 "index_names": index_names,
                 "vdb_core": get_vector_db_core(),
                 "embedding_model": get_embedding_model(tenant_id=tenant_id),
+                "name_resolver": build_knowledge_name_mapping(tenant_id=tenant_id, user_id=user_id),
             }
         elif tool_config.class_name == "AnalyzeTextFileTool":
             tool_config.metadata = {

backend/apps/vectordatabase_app.py

@@ -1,9 +1,11 @@
 import logging
+import json
 from http import HTTPStatus
 from typing import Any, Dict, List, Optional
 
 from fastapi import APIRouter, Body, Depends, Header, HTTPException, Path, Query
 from fastapi.responses import JSONResponse
+import re
 
 from consts.model import ChunkCreateRequest, ChunkUpdateRequest, HybridSearchRequest, IndexingResponse
 from nexent.vector_database.base import VectorDatabaseCore
@@ -15,6 +17,8 @@
 )
 from services.redis_service import get_redis_service
 from utils.auth_utils import get_current_user_id
+from utils.file_management_utils import get_all_files_status
+from database.knowledge_db import get_index_name_by_knowledge_name
 
 router = APIRouter(prefix="/indices")
 service = ElasticSearchService()
@@ -49,7 +53,8 @@ def create_new_index(
     """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, vdb_core, user_id, tenant_id)
+        # Treat path parameter as user-facing knowledge base name for new creations
+        return ElasticSearchService.create_knowledge_base(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)}")
@@ -99,7 +104,9 @@ def create_index_documents(
         data: List[Dict[str, Any]
                    ] = Body(..., description="Document List to process"),
         vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
-        authorization: Optional[str] = Header(None)
+        authorization: Optional[str] = Header(None),
+        task_id: Optional[str] = Header(
+            None, alias="X-Task-Id", description="Task ID for progress tracking"),
 ):
     """
     Index documents with embeddings, creating the index if it doesn't exist.
@@ -108,12 +115,21 @@ 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, vdb_core)
+        return ElasticSearchService.index_documents(
+            embedding_model=embedding_model,
+            index_name=index_name,
+            data=data,
+            vdb_core=vdb_core,
+            task_id=task_id,
+        )
     except Exception as e:
         error_msg = str(e)
         logger.error(f"Error indexing documents: {error_msg}")
+
         raise HTTPException(
-            status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=f"Error indexing documents: {error_msg}")
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+            detail=f"Error indexing documents: {error_msg}"
+        )
 
 
 @router.get("/{index_name}/files")
@@ -187,6 +203,66 @@ def delete_documents(
             status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=f"Error delete indexing documents: {e}")
 
 
+@router.get("/{index_name}/documents/{path_or_url:path}/error-info")
+async def get_document_error_info(
+        index_name: str = Path(..., description="Name of the index"),
+        path_or_url: str = Path(...,
+                                description="Path or URL of the document"),
+        authorization: Optional[str] = Header(None)
+):
+    """Get error information for a document"""
+    try:
+        celery_task_files = await get_all_files_status(index_name)
+        file_status = celery_task_files.get(path_or_url)
+
+        if not file_status:
+            raise HTTPException(
+                status_code=HTTPStatus.NOT_FOUND,
+                detail=f"Document {path_or_url} not found in index {index_name}"
+            )
+
+        task_id = file_status.get('latest_task_id', '')
+        if not task_id:
+            return {
+                "status": "success",
+                "error_code": None,
+            }
+
+        redis_service = get_redis_service()
+        raw_error = redis_service.get_error_info(task_id)
+        error_code = None
+
+        if raw_error:
+            # Try to parse JSON (new format with error_code only)
+            try:
+                parsed = json.loads(raw_error)
+                if isinstance(parsed, dict) and "error_code" in parsed:
+                    error_code = parsed.get("error_code")
+            except Exception:
+                # Fallback: regex extraction if JSON parsing fails
+                try:
+                    match = re.search(
+                        r'["\']error_code["\']\s*:\s*["\']([^"\']+)["\']', raw_error)
+                    if match:
+                        error_code = match.group(1)
+                except Exception:
+                    pass
+
+        return {
+            "status": "success",
+            "error_code": error_code,
+        }
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(
+            f"Error getting error info for document {path_or_url}: {str(e)}")
+        raise HTTPException(
+            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+            detail=f"Error getting error info: {str(e)}"
+        )
+
+
 # Health check
 @router.get("/health")
 def health_check(vdb_core: VectorDatabaseCore = Depends(get_vector_db_core)):
@@ -201,25 +277,35 @@ def health_check(vdb_core: VectorDatabaseCore = Depends(get_vector_db_core)):
 @router.post("/{index_name}/chunks")
 def get_index_chunks(
         index_name: str = Path(...,
-                               description="Name of the index to get chunks from"),
+                               description="Name of the index (or knowledge_name) to get chunks from"),
         page: int = Query(
             None, description="Page number (1-based) for pagination"),
         page_size: int = Query(
             None, description="Number of records per page for pagination"),
         path_or_url: Optional[str] = Query(
             None, description="Filter chunks by document path_or_url"),
-        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core)
+        vdb_core: VectorDatabaseCore = Depends(get_vector_db_core),
+        authorization: Optional[str] = Header(None)
 ):
     """Get chunks from the specified index, with optional pagination support"""
     try:
+        _, tenant_id = get_current_user_id(authorization)
+        actual_index_name = get_index_name_by_knowledge_name(
+            index_name, tenant_id)
+
         result = ElasticSearchService.get_index_chunks(
-            index_name=index_name,
+            index_name=actual_index_name,
             page=page,
             page_size=page_size,
             path_or_url=path_or_url,
             vdb_core=vdb_core,
         )
         return JSONResponse(status_code=HTTPStatus.OK, content=result)
+    except ValueError as e:
+        raise HTTPException(
+            status_code=HTTPStatus.NOT_FOUND,
+            detail=str(e)
+        )
     except Exception as e:
         error_msg = str(e)
         logger.error(
@@ -230,21 +316,29 @@ def get_index_chunks(
 
 @router.post("/{index_name}/chunk")
 def create_chunk(
-        index_name: str = Path(..., description="Name of the index"),
+        index_name: str = Path(...,
+                               description="Name of the index (or knowledge_name)"),
         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)
+        user_id, tenant_id = get_current_user_id(authorization)
+        actual_index_name = get_index_name_by_knowledge_name(
+            index_name, tenant_id)
         result = ElasticSearchService.create_chunk(
-            index_name=index_name,
+            index_name=actual_index_name,
             chunk_request=payload,
             vdb_core=vdb_core,
             user_id=user_id,
         )
         return JSONResponse(status_code=HTTPStatus.OK, content=result)
+    except ValueError as e:
+        raise HTTPException(
+            status_code=HTTPStatus.NOT_FOUND,
+            detail=str(e)
+        )
     except Exception as exc:
         logger.error(
             "Error creating chunk for index %s: %s", index_name, exc, exc_info=True
@@ -256,7 +350,8 @@ def create_chunk(
 
 @router.put("/{index_name}/chunk/{chunk_id}")
 def update_chunk(
-        index_name: str = Path(..., description="Name of the index"),
+        index_name: str = Path(...,
+                               description="Name of the index (or knowledge_name)"),
         chunk_id: str = Path(..., description="Chunk identifier"),
         payload: ChunkUpdateRequest = Body(...,
                                            description="Chunk update payload"),
@@ -265,18 +360,22 @@ def update_chunk(
 ):
     """Update an existing chunk."""
     try:
-        user_id, _ = get_current_user_id(authorization)
+        user_id, tenant_id = get_current_user_id(authorization)
+        actual_index_name = get_index_name_by_knowledge_name(
+            index_name, tenant_id)
         result = ElasticSearchService.update_chunk(
-            index_name=index_name,
+            index_name=actual_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:
+    except ValueError as e:
         raise HTTPException(
-            status_code=HTTPStatus.BAD_REQUEST, detail=str(exc))
+            status_code=HTTPStatus.NOT_FOUND,
+            detail=str(e)
+        )
     except Exception as exc:
         logger.error(
             "Error updating chunk %s for index %s: %s",
@@ -292,22 +391,28 @@ def update_chunk(
 
 @router.delete("/{index_name}/chunk/{chunk_id}")
 def delete_chunk(
-        index_name: str = Path(..., description="Name of the index"),
+        index_name: str = Path(...,
+                               description="Name of the index (or knowledge_name)"),
         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)
+        _, tenant_id = get_current_user_id(authorization)
+        actual_index_name = get_index_name_by_knowledge_name(
+            index_name, tenant_id)
         result = ElasticSearchService.delete_chunk(
-            index_name=index_name,
+            index_name=actual_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 ValueError as e:
+        raise HTTPException(
+            status_code=HTTPStatus.NOT_FOUND,
+            detail=str(e)
+        )
     except Exception as exc:
         logger.error(
             "Error deleting chunk %s for index %s: %s",

backend/consts/const.py

@@ -38,6 +38,7 @@ class VectorDatabaseType(str, Enum):
 MAX_FILE_SIZE = 100 * 1024 * 1024  # 100MB
 MAX_CONCURRENT_UPLOADS = 5
 UPLOAD_FOLDER = os.getenv('UPLOAD_FOLDER', 'uploads')
+ROOT_DIR = os.getenv("ROOT_DIR")
 
 
 # Supabase Configuration

backend/consts/model.py

@@ -59,6 +59,7 @@ class ModelRequest(BaseModel):
     connect_status: Optional[str] = ''
     expected_chunk_size: Optional[int] = None
     maximum_chunk_size: Optional[int] = None
+    chunk_batch: Optional[int] = None
 
 
 class ProviderModelRequest(BaseModel):