♻️ Refactor: Sort backend imports #1037

Author: WMC001

backend/services/agent_service.py

@@ -2,30 +2,50 @@
 import json
 import logging
 import os
-from collections import deque
 import uuid
+from collections import deque
 
 from fastapi import Header, Request
 from fastapi.responses import JSONResponse, StreamingResponse
+from nexent.core.agents.run_agent import agent_run
+from nexent.memory.memory_service import clear_memory
 
 from agents.agent_run_manager import agent_run_manager
-from agents.create_agent_info import create_agent_run_info
-from agents.create_agent_info import create_tool_config_list
+from agents.create_agent_info import create_agent_run_info, create_tool_config_list
 from agents.preprocess_manager import preprocess_manager
 from consts.exceptions import AgentRunException, MemoryPreparationException
-from consts.model import AgentInfoRequest, ExportAndImportAgentInfo, ExportAndImportDataFormat, ToolInstanceInfoRequest, \
-    ToolSourceEnum, MCPInfo
-from consts.model import AgentRequest
-from database.agent_db import create_agent, search_blank_sub_agent_by_main_agent_id, \
-    search_agent_info_by_agent_id, update_agent, delete_agent_by_id, query_all_agent_info_by_tenant_id, \
-    query_sub_agents_id_list, insert_related_agent, delete_agent_relationship, search_agent_id_by_agent_name, \
-    delete_related_agent
-from database.remote_mcp_db import get_mcp_server_by_name_and_tenant, check_mcp_name_exists
-from database.tool_db import create_or_update_tool_by_tool_info, query_all_tools, query_all_enabled_tool_instances, \
-    search_tools_for_sub_agent, check_tool_is_available, delete_tools_by_agent_id
-from nexent.core.agents.run_agent import agent_run
-from nexent.memory.memory_service import clear_memory
-from services.conversation_management_service import save_conversation_user, save_conversation_assistant
+from consts.model import (
+    AgentInfoRequest,
+    AgentRequest,
+    ExportAndImportAgentInfo,
+    ExportAndImportDataFormat,
+    MCPInfo,
+    ToolInstanceInfoRequest,
+    ToolSourceEnum
+)
+from database.agent_db import (
+    create_agent,
+    delete_agent_by_id,
+    delete_agent_relationship,
+    delete_related_agent,
+    insert_related_agent,
+    query_all_agent_info_by_tenant_id,
+    query_sub_agents_id_list,
+    search_agent_id_by_agent_name,
+    search_agent_info_by_agent_id,
+    search_blank_sub_agent_by_main_agent_id,
+    update_agent
+)
+from database.remote_mcp_db import check_mcp_name_exists, get_mcp_server_by_name_and_tenant
+from database.tool_db import (
+    check_tool_is_available,
+    create_or_update_tool_by_tool_info,
+    delete_tools_by_agent_id,
+    query_all_enabled_tool_instances,
+    query_all_tools,
+    search_tools_for_sub_agent
+)
+from services.conversation_management_service import save_conversation_assistant, save_conversation_user
 from services.memory_config_service import build_memory_context
 from services.remote_mcp_service import add_remote_mcp_server_list
 from services.tool_configuration_service import update_tool_list
@@ -709,7 +729,8 @@ def _memory_token(message_text: str) -> str:
             ):
                 yield data_chunk
         except Exception as run_exc:
-            logger.error(f"Agent run error after memory failure: {str(run_exc)}")
+            logger.error(
+                f"Agent run error after memory failure: {str(run_exc)}")
             raise AgentRunException(f"Agent run error: {str(run_exc)}")
     except Exception as e:
         logger.error(f"Generate stream with memory error: {str(e)}")
