diff --git a/backend/apps/file_management_app.py b/backend/apps/file_management_app.py index 151ddf926..e7607363a 100644 --- a/backend/apps/file_management_app.py +++ b/backend/apps/file_management_app.py @@ -154,7 +154,6 @@ async def process_files( index_name: Annotated[str, Body(...)], destination: Annotated[str, Body(...)], chunking_strategy: Annotated[Optional[str], Body(...)] = "basic", - model_id: Annotated[Optional[int], Body(...)] = None, authorization: Annotated[Optional[str], Header()] = None ): """ @@ -172,7 +171,6 @@ async def process_files( source_type=destination, index_name=index_name, authorization=authorization, - model_id=model_id ) process_result = await trigger_data_process(files, process_params) diff --git a/backend/consts/model.py b/backend/consts/model.py index dca8bc138..e82dad945 100644 --- a/backend/consts/model.py +++ b/backend/consts/model.py @@ -472,7 +472,6 @@ class ProcessParams(BaseModel): source_type: str index_name: str authorization: Optional[str] = None - model_id: Optional[int] = None class OpinionRequest(BaseModel): diff --git a/backend/utils/file_management_utils.py b/backend/utils/file_management_utils.py index 83c3957e7..43c59fe11 100644 --- a/backend/utils/file_management_utils.py +++ b/backend/utils/file_management_utils.py @@ -15,6 +15,7 @@ from consts.const import DATA_PROCESS_SERVICE, LIBREOFFICE_PROFILE_DIR from consts.model import ProcessParams from database.attachment_db import get_file_size_from_minio +from database.knowledge_db import get_knowledge_record from utils.auth_utils import get_current_user_id logger = logging.getLogger("file_management_utils") @@ -46,13 +47,27 @@ async def trigger_data_process(files: List[dict], process_params: ProcessParams) return None # Get tenant_id from authorization for downstream task processing - embedding_model_id = process_params.model_id tenant_id = None try: _, tenant_id = get_current_user_id(process_params.authorization) except Exception as e: logger.warning(f"Failed to get tenant_id from authorization: {e}") + # Resolve embedding model from the knowledge base record (not request model_id) + embedding_model_id = None + try: + knowledge_record = get_knowledge_record({ + "index_name": process_params.index_name, + "tenant_id": tenant_id, + }) + embedding_model_id = ( + knowledge_record.get("embedding_model_id") if knowledge_record else None + ) + except Exception as e: + logger.warning( + "Failed to resolve embedding_model_id from knowledge_record: %s", e + ) + # Build headers with authorization headers = { "Authorization": f"Bearer {process_params.authorization}" @@ -83,11 +98,11 @@ async def trigger_data_process(files: List[dict], process_params: ProcessParams) "Error from data process service: %s - %s", response, response.text if hasattr(response, 'text') else 'No response text') return {"status": "error", "code": response.status_code, - "message": f"Data process service error: {response.status_code}"} + "message": f"Data process service error: {response.status_code}"} except httpx.RequestError as e: logger.error("Failed to connect to data process service: %s", str(e)) return {"status": "error", "code": "CONNECTION_ERROR", - "message": f"Failed to connect to data process service: {str(e)}"} + "message": f"Failed to connect to data process service: {str(e)}"} else: # Batch file request @@ -117,11 +132,11 @@ async def trigger_data_process(files: List[dict], process_params: ProcessParams) "Error from data process service: %s - %s", response, response.text if hasattr(response, 'text') else 'No response text') return {"status": "error", "code": response.status_code, - "message": f"Data process service error: {response.status_code}"} + "message": f"Data process service error: {response.status_code}"} except httpx.RequestError as e: logger.error("Failed to connect to data process service: %s", str(e)) return {"status": "error", "code": "CONNECTION_ERROR", - "message": f"Failed to connect to data process service: {str(e)}"} + "message": f"Failed to connect to data process service: {str(e)}"} except Exception as e: logger.error("Error triggering data process: %s", str(e)) return {"status": "error", "code": "INTERNAL_ERROR", "message": f"Internal error: {str(e)}"} @@ -145,7 +160,7 @@ async def get_all_files_status(index_name: str): response = await client.get(f"{DATA_PROCESS_SERVICE}/tasks/indices/{index_name}", timeout=10.0) http_duration = time.time() - start_time logger.info(f"[get_all_files_status] HTTP request to {DATA_PROCESS_SERVICE}/tasks/indices/{index_name} " - f"completed in {http_duration:.3f}s, status={response.status_code}") + f"completed in {http_duration:.3f}s, status={response.status_code}") if response.status_code == 200: tasks_list = response.json() else: @@ -154,12 +169,12 @@ async def get_all_files_status(index_name: str): except Exception as e: logger.error(f"Failed to connect to data process service: {str(e)}") return {} - + logging.debug(f"Found {len(tasks_list)} tasks for index '{index_name}'") if not tasks_list: logger.warning(f"No tasks found for index '{index_name}'") return {} - + # Dictionary to store file statuses: # {path_or_url: {process_state, forward_state, timestamps, progress fields}} file_states = {} @@ -345,12 +360,12 @@ def get_file_size(source_type: str, path_or_url: str) -> int: async def convert_office_to_pdf(input_path: str, output_dir: str, timeout: int = 30) -> str: """ Convert Office document to PDF using LibreOffice. - + Args: input_path: Path to input Office file output_dir: Directory for output PDF file timeout: Conversion timeout in seconds (default: 30s) - + Returns: str: Path to generated PDF file """ @@ -382,34 +397,33 @@ def _run_libreoffice_conversion(): text=True, timeout=timeout ) - + try: # Run blocking subprocess in thread executor to avoid blocking event loop result = await asyncio.to_thread(_run_libreoffice_conversion) - + if result.returncode != 0: error_msg = result.stderr or result.stdout or "Unknown conversion error" logger.error(f"LibreOffice conversion failed: {error_msg}") raise RuntimeError(f"Office to PDF conversion failed: {error_msg}") - + # Find generated PDF file input_filename = os.path.basename(input_path) pdf_filename = os.path.splitext(input_filename)[0] + '.pdf' pdf_path = os.path.join(output_dir, pdf_filename) - + if not os.path.exists(pdf_path): raise RuntimeError(f"Converted PDF not found: {pdf_path}") - + return pdf_path - + except subprocess.TimeoutExpired: logger.error(f"Office to PDF conversion timeout after {timeout}s: {input_path}") raise TimeoutError(f"Office to PDF conversion timeout (>{timeout}s)") - + except FileNotFoundError as e: # LibreOffice executable not found in PATH logger.error(f"LibreOffice not available: {str(e)}") raise FileNotFoundError( "LibreOffice is not installed or not available in PATH. " ) from e - diff --git a/frontend/app/[locale]/knowledges/KnowledgeBaseConfiguration.tsx b/frontend/app/[locale]/knowledges/KnowledgeBaseConfiguration.tsx index bff93b2ed..efe3b68fb 100644 --- a/frontend/app/[locale]/knowledges/KnowledgeBaseConfiguration.tsx +++ b/frontend/app/[locale]/knowledges/KnowledgeBaseConfiguration.tsx @@ -876,7 +876,7 @@ function DataConfig({ isActive }: DataConfigProps) { setHasClickedUpload(false); setNewlyCreatedKbId(newKB.id); // Mark this KB as newly created - await uploadDocuments(newKB.id, filesToUpload, selectedModelId); + await uploadDocuments(newKB.id, filesToUpload); setUploadFiles([]); knowledgeBasePollingService @@ -924,7 +924,7 @@ function DataConfig({ isActive }: DataConfigProps) { : "embedding", }); - await uploadDocuments(kbId, filesToUpload, activeKbModelId); + await uploadDocuments(kbId, filesToUpload); setUploadFiles([]); knowledgeBasePollingService.triggerKnowledgeBaseListUpdate(true); diff --git a/frontend/app/[locale]/knowledges/contexts/DocumentContext.tsx b/frontend/app/[locale]/knowledges/contexts/DocumentContext.tsx index c38293e4e..55c53abe7 100644 --- a/frontend/app/[locale]/knowledges/contexts/DocumentContext.tsx +++ b/frontend/app/[locale]/knowledges/contexts/DocumentContext.tsx @@ -139,8 +139,7 @@ export const DocumentContext = createContext<{ ) => Promise; uploadDocuments: ( kbId: string, - files: File[], - modelId?: number + files: File[] ) => Promise<{ quota_status?: QuotaStatusResponse } | undefined>; deleteDocument: (kbId: string, docId: string) => Promise; }>({ @@ -263,15 +262,13 @@ export const DocumentProvider: React.FC = ({ // Upload documents to a knowledge base const uploadDocuments = useCallback( - async (kbId: string, files: File[], modelId?: number) => { + async (kbId: string, files: File[]) => { dispatch({ type: DOCUMENT_ACTION_TYPES.SET_UPLOADING, payload: true }); try { const uploadResult = await knowledgeBaseService.uploadDocuments( kbId, - files, - undefined, - modelId + files ); // Set loading state before fetching latest documents diff --git a/frontend/services/knowledgeBaseService.ts b/frontend/services/knowledgeBaseService.ts index 4878746e5..5846959bf 100644 --- a/frontend/services/knowledgeBaseService.ts +++ b/frontend/services/knowledgeBaseService.ts @@ -1160,8 +1160,7 @@ class KnowledgeBaseService { async uploadDocuments( kbId: string, files: File[], - chunkingStrategy?: string, - modelId?: number + chunkingStrategy?: string ): Promise<{ quota_status?: QuotaStatusResponse }> { try { if ( @@ -1231,7 +1230,6 @@ class KnowledgeBaseService { files: filesToProcess, chunking_strategy: chunkingStrategy, destination: "minio", - model_id: modelId, }), }); diff --git a/test/backend/app/test_file_management_app.py b/test/backend/app/test_file_management_app.py index 801f025a9..84983545e 100644 --- a/test/backend/app/test_file_management_app.py +++ b/test/backend/app/test_file_management_app.py @@ -170,12 +170,11 @@ async def _stub_trigger_data_process(files: List[dict], params: Any): model_stub = types.ModuleType("consts.model") class ProcessParams: # minimal stub - def __init__(self, chunking_strategy: str, source_type: str, index_name: str, authorization: str | None, model_id: int | None = None): + def __init__(self, chunking_strategy: str, source_type: str, index_name: str, authorization: str | None): self.chunking_strategy = chunking_strategy self.source_type = source_type self.index_name = index_name self.authorization = authorization - self.model_id = model_id model_stub.ProcessParams = ProcessParams sys.modules.setdefault("consts.model", model_stub) setattr(consts_pkg, "model", model_stub) @@ -328,7 +327,6 @@ async def fake_trigger(files, params): index_name="kb1", destination="local", authorization="Bearer x", - model_id=1, ) assert resp.status_code == 201 assert "Files processing triggered successfully" in resp.body.decode() @@ -347,7 +345,6 @@ async def fake_trigger(files, params): index_name="kb", destination="local", authorization=None, - model_id=1, ) assert "Data process service failed" in str(ei.value) @@ -365,7 +362,6 @@ async def fake_trigger(files, params): index_name="kb", destination="local", authorization=None, - model_id=1, ) assert "boom" in str(ei.value) diff --git a/test/backend/utils/test_file_management_utils.py b/test/backend/utils/test_file_management_utils.py index ce15596a0..c3099afcd 100644 --- a/test/backend/utils/test_file_management_utils.py +++ b/test/backend/utils/test_file_management_utils.py @@ -7,13 +7,12 @@ class _ProcessParams: - def __init__(self, authorization: str, source_type: str, chunking_strategy: str, index_name: Optional[str], model_id: Optional[int] = 42, + def __init__(self, authorization: str, source_type: str, chunking_strategy: str, index_name: Optional[str], tenant_id: Optional[str] = "tenant-1"): self.authorization = authorization self.source_type = source_type self.chunking_strategy = chunking_strategy self.index_name = index_name - self.model_id = model_id self.tenant_id = tenant_id @@ -35,12 +34,22 @@ def stub_project_modules(monkeypatch): setattr(attach_mod, "get_file_size_from_minio", lambda object_name, bucket=None: 777) sys.modules["database.attachment_db"] = attach_mod + # database.knowledge_db + knowledge_mod = types.ModuleType("database.knowledge_db") + setattr( + knowledge_mod, + "get_knowledge_record", + lambda query=None: {"embedding_model_id": 42, "index_name": (query or {}).get("index_name")}, + ) + sys.modules["database.knowledge_db"] = knowledge_mod + # Ensure parent package exists if "database" not in sys.modules: pkg = types.ModuleType("database") setattr(pkg, "__path__", []) sys.modules["database"] = pkg setattr(sys.modules["database"], "attachment_db", attach_mod) + setattr(sys.modules["database"], "knowledge_db", knowledge_mod) # utils.auth_utils auth_mod = types.ModuleType("utils.auth_utils") @@ -175,6 +184,11 @@ async def test_trigger_data_process_single_success_with_embedding(fmu, monkeypat fake_client = _FakeAsyncClient(_Resp(201, {"task_id": "t1"})) fake_httpx = types.SimpleNamespace(AsyncClient=lambda: fake_client, RequestError=_FakeRequestError) monkeypatch.setattr(fmu, "httpx", fake_httpx) + monkeypatch.setattr( + fmu, + "get_knowledge_record", + lambda query=None: {"embedding_model_id": 42, "index_name": "idx"}, + ) params = _ProcessParams("tok", "local", "basic", "idx") files = [{"path_or_url": "/data/a.txt", "filename": "a.txt"}] @@ -186,6 +200,75 @@ async def test_trigger_data_process_single_success_with_embedding(fmu, monkeypat assert fake_client.last_post["json"]["tenant_id"] == "tenant-1" +@pytest.mark.asyncio +async def test_trigger_data_process_queries_knowledge_record_by_index(fmu, monkeypatch): + fake_client = _FakeAsyncClient(_Resp(201, {"task_id": "t1"})) + fake_httpx = types.SimpleNamespace(AsyncClient=lambda: fake_client, RequestError=_FakeRequestError) + monkeypatch.setattr(fmu, "httpx", fake_httpx) + + captured_query: Dict[str, Any] = {} + + def _fake_get_knowledge_record(query=None): + captured_query.update(query or {}) + return {"embedding_model_id": 77, "index_name": "idx"} + + monkeypatch.setattr(fmu, "get_knowledge_record", _fake_get_knowledge_record) + + params = _ProcessParams("tok", "local", "basic", "idx") + files = [{"path_or_url": "/data/a.txt", "filename": "a.txt"}] + out = await fmu.trigger_data_process(files, params) + assert out == {"task_id": "t1"} + assert captured_query.get("index_name") == "idx" + assert captured_query.get("tenant_id") == "tenant-1" + assert fake_client.last_post["json"]["embedding_model_id"] == 77 + + +@pytest.mark.asyncio +async def test_trigger_data_process_missing_embedding_model_id_sends_none(fmu, monkeypatch): + fake_client = _FakeAsyncClient(_Resp(201, {"task_id": "t1"})) + fake_httpx = types.SimpleNamespace(AsyncClient=lambda: fake_client, RequestError=_FakeRequestError) + monkeypatch.setattr(fmu, "httpx", fake_httpx) + monkeypatch.setattr(fmu, "get_knowledge_record", lambda query=None: {}) + + params = _ProcessParams("tok", "local", "basic", "idx") + files = [{"path_or_url": "/data/a.txt", "filename": "a.txt"}] + out = await fmu.trigger_data_process(files, params) + assert out == {"task_id": "t1"} + assert fake_client.last_post["json"]["embedding_model_id"] is None + + +@pytest.mark.asyncio +async def test_trigger_data_process_knowledge_record_none_sends_none(fmu, monkeypatch): + fake_client = _FakeAsyncClient(_Resp(201, {"task_id": "t1"})) + fake_httpx = types.SimpleNamespace(AsyncClient=lambda: fake_client, RequestError=_FakeRequestError) + monkeypatch.setattr(fmu, "httpx", fake_httpx) + monkeypatch.setattr(fmu, "get_knowledge_record", lambda query=None: None) + + params = _ProcessParams("tok", "local", "basic", "idx") + files = [{"path_or_url": "/data/a.txt", "filename": "a.txt"}] + out = await fmu.trigger_data_process(files, params) + assert out == {"task_id": "t1"} + assert fake_client.last_post["json"]["embedding_model_id"] is None + + +@pytest.mark.asyncio +async def test_trigger_data_process_knowledge_record_raises_sends_none(fmu, monkeypatch): + fake_client = _FakeAsyncClient(_Resp(201, {"task_id": "t1"})) + fake_httpx = types.SimpleNamespace(AsyncClient=lambda: fake_client, RequestError=_FakeRequestError) + monkeypatch.setattr(fmu, "httpx", fake_httpx) + + def _raise_get_knowledge_record(query=None): + raise RuntimeError("db unavailable") + + monkeypatch.setattr(fmu, "get_knowledge_record", _raise_get_knowledge_record) + + params = _ProcessParams("tok", "local", "basic", "idx") + files = [{"path_or_url": "/data/a.txt", "filename": "a.txt"}] + out = await fmu.trigger_data_process(files, params) + assert out == {"task_id": "t1"} + assert fake_client.last_post["json"]["embedding_model_id"] is None + + @pytest.mark.asyncio async def test_trigger_data_process_single_non201_error(fmu, monkeypatch): fake_client = _FakeAsyncClient(_Resp(400, None, text="boom")) @@ -215,6 +298,11 @@ async def test_trigger_data_process_batch_success(fmu, monkeypatch): fake_client = _FakeAsyncClient(_Resp(201, {"task_ids": ["t1", "t2"]})) fake_httpx = types.SimpleNamespace(AsyncClient=lambda: fake_client, RequestError=_FakeRequestError) monkeypatch.setattr(fmu, "httpx", fake_httpx) + monkeypatch.setattr( + fmu, + "get_knowledge_record", + lambda query=None: {"embedding_model_id": 42, "index_name": "idx"}, + ) params = _ProcessParams("tok", "minio", "basic", "idx") files = [ @@ -225,7 +313,7 @@ async def test_trigger_data_process_batch_success(fmu, monkeypatch): assert out == {"task_ids": ["t1", "t2"]} assert fake_client.last_post["url"].endswith("/tasks/batch") assert len(fake_client.last_post["json"]["sources"]) == 2 - + assert all(s["embedding_model_id"] == 42 for s in fake_client.last_post["json"]["sources"]) @pytest.mark.asyncio async def test_trigger_data_process_batch_non201_and_request_error(fmu, monkeypatch):