✨ Multi modal agent.

delete unuse function

Author: Zhi-a

backend/apps/file_management_app.py

@@ -1,16 +1,13 @@
 import logging
-import os
 from http import HTTPStatus
 from typing import List, Optional
 
-from fastapi import APIRouter, Body, File, Form, Header, HTTPException, Path as PathParam, Query, Request, UploadFile
+from fastapi import APIRouter, Body, File, Form, Header, HTTPException, Path as PathParam, Query, UploadFile
 from fastapi.responses import JSONResponse, RedirectResponse, StreamingResponse
 
 from consts.model import ProcessParams
 from services.file_management_service import upload_to_minio, upload_files_impl, \
-    get_file_url_impl, get_file_stream_impl, delete_file_impl, list_files_impl, \
-    preprocess_files_generator
-from utils.auth_utils import get_current_user_info
+    get_file_url_impl, get_file_stream_impl, delete_file_impl, list_files_impl
 from utils.file_management_utils import trigger_data_process
 
 logger = logging.getLogger("file_management_app")
@@ -271,61 +268,3 @@ async def get_storage_file_batch_urls(
         "failed_count": sum(1 for r in results if not r.get("success", False)),
         "results": results
     }
-
-
-@file_management_runtime_router.post("/preprocess")
-async def agent_preprocess_api(
-        request: Request, query: str = Form(...),
-        files: List[UploadFile] = File(...),
-        authorization: Optional[str] = Header(None)
-):
-    """
-    Preprocess uploaded files and return streaming response
-    """
-    try:
-        # Pre-read and cache all file contents
-        user_id, tenant_id, language = get_current_user_info(
-            authorization, request)
-        file_cache = []
-        for file in files:
-            try:
-                content = await file.read()
-                file_cache.append({
-                    "filename": file.filename or "",
-                    "content": content,
-                    "ext": os.path.splitext(file.filename or "")[1].lower()
-                })
-            except Exception as e:
-                file_cache.append({
-                    "filename": file.filename or "",
-                    "error": str(e)
-                })
-
-        # Generate unique task ID for this preprocess operation
-        import uuid
-        task_id = str(uuid.uuid4())
-        conversation_id = request.query_params.get("conversation_id")
-        if conversation_id:
-            conversation_id = int(conversation_id)
-        else:
-            conversation_id = -1  # Default for cases without conversation_id
-
-        # Call service layer to generate streaming response
-        return StreamingResponse(
-            preprocess_files_generator(
-                query=query,
-                file_cache=file_cache,
-                tenant_id=tenant_id,
-                language=language,
-                task_id=task_id,
-                conversation_id=conversation_id
-            ),
-            media_type="text/event-stream",
-            headers={
-                "Cache-Control": "no-cache",
-                "Connection": "keep-alive"
-            }
-        )
-    except Exception as e:
-        raise HTTPException(
-            status_code=500, detail=f"File preprocessing error: {str(e)}")

backend/services/file_management_service.py

@@ -20,9 +20,7 @@
     list_files
 )
 from services.vectordatabase_service import ElasticSearchService, get_vector_db_core
-from utils.attachment_utils import convert_image_to_text, convert_long_text_to_text
 from utils.config_utils import tenant_config_manager, get_model_name_from_config
-from utils.prompt_template_utils import get_file_processing_messages_template
 from utils.file_management_utils import save_upload_file
 
 from nexent import MessageObserver
@@ -188,225 +186,6 @@ async def list_files_impl(prefix: str, limit: Optional[int] = None):
     return files
 
 
-def get_parsing_file_data(index: int, total_files: int, filename: str) -> dict:
-    """
-    Get structured data for parsing file message
-
-    Args:
-        index: Current file index (0-based)
-        total_files: Total number of files
-        filename: Name of the file being parsed
-
-    Returns:
-        dict: Structured data with parameters for internationalization
-    """
-    return {
-        "params": {
-            "index": index + 1,
-            "total": total_files,
-            "filename": filename
-        }
-    }
-
-
-def get_truncation_data(filename: str, truncation_percentage: int) -> dict:
-    """
-    Get structured data for truncation message
-
-    Args:
-        filename: Name of the file being truncated
-        truncation_percentage: Percentage of content that was read
-
-    Returns:
-        dict: Structured data with parameters for internationalization
-    """
-    return {
-        "params": {
-            "filename": filename,
-            "percentage": truncation_percentage
-        }
-    }
-
-
-async def preprocess_files_generator(
-    query: str,
-    file_cache: List[dict],
-    tenant_id: str,
-    language: str,
-    task_id: str,
-    conversation_id: int
-) -> AsyncGenerator[str, None]:
-    """
-    Generate streaming response for file preprocessing
-
-    Args:
-        query: User query string
-        file_cache: List of cached file data
-        tenant_id: Tenant ID
-        language: Language preference
-        task_id: Unique task ID
-        conversation_id: Conversation ID
-
-    Yields:
-        str: JSON formatted streaming messages
-    """
-    file_descriptions = []
-    total_files = len(file_cache)
-
-    # Create and register the preprocess task
-    task = asyncio.current_task()
-    if task:
-        preprocess_manager.register_preprocess_task(
-            task_id, conversation_id, task)
-
-    try:
-        for index, file_data in enumerate(file_cache):
-            if task and task.done():
-                logger.info(f"Preprocess task {task_id} was cancelled")
-                break
-
-            progress = int((index / total_files) * 100)
-            progress_message = json.dumps({
-                "type": "progress",
-                "progress": progress,
-                "message_data": get_parsing_file_data(index, total_files, file_data['filename'])
-            }, ensure_ascii=False)
-            yield f"data: {progress_message}\n\n"
-            await asyncio.sleep(0.1)
-
-            try:
-                # Check if file already has an error
-                if "error" in file_data:
-                    raise Exception(file_data["error"])
-
-                description = ""
-                truncation_percentage = None
-                file_descriptions.append(description)
-
-                # Send processing result for each file
-                file_message_data = {
-                    "type": "file_processed",
-                    "filename": file_data["filename"],
-                    "description": description
-                }
-                file_message = json.dumps(
-                    file_message_data, ensure_ascii=False)
-                yield f"data: {file_message}\n\n"
-                await asyncio.sleep(0.1)
-
-                # Send truncation notice immediately if file was truncated
-                if truncation_percentage is not None and int(truncation_percentage) < 100:
-                    if int(truncation_percentage) == 0:
-                        truncation_percentage = "< 1"
-
-                    truncation_message = json.dumps({
-                        "type": "truncation",
-                        "message_data": get_truncation_data(file_data['filename'], truncation_percentage)
-                    }, ensure_ascii=False)
-                    yield f"data: {truncation_message}\n\n"
-                    await asyncio.sleep(0.1)
-            except Exception as e:
-                error_description = f"Error parsing file {file_data['filename']}: {str(e)}"
-                logger.exception(error_description)
-                file_descriptions.append(error_description)
-                error_message = json.dumps({
-                    "type": "error",
-                    "filename": file_data["filename"],
-                    "message": error_description
-                }, ensure_ascii=False)
-                yield f"data: {error_message}\n\n"
-                await asyncio.sleep(0.1)
-
-        # Send completion message
-        complete_message = json.dumps({
-            "type": "complete",
-            "progress": 100,
-            "final_query": query
-        }, ensure_ascii=False)
-        yield f"data: {complete_message}\n\n"
-    finally:
-        preprocess_manager.unregister_preprocess_task(task_id)
-
-
-async def process_image_file(query: str, filename: str, file_content: bytes, tenant_id: str, language: str = LANGUAGE["ZH"]) -> str:
-    """
-    Process image file, convert to text using external API
-    """
-    # Load messages based on language
-    messages = get_file_processing_messages_template(language)
-    
-    try:
-        image_stream = BytesIO(file_content)
-        text = convert_image_to_text(query, image_stream, tenant_id, language)
-        return messages["IMAGE_CONTENT_SUCCESS"].format(filename=filename, content=text)
-    except Exception as e:
-        return messages["IMAGE_CONTENT_ERROR"].format(filename=filename, error=str(e))
-
-
-async def process_text_file(query: str, filename: str, file_content: bytes, tenant_id: str, language: str = LANGUAGE["ZH"]) -> tuple[str, Optional[str]]:
-    """
-    Process text file, convert to text using external API
-    """
-    # Load messages based on language
-    messages = get_file_processing_messages_template(language)
-    
-    # file_content is byte data, need to send to API through file upload
-    data_process_service_url = DATA_PROCESS_SERVICE
-    api_url = f"{data_process_service_url}/tasks/process_text_file"
-    logger.info(f"Processing text file {filename} with API: {api_url}")
-
-    try:
-        # Upload byte data as a file
-        files = {
-            'file': (filename, file_content, 'application/octet-stream')
-        }
-        data = {
-            'chunking_strategy': 'basic',
-            'timeout': 60
-        }
-        async with httpx.AsyncClient() as client:
-            response = await client.post(api_url, files=files, data=data, timeout=60)
-
-        if response.status_code == 200:
-            result = response.json()
-            raw_text = result.get("text", "")
-            logger.info(
-                f"File processed successfully: {raw_text[:200]}...{raw_text[-200:]}...， length: {len(raw_text)}")
-        else:
-            error_detail = response.json().get('detail', 'unknown error') if response.headers.get(
-                'content-type', '').startswith('application/json') else response.text
-            logger.error(
-                f"File processing failed (status code: {response.status_code}): {error_detail}")
-            raise Exception(
-                messages["FILE_PROCESSING_ERROR"].format(status_code=response.status_code, error_detail=error_detail))
-
-    except Exception as e:
-        return messages["FILE_CONTENT_ERROR"].format(filename=filename, error=str(e)), None
-
-    try:
-        text, truncation_percentage = convert_long_text_to_text(
-            query, raw_text, tenant_id, language)
-        return messages["FILE_CONTENT_SUCCESS"].format(filename=filename, content=text), truncation_percentage
-    except Exception as e:
-        return messages["FILE_CONTENT_ERROR"].format(filename=filename, error=str(e)), None
-
-
-def get_file_description(files: List[UploadFile]) -> str:
-    """
-    Generate file description text
-    """
-    if not files:
-        return "User provided some reference files:\nNo files provided"
-
-    description = "User provided some reference files:\n"
-    for file in files:
-        ext = os.path.splitext(file.filename or "")[1].lower()
-        if ext in ['.jpg', '.jpeg', '.png', '.gif', '.bmp']:
-            description += f"- Image file {file.filename or ''}\n"
-        else:
-            description += f"- File {file.filename or ''}\n"
-    return description
-
 def get_llm_model(tenant_id: str):
     # Get the tenant config
     main_model_config = tenant_config_manager.get_model_config(

test/backend/app/test_file_management_app.py

@@ -357,66 +357,3 @@ def fake_get(object_name, expires):
     assert any(item["object_name"] == "bad" and item["success"] is False for item in out["results"])
 
 
-@pytest.mark.asyncio
-async def test_agent_preprocess_api_success(monkeypatch):
-    # Patch get_current_user_info
-    monkeypatch.setattr(file_management_app, "get_current_user_info", lambda a, r: ("u", "t", "en"))
-
-    # Provide an async generator object for streaming
-    async def gen():
-        yield "data: {\\\"type\\\": \\\"complete\\\"}\n\n"
-
-    monkeypatch.setattr(file_management_app, "preprocess_files_generator", lambda **_: gen())
-
-    # Mock files
-    f1 = make_upload_file("a.txt", b"hello")
-    f2 = make_upload_file("b.jpg", b"img")
-
-    # Minimal ASGI scope for Request
-    from starlette.requests import Request
-    scope = {
-        "type": "http",
-        "method": "POST",
-        "path": "/file/preprocess",
-        "headers": [],
-        "query_string": b"conversation_id=42",
-    }
-    req = Request(scope)
-
-    resp = await file_management_app.agent_preprocess_api(
-        request=req, query="q", files=[f1, f2], authorization="Bearer x"
-    )
-    assert resp.media_type == "text/event-stream"
-    assert resp.headers.get("Cache-Control") == "no-cache"
-    # Consume a small portion of stream
-    chunks = []
-    async for part in resp.body_iterator:  # type: ignore[attr-defined]
-        chunks.append(part)
-        break
-    assert chunks
-
-
-@pytest.mark.asyncio
-async def test_agent_preprocess_api_error_from_auth(monkeypatch):
-    def boom_auth(a, r):
-        raise RuntimeError("auth failed")
-
-    monkeypatch.setattr(file_management_app, "get_current_user_info", boom_auth)
-
-    from starlette.requests import Request
-    scope = {
-        "type": "http",
-        "method": "POST",
-        "path": "/file/preprocess",
-        "headers": [],
-        "query_string": b"",
-    }
-    req = Request(scope)
-
-    with pytest.raises(Exception) as ei:
-        await file_management_app.agent_preprocess_api(
-            request=req, query="q", files=[make_upload_file("a.txt")], authorization=None
-        )
-    assert "File preprocessing error" in str(ei.value)
-
-