@@ -758,7 +779,8 @@ async def run_agent_stream(
     """
     # Save user message only if not in debug mode (before streaming starts)
     if not agent_request.is_debug:
-        save_messages(agent_request, target="user", authorization=authorization)
+        save_messages(agent_request, target="user",
+                      authorization=authorization)
 
     # Choose streaming strategy based on user's memory switch
     resolved_user_id, resolved_tenant_id, _ = _resolve_user_tenant_language(

backend/services/config_sync_service.py

@@ -1,11 +1,25 @@
 import logging
 
-from consts.const import DEFAULT_APP_NAME_ZH, DEFAULT_APP_NAME_EN, DEFAULT_APP_DESCRIPTION_ZH, \
-    DEFAULT_APP_DESCRIPTION_EN, DEFAULT_APP_ICON_URL, MODEL_CONFIG_MAPPING, APP_NAME, APP_DESCRIPTION, ICON_TYPE, \
-    AVATAR_URI, CUSTOM_ICON_URL
+from consts.const import (
+    APP_DESCRIPTION,
+    APP_NAME,
+    AVATAR_URI,
+    CUSTOM_ICON_URL,
+    DEFAULT_APP_DESCRIPTION_EN,
+    DEFAULT_APP_DESCRIPTION_ZH,
+    DEFAULT_APP_ICON_URL,
+    DEFAULT_APP_NAME_EN,
+    DEFAULT_APP_NAME_ZH,
+    ICON_TYPE,
+    MODEL_CONFIG_MAPPING
+)
 from database.model_management_db import get_model_id_by_display_name
-from utils.config_utils import tenant_config_manager, get_env_key, safe_value, \
-    get_model_name_from_config
+from utils.config_utils import (
+    get_env_key,
+    get_model_name_from_config,
+    safe_value,
+    tenant_config_manager
+)
 
 logger = logging.getLogger("config_sync_service")
 

backend/services/conversation_management_service.py

@@ -2,23 +2,36 @@
 import json
 import logging
 from datetime import datetime
-from typing import List, Optional, Dict, Any
+from typing import Any, Dict, List, Optional
 
 from fastapi import Header
 from jinja2 import StrictUndefined, Template
+from nexent.core.utils.observer import ProcessType
 from smolagents import OpenAIServerModel
 
-from consts.model import MessageRequest, ConversationResponse, AgentRequest, MessageUnit
-from database.conversation_db import create_conversation_message, create_source_search, create_message_units, \
-    create_source_image, rename_conversation, get_conversation_list, get_conversation_history, \
-    get_source_images_by_message, \
-    get_source_images_by_conversation, get_source_searches_by_message, get_source_searches_by_conversation, \
-    delete_conversation, get_conversation, create_conversation, update_message_opinion, get_message_id_by_index
-from nexent.core.utils.observer import ProcessType
+from consts.model import AgentRequest, ConversationResponse, MessageRequest, MessageUnit
+from database.conversation_db import (
+    create_conversation,
+    create_conversation_message,
+    create_message_units,
+    create_source_image,
+    create_source_search,
+    delete_conversation,
+    get_conversation,
+    get_conversation_history,
+    get_conversation_list,
+    get_message_id_by_index,
+    get_source_images_by_conversation,
+    get_source_images_by_message,
+    get_source_searches_by_conversation,
+    get_source_searches_by_message,
+    rename_conversation,
+    update_message_opinion
+)
 from utils.auth_utils import get_current_user_id
-from utils.config_utils import tenant_config_manager, get_model_name_from_config
+from utils.config_utils import get_model_name_from_config, tenant_config_manager
 from utils.prompt_template_utils import get_generate_title_prompt_template
-from utils.str_utils import remove_think_tags, add_no_think_token
+from utils.str_utils import add_no_think_token, remove_think_tags
 
 logger = logging.getLogger("conversation_management_service")
 
@@ -56,7 +69,8 @@ def save_message(request: MessageRequest, authorization: Optional[str] = Header(
         # Validate conversation_id
         conversation_id = message_data.get('conversation_id')
         if not conversation_id:
-            raise Exception("conversation_id is required, please call /conversation/create to create a conversation first")
+            raise Exception(
+                "conversation_id is required, please call /conversation/create to create a conversation first")
 
         # Process different types of message units
         message_units = message_data['message']
@@ -289,7 +303,8 @@ def update_conversation_title(conversation_id: int, title: str, user_id: str = N
     """
     success = rename_conversation(conversation_id, title, user_id)
     if not success:
-        raise Exception(f"Conversation {conversation_id} does not exist or has been deleted")
+        raise Exception(
+            f"Conversation {conversation_id} does not exist or has been deleted")
     return success
 
 
@@ -342,7 +357,8 @@ def rename_conversation_service(conversation_id: int, name: str, user_id: str) -
     try:
         success = rename_conversation(conversation_id, name, user_id)
         if not success:
-            raise Exception(f"Conversation {conversation_id} does not exist or has been deleted")
+            raise Exception(
+                f"Conversation {conversation_id} does not exist or has been deleted")
         return True
     except Exception as e:
         logging.error(f"Failed to rename conversation: {str(e)}")
@@ -363,7 +379,8 @@ def delete_conversation_service(conversation_id: int, user_id: str) -> bool:
     try:
         success = delete_conversation(conversation_id, user_id)
         if not success:
-            raise Exception(f"Conversation {conversation_id} does not exist or has been deleted")
+            raise Exception(
+                f"Conversation {conversation_id} does not exist or has been deleted")
         return True
     except Exception as e:
         logging.error(f"Failed to delete conversation: {str(e)}")
@@ -688,4 +705,4 @@ async def get_message_id_by_index_impl(conversation_id: int, message_index: int)
     message_id = get_message_id_by_index(conversation_id, message_index)
     if message_id is None:
         raise Exception("Message not found.")
-    return message_id
+    return message_id

backend/services/data_process_service.py

@@ -1,8 +1,11 @@
+import asyncio
 import base64
+import concurrent.futures
 import io
 import logging
 import os
 import tempfile
+import threading
 import time
 import warnings
 from typing import Optional, List, Dict, Any
@@ -13,14 +16,11 @@
 from PIL import Image
 from transformers import CLIPProcessor, CLIPModel
 
-from consts.model import BatchTaskRequest
-from data_process.utils import get_task_info, get_all_task_ids_from_redis
-import concurrent.futures
-import asyncio
-
 from consts.const import CLIP_MODEL_PATH, IMAGE_FILTER, REDIS_BACKEND_URL, REDIS_URL
+from consts.model import BatchTaskRequest
 from data_process.app import app as celery_app
 from data_process.tasks import process_and_forward
+from data_process.utils import get_task_info, get_all_task_ids_from_redis
 
 # Configure logging
 logger = logging.getLogger("data_process.service")
@@ -45,9 +45,8 @@ def __init__(self):
 
         self._inspector = None
         self._inspector_last_time = 0
-        self._inspector_ttl = 60  # inspector缓存时间，秒
+        self._inspector_ttl = 60  # Inspector cache time in seconds
         self._inspector_lock = None
-        import threading
         self._inspector_lock = threading.Lock()
 
     def _init_redis_client(self):
@@ -180,8 +179,10 @@ def get_reserved():
                     # Add to the set, duplicates will be handled
                     task_ids.add(task_id)
             except Exception as redis_error:
-                logger.warning(f"Failed to query Redis for stored task IDs: {str(redis_error)}")
-            logger.debug(f"Total unique task IDs collected (inspector + Redis): {len(task_ids)}")
+                logger.warning(
+                    f"Failed to query Redis for stored task IDs: {str(redis_error)}")
+            logger.debug(
+                f"Total unique task IDs collected (inspector + Redis): {len(task_ids)}")
             tasks = [get_task_info(task_id) for task_id in task_ids]
             all_task_infos = await asyncio.gather(*tasks, return_exceptions=True)
             for task_info in all_task_infos: