From 044a40175b4c25365ed22b37927ba24a7dc7cb04 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9C=B1=E6=BD=AE?= Date: Thu, 23 Oct 2025 16:31:37 +0800 Subject: [PATCH] add robot_type --- fastapi_app.py | 5 +- prompt/system_prompt_default.md | 2 +- utils/api_models.py | 1 + utils/file_loaded_agent_manager.py | 6 +- utils/prompt_loader.py | 89 +++++++++++++++--------------- 5 files changed, 55 insertions(+), 48 deletions(-) diff --git a/fastapi_app.py b/fastapi_app.py index 1be3334..e8c7bba 100644 --- a/fastapi_app.py +++ b/fastapi_app.py @@ -430,7 +430,7 @@ async def chat_completions(request: ChatRequest, authorization: Optional[str] = project_dir = "" # 收集额外参数作为 generate_cfg - exclude_fields = {'messages', 'model', 'model_server', 'unique_id', 'language', 'tool_response', 'system_prompt', 'mcp_settings' ,'stream'} + exclude_fields = {'messages', 'model', 'model_server', 'unique_id', 'language', 'tool_response', 'system_prompt', 'mcp_settings' ,'stream', 'robot_type'} generate_cfg = {k: v for k, v in request.model_dump().items() if k not in exclude_fields} # 从全局管理器获取或创建助手实例（配置读取逻辑已在agent_manager内部处理） @@ -443,7 +443,8 @@ async def chat_completions(request: ChatRequest, authorization: Optional[str] = generate_cfg=generate_cfg, language=request.language, system_prompt=request.system_prompt, - mcp_settings=request.mcp_settings + mcp_settings=request.mcp_settings, + robot_type=request.robot_type ) # 构建包含项目信息的消息上下文 messages = [] diff --git a/prompt/system_prompt_default.md b/prompt/system_prompt_default.md index 421637c..b7cf57d 100644 --- a/prompt/system_prompt_default.md +++ b/prompt/system_prompt_default.md @@ -194,4 +194,4 @@ **工具调用前声明**：每次调用工具之前，必须输出工具选择理由和预期结果 **工具调用后评估**：每次调用工具之后，必须输出结果分析和下一步规划 **语言要求**：所有用户交互和结果输出，必须使用[{language}] - +{extra_prompt} diff --git a/utils/api_models.py b/utils/api_models.py index 4a81a95..98e9c98 100644 --- a/utils/api_models.py +++ b/utils/api_models.py @@ -49,6 +49,7 @@ class ChatRequest(BaseModel): tool_response: Optional[bool] = False system_prompt: Optional[str] = None mcp_settings: Optional[List[Dict]] = None + robot_type: Optional[str] = "AGENT" class FileProcessRequest(BaseModel): diff --git a/utils/file_loaded_agent_manager.py b/utils/file_loaded_agent_manager.py index 0fd68b1..3d73b18 100644 --- a/utils/file_loaded_agent_manager.py +++ b/utils/file_loaded_agent_manager.py @@ -81,7 +81,8 @@ class FileLoadedAgentManager: generate_cfg: Optional[Dict] = None, language: Optional[str] = None, system_prompt: Optional[str] = None, - mcp_settings: Optional[List[Dict]] = None) -> Assistant: + mcp_settings: Optional[List[Dict]] = None, + robot_type: Optional[str] = "AGENT") -> Assistant: """获取或创建文件预加载的助手实例 Args: @@ -94,6 +95,7 @@ class FileLoadedAgentManager: language: 语言代码，用于选择对应的系统提示词 system_prompt: 可选的系统提示词，优先级高于项目配置 mcp_settings: 可选的MCP设置，优先级高于项目配置 + robot_type: 机器人类型，取值 AGENT/CATALOG_AGENT Returns: Assistant: 配置好的助手实例 @@ -101,7 +103,7 @@ class FileLoadedAgentManager: import os # 实现参数优先级逻辑：传入参数 > 项目配置 > 默认配置 - final_system_prompt = load_system_prompt(project_dir, language, system_prompt) + final_system_prompt = load_system_prompt(project_dir, language, system_prompt, robot_type) final_mcp_settings = load_mcp_settings(project_dir, mcp_settings) cache_key = self._get_cache_key(unique_id) diff --git a/utils/prompt_loader.py b/utils/prompt_loader.py index 9f0452d..fb370ab 100644 --- a/utils/prompt_loader.py +++ b/utils/prompt_loader.py @@ -7,59 +7,63 @@ import json from typing import List, Dict, Optional -def load_system_prompt(project_dir: str, language: str = None, system_prompt: str=None) -> str: +def load_system_prompt(project_dir: str, language: str = None, system_prompt: str=None, robot_type: str = "AGENT") -> str: + if robot_type == "AGENT": + return system_prompt or "" + if robot_type == "CATALOG_AGENT": + """ + 优先使用项目目录的system_prompt，没有才使用默认的system_prompt_default.md + Args: + project_dir: 项目目录路径 + language: 语言代码，如 'zh', 'en', 'jp' 等（此参数将被忽略） + system_prompt: 可选的系统提示词，优先级高于项目配置 + robot_type: 机器人类型，取值 AGENT/CATALOG_AGENT - """ - 优先使用项目目录的system_prompt，没有才使用默认的system_prompt_default.md - - Args: - project_dir: 项目目录路径 - language: 语言代码，如 'zh', 'en', 'jp' 等（此参数将被忽略） - - Returns: - str: 加载到的系统提示词内容，如果都未找到则返回空字符串 - """ - - # 1. 优先读取项目目录中的system_prompt - if not system_prompt: + Returns: + str: 加载到的系统提示词内容，如果都未找到则返回空字符串 + """ + system_prompt_default = None + # 1. 优先读取项目目录中的system_prompt system_prompt_file = os.path.join(project_dir, "system_prompt.md") if os.path.exists(system_prompt_file): try: with open(system_prompt_file, 'r', encoding='utf-8') as f: - system_prompt = f.read() + system_prompt_default = f.read() print(f"Using project-specific system prompt") except Exception as e: print(f"Failed to load project system prompt: {str(e)}") - system_prompt = None - - # 2. 如果项目目录没有，使用默认提示词 - if not system_prompt: - try: - default_prompt_file = os.path.join("prompt", "system_prompt_default.md") - with open(default_prompt_file, 'r', encoding='utf-8') as f: - system_prompt = f.read() - print(f"Using default system prompt from prompt folder") - except Exception as e: - print(f"Failed to load default system prompt: {str(e)}") - system_prompt = None - - readme = "" - readme_path = os.path.join(project_dir, "README.md") - if os.path.exists(readme_path): - with open(readme_path, "r", encoding="utf-8") as f: - readme = f.read().strip() + system_prompt_default = None - # 获取语言显示名称 - language_display_map = { - 'zh': '中文', - 'en': 'English', - 'ja': '日本語', - 'jp': '日本語' - } - language_display = language_display_map.get(language, language if language else 'English') - return system_prompt.replace("{readme}", str(readme)).replace("{language}", language_display) or "" + # 2. 如果项目目录没有，使用默认提示词 + if not system_prompt_default: + try: + default_prompt_file = os.path.join("prompt", "system_prompt_default.md") + with open(default_prompt_file, 'r', encoding='utf-8') as f: + system_prompt_default = f.read() + print(f"Using default system prompt from prompt folder") + except Exception as e: + print(f"Failed to load default system prompt: {str(e)}") + system_prompt_default = None + readme = "" + readme_path = os.path.join(project_dir, "README.md") + if os.path.exists(readme_path): + with open(readme_path, "r", encoding="utf-8") as f: + readme = f.read().strip() + + # 获取语言显示名称 + language_display_map = { + 'zh': '中文', + 'en': 'English', + 'ja': '日本語', + 'jp': '日本語' + } + language_display = language_display_map.get(language, language if language else 'English') + + return system_prompt_default.replace("{readme}", str(readme)).replace("{language}", language_display).replace("{extra_prompt}", system_prompt or "") or "" + else: + return system_prompt or "" def get_available_prompt_languages() -> list: """ @@ -161,7 +165,6 @@ def load_mcp_settings(project_dir: str, mcp_settings: list=None) -> List[Dict]: dataset_dir = os.path.join(project_dir, "dataset") # 替换 MCP 配置中的 {dataset_dir} 占位符 mcp_settings = replace_mcp_placeholders(mcp_settings, dataset_dir) - print(mcp_settings) return mcp_settings