✨ Now user can adjust the chunking size when adding or editing the embedding model

🔨 Change the default chunking size from 1200-1500 to 1024-1536
🎨 Improve the coding style of data_process module in sdk

Author: Jasonxia007

backend/apps/data_process_app.py

@@ -11,26 +11,24 @@
     ConvertStateRequest,
     TaskRequest,
 )
-import importlib
-import sys
+from data_process.tasks import process_and_forward, process_sync
+from services.data_process_service import get_data_process_service
 
 logger = logging.getLogger("data_process.app")
 
+# Use shared service instance
+service = get_data_process_service()
+
+
 @asynccontextmanager
 async def lifespan(app: APIRouter):
     # Startup
     try:
-        svc_mod = sys.modules.get("services.data_process_service") or importlib.import_module(
-            "services.data_process_service")
-        svc = svc_mod.get_data_process_service()
-        await svc.start()
+        await service.start()
         yield
     finally:
         # Shutdown
-        svc_mod = sys.modules.get("services.data_process_service") or importlib.import_module(
-            "services.data_process_service")
-        svc = svc_mod.get_data_process_service()
-        await svc.stop()
+        await service.stop()
 
 
 router = APIRouter(
@@ -51,9 +49,7 @@ async def create_task(request: TaskRequest, authorization: Optional[str] = Heade
 
     logger.info(
         f"Creating task with source_type: {request.source_type}, model_id: {request.embedding_model_id}")
-    tasks_mod = sys.modules.get(
-        "data_process.tasks") or importlib.import_module("data_process.tasks")
-    task_result = tasks_mod.process_and_forward.delay(
+    task_result = process_and_forward.delay(
         source=request.source,
         source_type=request.source_type,
         chunking_strategy=request.chunking_strategy,
@@ -90,9 +86,7 @@ async def process_sync_endpoint(
     """
     try:
         # Use the synchronous process task with high priority
-        tasks_mod = sys.modules.get(
-            "data_process.tasks") or importlib.import_module("data_process.tasks")
-        task_result = tasks_mod.process_sync.apply_async(
+        task_result = process_sync.apply_async(
             kwargs={
                 'source': source,
                 'source_type': source_type,
@@ -137,10 +131,7 @@ async def create_batch_tasks(request: BatchTaskRequest, authorization: Optional[
     Processing happens in the background for each file independently.
     """
     try:
-        svc_mod = sys.modules.get("services.data_process_service") or importlib.import_module(
-            "services.data_process_service")
-        svc = svc_mod.get_data_process_service()
-        task_ids = await svc.create_batch_tasks_impl(authorization=authorization, request=request)
+        task_ids = await service.create_batch_tasks_impl(authorization=authorization, request=request)
         return JSONResponse(status_code=HTTPStatus.CREATED, content={"task_ids": task_ids})
     except HTTPException:
         raise
@@ -163,16 +154,13 @@ async def load_image(url: str):
     """
     try:
         # Use the service to load the image
-        svc_mod = sys.modules.get("services.data_process_service") or importlib.import_module(
-            "services.data_process_service")
-        svc = svc_mod.get_data_process_service()
-        image = await svc.load_image(url)
+        image = await service.load_image(url)
 
         if image is None:
             raise HTTPException(
                 status_code=HTTPStatus.NOT_FOUND, detail="Failed to load image or image format not supported")
 
-        image_data, content_type = await svc.convert_to_base64(image)
+        image_data, content_type = await service.convert_to_base64(image)
         return JSONResponse(status_code=HTTPStatus.OK,
                             content={"success": True, "base64": image_data, "content_type": content_type})
     except HTTPException:
@@ -186,10 +174,7 @@ async def load_image(url: str):
 @router.get("")
 async def list_tasks():
     """Get a list of all tasks with their basic status information"""
-    svc_mod = sys.modules.get("services.data_process_service") or importlib.import_module(
-        "services.data_process_service")
-    svc = svc_mod.get_data_process_service()
-    tasks = await svc.get_all_tasks()
+    tasks = await service.get_all_tasks()
 
     task_responses = []
     for task in tasks:
@@ -219,10 +204,7 @@ async def get_index_tasks(index_name: str):
     Returns tasks that are being processed or waiting to be processed
     """
     try:
-        svc_mod = sys.modules.get("services.data_process_service") or importlib.import_module(
-            "services.data_process_service")
-        svc = svc_mod.get_data_process_service()
-        return await svc.get_index_tasks(index_name)
+        return await service.get_index_tasks(index_name)
     except Exception as e:
         raise HTTPException(
             status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))
@@ -231,9 +213,7 @@ async def get_index_tasks(index_name: str):
 @router.get("/{task_id}/details")
 async def get_task_details(task_id: str):
     """Get detailed information about a task, including results"""
-    utils_mod = sys.modules.get(
-        "data_process.utils") or importlib.import_module("data_process.utils")
-    task = await utils_mod.get_task_details(task_id)
+    task = await service.get_task_details(task_id)
     if not task:
         raise HTTPException(status_code=HTTPStatus.NOT_FOUND,
                             detail="Task not found")
@@ -253,8 +233,7 @@ async def filter_important_image(
     Returns importance score and confidence level.
     """
     try:
-        svc = get_data_process_service()
-        result = await svc.filter_important_image(
+        result = await service.filter_important_image(
             image_url=image_url,
             positive_prompt=positive_prompt,
             negative_prompt=negative_prompt
@@ -291,8 +270,7 @@ async def process_text_file(
         file_content = await file.read()
         filename = file.filename or "unknown_file"
 
-        svc = get_data_process_service()
-        result = await svc.process_uploaded_text_file(
+        result = await service.process_uploaded_text_file(
             file_content=file_content,
             filename=filename,
             chunking_strategy=chunking_strategy,
@@ -317,10 +295,7 @@ async def convert_state(request: ConvertStateRequest):
     This endpoint converts a process state string to a forward state string.
     """
     try:
-        svc_mod = sys.modules.get("services.data_process_service") or importlib.import_module(
-            "services.data_process_service")
-        svc = svc_mod.get_data_process_service()
-        result = svc.convert_celery_states_to_custom(
+        result = service.convert_celery_states_to_custom(
             process_celery_state=request.process_state or "",
             forward_celery_state=request.forward_state or ""
         )

docker/init.sql

@@ -163,6 +163,8 @@ CREATE TABLE IF NOT EXISTS "model_record_t" (
   "base_url" varchar(500) COLLATE "pg_catalog"."default",
   "max_tokens" int4,
   "used_token" int4,
+  "expected_chunk_size" int4,
+  "maximum_chunk_size" int4,
   "display_name" varchar(100) COLLATE "pg_catalog"."default",
   "connect_status" varchar(100) COLLATE "pg_catalog"."default",
   "create_time" timestamp(0) DEFAULT CURRENT_TIMESTAMP,
@@ -183,6 +185,8 @@ COMMENT ON COLUMN "model_record_t"."api_key" IS 'Model API key, used for authent
 COMMENT ON COLUMN "model_record_t"."base_url" IS 'Base URL address, used for requesting remote model services';
 COMMENT ON COLUMN "model_record_t"."max_tokens" IS 'Maximum available tokens for the model';
 COMMENT ON COLUMN "model_record_t"."used_token" IS 'Number of tokens already used by the model in Q&A';
+COMMENT ON COLUMN "model_record_t".expected_chunk_size IS 'Expected chunk size for embedding models, used during document chunking';
+COMMENT ON COLUMN "model_record_t".maximum_chunk_size IS 'Maximum chunk size for embedding models, used during document chunking';
 COMMENT ON COLUMN "model_record_t"."display_name" IS 'Model name displayed directly in frontend, customized by user';
 COMMENT ON COLUMN "model_record_t"."connect_status" IS 'Model connectivity status from last check, optional values: "检测中"、"可用"、"不可用"';
 COMMENT ON COLUMN "model_record_t"."create_time" IS 'Creation time, audit field';

test/backend/app/test_data_process_app.py

@@ -1,10 +1,10 @@
 import sys
 import types
-import asyncio
 from typing import Any, Dict, List, Optional, Tuple
+from http import HTTPStatus
 
 import pytest
-from fastapi import FastAPI
+from fastapi import FastAPI, HTTPException
 from fastapi.testclient import TestClient
 from pydantic import BaseModel
 
@@ -116,6 +116,11 @@ async def get_index_tasks(self, index_name: str):
             raise RuntimeError("oops")
         return [{"id": "x"}]
 
+    async def get_task_details(self, task_id: str):
+        if task_id == "missing":
+            return None
+        return {"id": task_id, "ok": True}
+
     async def filter_important_image(self, image_url: str, positive_prompt: str, negative_prompt: str):
         if image_url == "err":
             raise RuntimeError("bad")
@@ -212,7 +217,6 @@ def test_process_sync_endpoint_success():
 def test_process_sync_endpoint_error(monkeypatch):
     # Reconfigure tasks stub to raise when getting result
     from backend.apps import data_process_app as app_module
-    tasks_mod = sys.modules["data_process.tasks"]
 
     class _ErrResult(_DummyResult):
         def get(self, timeout=None):
@@ -222,7 +226,7 @@ class _PSyncErr:
         def apply_async(self, **kwargs):
             return _ErrResult("tid")
 
-    setattr(tasks_mod, "process_sync", _PSyncErr())
+    monkeypatch.setattr(app_module, "process_sync", _PSyncErr(), raising=True)
 
     app = _build_app()
     client = TestClient(app)
@@ -233,6 +237,25 @@ def apply_async(self, **kwargs):
     assert resp.status_code == 500
 
 
+def test_process_sync_endpoint_http_exception(monkeypatch):
+    from backend.apps import data_process_app as app_module
+
+    class _PSyncHTTP:
+        def apply_async(self, **kwargs):
+            raise HTTPException(
+                status_code=HTTPStatus.BAD_REQUEST, detail="bad req")
+
+    monkeypatch.setattr(app_module, "process_sync", _PSyncHTTP(), raising=True)
+
+    app = _build_app()
+    client = TestClient(app)
+    resp = client.post(
+        "/tasks/process",
+        data={"source": "/tmp/a.txt", "source_type": "local"},
+    )
+    assert resp.status_code == HTTPStatus.BAD_REQUEST
+
+
 def test_batch_tasks_success():
     app = _build_app()
     client = TestClient(app)
@@ -249,20 +272,37 @@ def test_batch_tasks_success():
 
 def test_batch_tasks_error(monkeypatch):
     # Make service raise
-    svc_mod = sys.modules["services.data_process_service"]
-    service: _ServiceStub = svc_mod.get_data_process_service()
+    from backend.apps import data_process_app as app_module
 
     async def err(*args, **kwargs):
         raise RuntimeError("x")
 
-    service.create_batch_tasks_impl = err  # type: ignore
+    monkeypatch.setattr(app_module.service,
+                        "create_batch_tasks_impl", err, raising=True)
 
     app = _build_app()
     client = TestClient(app)
     resp = client.post("/tasks/batch", json={"sources": []}, headers={"Authorization": "Bearer t"})
     assert resp.status_code == 500
 
 
+def test_batch_tasks_http_exception(monkeypatch):
+    from backend.apps import data_process_app as app_module
+
+    async def err_http(*args, **kwargs):
+        raise HTTPException(
+            status_code=HTTPStatus.NOT_ACCEPTABLE, detail="bad batch")
+
+    monkeypatch.setattr(app_module.service,
+                        "create_batch_tasks_impl", err_http, raising=True)
+
+    app = _build_app()
+    client = TestClient(app)
+    resp = client.post(
+        "/tasks/batch", json={"sources": []}, headers={"Authorization": "Bearer t"})
+    assert resp.status_code == HTTPStatus.NOT_ACCEPTABLE
+
+
 def test_load_image_success_and_not_found():
     app = _build_app()
     client = TestClient(app)
@@ -274,19 +314,37 @@ def test_load_image_success_and_not_found():
 
 
 def test_load_image_internal_error(monkeypatch):
-    svc_mod = sys.modules["services.data_process_service"]
-    service: _ServiceStub = svc_mod.get_data_process_service()
+    from backend.apps import data_process_app as app_module
 
     async def err(url: str):
         raise RuntimeError("bad")
 
-    service.load_image = err  # type: ignore
+    monkeypatch.setattr(app_module.service, "load_image", err, raising=True)
     app = _build_app()
     client = TestClient(app)
     resp = client.get("/tasks/load_image", params={"url": "x"})
     assert resp.status_code == 500
 
 
+def test_filter_important_image_http_exception(monkeypatch):
+    from backend.apps import data_process_app as app_module
+
+    async def err_http(*args, **kwargs):
+        raise HTTPException(
+            status_code=HTTPStatus.BAD_REQUEST, detail="bad image")
+
+    monkeypatch.setattr(app_module.service,
+                        "filter_important_image", err_http, raising=True)
+
+    app = _build_app()
+    client = TestClient(app)
+    resp = client.post(
+        "/tasks/filter_important_image",
+        data={"image_url": "u"},
+    )
+    assert resp.status_code == HTTPStatus.BAD_REQUEST
+
+
 def test_list_tasks():
     app = _build_app()
     client = TestClient(app)
@@ -342,19 +400,54 @@ def test_process_text_file_success_and_error(tmp_path):
     assert bad.status_code == 500
 
 
+def test_process_text_file_http_exception(monkeypatch):
+    from backend.apps import data_process_app as app_module
+
+    async def err_http(*args, **kwargs):
+        raise HTTPException(
+            status_code=HTTPStatus.UNPROCESSABLE_ENTITY, detail="bad file")
+
+    monkeypatch.setattr(app_module.service,
+                        "process_uploaded_text_file", err_http, raising=True)
+
+    app = _build_app()
+    client = TestClient(app)
+    files = {"file": ("x.txt", b"hello", "text/plain")}
+    resp = client.post("/tasks/process_text_file", files=files,
+                       data={"chunking_strategy": "basic"})
+    assert resp.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
+
+
 def test_convert_state_success_and_error(monkeypatch):
     app = _build_app()
     client = TestClient(app)
     ok = client.post("/tasks/convert_state", json={"process_state": "SUCCESS", "forward_state": "SUCCESS"})
     assert ok.status_code == 200 and ok.json()["state"] == "COMPLETED"
 
     # Make service raise
-    svc_mod = sys.modules["services.data_process_service"]
-    service: _ServiceStub = svc_mod.get_data_process_service()
+    from backend.apps import data_process_app as app_module
     def raise_convert(*args, **kwargs):
         raise RuntimeError("x")
-    service.convert_celery_states_to_custom = raise_convert  # type: ignore
+    monkeypatch.setattr(
+        app_module.service, "convert_celery_states_to_custom", raise_convert, raising=True)
     err = client.post("/tasks/convert_state", json={"process_state": "PENDING", "forward_state": ""})
     assert err.status_code == 500
 
 
+def test_convert_state_http_exception(monkeypatch):
+    app = _build_app()
+    client = TestClient(app)
+
+    from backend.apps import data_process_app as app_module
+
+    def raise_convert_http(*args, **kwargs):
+        raise HTTPException(
+            status_code=HTTPStatus.NOT_ACCEPTABLE, detail="bad convert")
+
+    monkeypatch.setattr(
+        app_module.service, "convert_celery_states_to_custom", raise_convert_http, raising=True
+    )
+
+    resp = client.post("/tasks/convert_state",
+                       json={"process_state": "PENDING", "forward_state": ""})
+    assert resp.status_code == HTTPStatus.NOT_ACCEPTABLE