From 7d3f418768f24119b3ac98c644fef203af1bc78b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=85=E8=AF=BA=E7=8B=90?= <212194964+foxcyber907@users.noreply.github.com> Date: Tue, 30 Sep 2025 22:40:10 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BA=8C=E6=AD=A5=E5=B7=A5=E5=85=B7=E4=B8=8Eac?= =?UTF-8?q?tion?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/plugin_system/base/base_action.py | 80 +++++++++++++- src/plugin_system/base/base_tool.py | 128 +++++++++++++++++++++- src/plugin_system/base/component_types.py | 6 + src/plugin_system/core/tool_use.py | 53 ++++++++- 4 files changed, 262 insertions(+), 5 deletions(-) diff --git a/src/plugin_system/base/base_action.py b/src/plugin_system/base/base_action.py index 725619adb..e58076ba6 100644 --- a/src/plugin_system/base/base_action.py +++ b/src/plugin_system/base/base_action.py @@ -2,7 +2,7 @@ import time import asyncio from abc import ABC, abstractmethod -from typing import Tuple, Optional +from typing import Tuple, Optional, List, Dict, Any from src.common.logger import get_logger from src.chat.message_receive.chat_stream import ChatStream @@ -27,8 +27,21 @@ class BaseAction(ABC): - parallel_action: 是否允许并行执行 - random_activation_probability: 随机激活概率 - llm_judge_prompt: LLM判断提示词 + + 二步Action相关属性： + - is_two_step_action: 是否为二步Action + - step_one_description: 第一步的描述 + - sub_actions: 子Action列表 """ + # 二步Action相关类属性 + is_two_step_action: bool = False + """是否为二步Action。如果为True，Action将分两步执行：第一步选择操作，第二步执行具体操作""" + step_one_description: str = "" + """第一步的描述，用于向LLM展示Action的基本功能""" + sub_actions: List[Tuple[str, str, Dict[str, str]]] = [] + """子Action列表，格式为[(子Action名, 子Action描述, 子Action参数)]。仅在二步Action中使用""" + def __init__( self, action_data: dict, @@ -93,6 +106,13 @@ class BaseAction(ABC): self.associated_types: list[str] = getattr(self.__class__, "associated_types", []).copy() self.chat_type_allow: ChatType = getattr(self.__class__, "chat_type_allow", ChatType.ALL) + # 二步Action相关实例属性 + self.is_two_step_action: bool = getattr(self.__class__, "is_two_step_action", False) + self.step_one_description: str = getattr(self.__class__, "step_one_description", "") + self.sub_actions: List[Tuple[str, str, Dict[str, str]]] = getattr(self.__class__, "sub_actions", []).copy() + self._selected_sub_action: Optional[str] = None + """当前选择的子Action名称，用于二步Action的状态管理""" + # ============================================================================= # 便捷属性 - 直接在初始化时获取常用聊天信息（带类型注解） # ============================================================================= @@ -477,15 +497,73 @@ class BaseAction(ABC): action_require=getattr(cls, "action_require", []).copy(), associated_types=getattr(cls, "associated_types", []).copy(), chat_type_allow=getattr(cls, "chat_type_allow", ChatType.ALL), + # 二步Action相关属性 + is_two_step_action=getattr(cls, "is_two_step_action", False), + step_one_description=getattr(cls, "step_one_description", ""), + sub_actions=getattr(cls, "sub_actions", []).copy(), ) + async def handle_step_one(self) -> Tuple[bool, str]: + """处理二步Action的第一步 + + Returns: + Tuple[bool, str]: (是否执行成功, 回复文本) + """ + if not self.is_two_step_action: + return False, "此Action不是二步Action" + + # 检查action_data中是否包含选择的子Action + selected_action = self.action_data.get("selected_action") + if not selected_action: + # 第一步：展示可用的子Action + available_actions = [sub_action[0] for sub_action in self.sub_actions] + description = self.step_one_description or f"{self.action_name}支持以下操作" + + actions_list = "\n".join([f"- {action}: {desc}" for action, desc, _ in self.sub_actions]) + response = f"{description}\n\n可用操作：\n{actions_list}\n\n请选择要执行的操作。" + + return True, response + else: + # 验证选择的子Action是否有效 + valid_actions = [sub_action[0] for sub_action in self.sub_actions] + if selected_action not in valid_actions: + return False, f"无效的操作选择: {selected_action}。可用操作: {valid_actions}" + + # 保存选择的子Action + self._selected_sub_action = selected_action + + # 调用第二步执行 + return await self.execute_step_two(selected_action) + + async def execute_step_two(self, sub_action_name: str) -> Tuple[bool, str]: + """执行二步Action的第二步 + + Args: + sub_action_name: 子Action名称 + + Returns: + Tuple[bool, str]: (是否执行成功, 回复文本) + """ + if not self.is_two_step_action: + return False, "此Action不是二步Action" + + # 子类需要重写此方法来实现具体的第二步逻辑 + return False, f"二步Action必须实现execute_step_two方法来处理操作: {sub_action_name}" + @abstractmethod async def execute(self) -> Tuple[bool, str]: """执行Action的抽象方法，子类必须实现 + 对于二步Action，会自动处理第一步逻辑 + Returns: Tuple[bool, str]: (是否执行成功, 回复文本) """ + # 如果是二步Action，自动处理第一步 + if self.is_two_step_action: + return await self.handle_step_one() + + # 普通Action由子类实现 pass async def handle_action(self) -> Tuple[bool, str]: diff --git a/src/plugin_system/base/base_tool.py b/src/plugin_system/base/base_tool.py index b5022ea2a..84dc8b150 100644 --- a/src/plugin_system/base/base_tool.py +++ b/src/plugin_system/base/base_tool.py @@ -38,6 +38,14 @@ class BaseTool(ABC): semantic_cache_query_key: Optional[str] = None """用于语义缓存的查询参数键名。如果设置，将使用此参数的值进行语义相似度搜索""" + # 二步工具调用相关属性 + is_two_step_tool: bool = False + """是否为二步工具。如果为True，工具将分两步调用：第一步展示工具信息，第二步执行具体操作""" + step_one_description: str = "" + """第一步的描述，用于向LLM展示工具的基本功能""" + sub_tools: List[Tuple[str, str, List[Tuple[str, ToolParamType, str, bool, List[str] | None]]]] = [] + """子工具列表，格式为[(子工具名, 子工具描述, 子工具参数)]。仅在二步工具中使用""" + def __init__(self, plugin_config: Optional[dict] = None): self.plugin_config = plugin_config or {} # 直接存储插件配置字典 @@ -48,10 +56,64 @@ class BaseTool(ABC): Returns: dict: 工具定义字典 """ - if not cls.name or not cls.description or not cls.parameters: - raise NotImplementedError(f"工具类 {cls.__name__} 必须定义 name, description 和 parameters 属性") + if not cls.name or not cls.description: + raise NotImplementedError(f"工具类 {cls.__name__} 必须定义 name 和 description 属性") - return {"name": cls.name, "description": cls.description, "parameters": cls.parameters} + # 如果是二步工具，第一步只返回基本信息 + if cls.is_two_step_tool: + return { + "name": cls.name, + "description": cls.step_one_description or cls.description, + "parameters": [("action", ToolParamType.STRING, "选择要执行的操作", True, [sub_tool[0] for sub_tool in cls.sub_tools])] + } + else: + # 普通工具需要parameters + if not cls.parameters: + raise NotImplementedError(f"工具类 {cls.__name__} 必须定义 parameters 属性") + return {"name": cls.name, "description": cls.description, "parameters": cls.parameters} + + @classmethod + def get_step_two_tool_definition(cls, sub_tool_name: str) -> dict[str, Any]: + """获取二步工具的第二步定义 + + Args: + sub_tool_name: 子工具名称 + + Returns: + dict: 第二步工具定义字典 + """ + if not cls.is_two_step_tool: + raise ValueError(f"工具 {cls.name} 不是二步工具") + + # 查找对应的子工具 + for sub_name, sub_desc, sub_params in cls.sub_tools: + if sub_name == sub_tool_name: + return { + "name": f"{cls.name}_{sub_tool_name}", + "description": sub_desc, + "parameters": sub_params + } + + raise ValueError(f"未找到子工具: {sub_tool_name}") + + @classmethod + def get_all_sub_tool_definitions(cls) -> List[dict[str, Any]]: + """获取所有子工具的定义 + + Returns: + List[dict]: 所有子工具定义列表 + """ + if not cls.is_two_step_tool: + return [] + + definitions = [] + for sub_name, sub_desc, sub_params in cls.sub_tools: + definitions.append({ + "name": f"{cls.name}_{sub_name}", + "description": sub_desc, + "parameters": sub_params + }) + return definitions @classmethod def get_tool_info(cls) -> ToolInfo: @@ -79,8 +141,68 @@ class BaseTool(ABC): Returns: dict: 工具执行结果 """ + # 如果是二步工具，处理第一步调用 + if self.is_two_step_tool and "action" in function_args: + return await self._handle_step_one(function_args) + raise NotImplementedError("子类必须实现execute方法") + async def _handle_step_one(self, function_args: dict[str, Any]) -> dict[str, Any]: + """处理二步工具的第一步调用 + + Args: + function_args: 包含action参数的函数参数 + + Returns: + dict: 第一步执行结果，包含第二步的工具定义 + """ + action = function_args.get("action") + if not action: + return {"error": "缺少action参数"} + + # 查找对应的子工具 + sub_tool_found = None + for sub_name, sub_desc, sub_params in self.sub_tools: + if sub_name == action: + sub_tool_found = (sub_name, sub_desc, sub_params) + break + + if not sub_tool_found: + available_actions = [sub_tool[0] for sub_tool in self.sub_tools] + return {"error": f"未知的操作: {action}。可用操作: {available_actions}"} + + sub_name, sub_desc, sub_params = sub_tool_found + + # 返回第二步工具定义 + step_two_definition = { + "name": f"{self.name}_{sub_name}", + "description": sub_desc, + "parameters": sub_params + } + + return { + "type": "two_step_tool_step_one", + "content": f"已选择操作: {action}。请使用以下工具进行具体调用:", + "next_tool_definition": step_two_definition, + "selected_action": action + } + + async def execute_step_two(self, sub_tool_name: str, function_args: dict[str, Any]) -> dict[str, Any]: + """执行二步工具的第二步 + + Args: + sub_tool_name: 子工具名称 + function_args: 工具调用参数 + + Returns: + dict: 工具执行结果 + """ + if not self.is_two_step_tool: + raise ValueError(f"工具 {self.name} 不是二步工具") + + # 子类需要重写此方法来实现具体的第二步逻辑 + raise NotImplementedError("二步工具必须实现execute_step_two方法") + async def direct_execute(self, **kwargs: dict[str, Any]) -> dict[str, Any]: """直接执行工具函数(供插件调用) 通过该方法，插件可以直接调用工具，而不需要传入字典格式的参数 diff --git a/src/plugin_system/base/component_types.py b/src/plugin_system/base/component_types.py index 3fc943bd5..6d0590d43 100644 --- a/src/plugin_system/base/component_types.py +++ b/src/plugin_system/base/component_types.py @@ -142,6 +142,10 @@ class ActionInfo(ComponentInfo): mode_enable: ChatMode = ChatMode.ALL parallel_action: bool = False chat_type_allow: ChatType = ChatType.ALL # 允许的聊天类型 + # 二步Action相关属性 + is_two_step_action: bool = False # 是否为二步Action + step_one_description: str = "" # 第一步的描述 + sub_actions: List[Tuple[str, str, Dict[str, str]]] = field(default_factory=list) # 子Action列表 def __post_init__(self): super().__post_init__() @@ -153,6 +157,8 @@ class ActionInfo(ComponentInfo): self.action_require = [] if self.associated_types is None: self.associated_types = [] + if self.sub_actions is None: + self.sub_actions = [] self.component_type = ComponentType.ACTION diff --git a/src/plugin_system/core/tool_use.py b/src/plugin_system/core/tool_use.py index 1b2618f43..daa8244cf 100644 --- a/src/plugin_system/core/tool_use.py +++ b/src/plugin_system/core/tool_use.py @@ -55,6 +55,10 @@ class ToolExecutor: self.llm_model = LLMRequest(model_set=model_config.model_task_config.tool_use, request_type="tool_executor") + # 二步工具调用状态管理 + self._pending_step_two_tools: Dict[str, Dict[str, Any]] = {} + """待处理的第二步工具调用，格式为 {tool_name: step_two_definition}""" + logger.info(f"{self.log_prefix}工具执行器初始化完成") async def execute_from_chat_message( @@ -112,7 +116,18 @@ class ToolExecutor: def _get_tool_definitions(self) -> List[Dict[str, Any]]: all_tools = get_llm_available_tool_definitions() user_disabled_tools = global_announcement_manager.get_disabled_chat_tools(self.chat_id) - return [definition for name, definition in all_tools if name not in user_disabled_tools] + + # 获取基础工具定义（包括二步工具的第一步） + tool_definitions = [definition for name, definition in all_tools if name not in user_disabled_tools] + + # 检查是否有待处理的二步工具第二步调用 + pending_step_two = getattr(self, '_pending_step_two_tools', {}) + if pending_step_two: + # 添加第二步工具定义 + for tool_name, step_two_def in pending_step_two.items(): + tool_definitions.append(step_two_def) + + return tool_definitions async def execute_tool_calls(self, tool_calls: Optional[List[ToolCall]]) -> Tuple[List[Dict[str, Any]], List[str]]: """执行工具调用 @@ -251,6 +266,32 @@ class ToolExecutor: f"{self.log_prefix} 正在执行工具: [bold green]{function_name}[/bold green] | 参数: {function_args}" ) function_args["llm_called"] = True # 标记为LLM调用 + + # 检查是否是二步工具的第二步调用 + if "_" in function_name and function_name.count("_") >= 1: + # 可能是二步工具的第二步调用，格式为 "tool_name_sub_tool_name" + parts = function_name.split("_", 1) + if len(parts) == 2: + base_tool_name, sub_tool_name = parts + base_tool_instance = get_tool_instance(base_tool_name) + + if base_tool_instance and base_tool_instance.is_two_step_tool: + logger.info(f"{self.log_prefix}执行二步工具第二步: {base_tool_name}.{sub_tool_name}") + result = await base_tool_instance.execute_step_two(sub_tool_name, function_args) + + # 清理待处理的第二步工具 + self._pending_step_two_tools.pop(base_tool_name, None) + + if result: + logger.debug(f"{self.log_prefix}二步工具第二步 {function_name} 执行成功") + return { + "tool_call_id": tool_call.call_id, + "role": "tool", + "name": function_name, + "type": "function", + "content": result.get("content", ""), + } + # 获取对应工具实例 tool_instance = tool_instance or get_tool_instance(function_name) if not tool_instance: @@ -260,6 +301,16 @@ class ToolExecutor: # 执行工具并记录日志 logger.debug(f"{self.log_prefix}执行工具 {function_name}，参数: {function_args}") result = await tool_instance.execute(function_args) + + # 检查是否是二步工具的第一步结果 + if result and result.get("type") == "two_step_tool_step_one": + logger.info(f"{self.log_prefix}二步工具第一步完成: {function_name}") + # 保存第二步工具定义 + next_tool_def = result.get("next_tool_definition") + if next_tool_def: + self._pending_step_two_tools[function_name] = next_tool_def + logger.debug(f"{self.log_prefix}已保存第二步工具定义: {next_tool_def['name']}") + if result: logger.debug(f"{self.log_prefix}工具 {function_name} 执行成功，结果: {result}") return {