二步工具与action

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]:

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]:
         """直接执行工具函数(供插件调用)
            通过该方法，插件可以直接调用工具，而不需要传入字典格式的参数

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
 
 

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 {