• 模板方法模式实战:重构Agent工具审批,告别重复代码


    设计模式-模板方法实战

    什么是模板方法

    在软件设计模式中,模板方法设计模式是一种行为型设计模式,其核心思想是在抽象类中定义一个算法的骨架(即模板),并将某些具体的实现步骤延迟到子类去完成

    实战概览

    在自己编写的agent框架中实现工具调用需要审批

    为什么做这篇

    在学习设计模式时,应当结合实战。不久前刚学到的模板方法设计模式,今天在搓自己的agent框架实现功能——工具调用需要审批时刚好用得上

    代码实战

    未使用模板方法的缺陷

    要实现的功能:工具调用需要审批

    tool_registry.py中,调用的逻辑伪代码如下

    # 触发审批功能
    approved = self.approval_tool.approve(tool, parameters, tool_call_id)
    # 根据审批结果,调用对应的功能
    if not approved:
    	return '未通过审批!'
    #开始调用工具
    tool.run(parameters, tool_call_id)
    # ...省略
    

    根据要实现的功能,不难得出在调用Tool前需要触发审批的逻辑,由于审批的实现有多种,比如可以实现以下审批:

    1. 交互式命令行审批
    2. Web界面审批
    3. AI辅助审批

    鉴于有多种实现,我们需要先创建一个抽象类,让子类来实现对应的审批逻辑,上述代码块approval_tool是对应的实现类对象

    class ApprovalTool(ABC):
        """审批工具基类"""
        
        @abstractmethod
        def approve(self, tool: Tool, parameters: Dict[str, Any], tool_call_id: str) -> bool:
            """具体的审批逻辑,由子类实现
            
            Args:
                tool (Tool): 被调用的工具实例
                parameters (Dict[str, Any]): 工具参数
                tool_call_id (str): 工具调用ID
            """
            pass
    

    然后让子类去实现该抽象方法,例子如下(交互式命令行审批):

    class DefaultApprovalTool(ApprovalTool):
        """默认审批工具"""
        def approve(self, tool: Tool, parameters: Dict[str, Any], tool_call_id: str) -> bool:
            max_attempts = self.max_attempts
            while max_attempts > 0:
                is_approved = input(f"{prompt}\n是否批准该工具调用?(y/n): ")
                if is_approved.lower() == "y":
                    return True
                elif is_approved.lower() == "n":
                    return False
                else:
                    print(f"无效输入,请输入 'y' 或 'n'\n 剩余尝试次数: {max_attempts - 1}")
                    max_attempts -= 1
            print("超过最大尝试次数,默认拒绝该工具调用")
            return False
    

    至此,一个审批工具实现类便以完成,伪代码使用示例如下(approval_tool为对应的实现类对象):

    # 触发审批功能
    approved = self.approval_tool.approve(tool, parameters, tool_call_id)
    # 根据审批结果,调用对应的功能
    if not approved:
    	return '未通过审批!'
    #开始调用工具
    tool.run(parameters, tool_call_id)
    # ...省略
    

    缺陷展现

    在完成该功能后,假设想为工具审批添加自动批准指定工具、需要审批的指定工具功能,那么这些逻辑都需要在实现类中编写,这些功能在代码编写上都是相似的,每个实现类都要为实现这些功能编写重复性的代码,这并不利于日后的维护和保持代码简洁优雅

    模板方法设计模式可以有效解决上述缺陷

    使用模板方法

    再次回顾:在软件设计模式中,模板方法设计模式是一种行为型设计模式,其核心思想是在抽象类中定义一个算法的骨架(即模板),并将某些具体的实现步骤延迟到子类去完成

    为了解决上述缺陷,我们需要使用模板方法设计模式,现在我们修改ApprovalTool类,将抽象方法改为do_approve()

    class ApprovalTool(ABC):
        """审批工具基类"""
        
        def __init__(self,
                     auto_approve_tools: Optional[List[Type[Tool]]] = None,
                     require_approval_tools: Optional[List[Type[Tool]]] = None,
                     auto_approve_if_no_rules: bool = False
                     ):
            """初始化审批工具
            假设一个tool在auto_approve_tools和require_approval_tools中都存在,则以require_approval_tools为准,即需要审批。
            假设一个tool既不在auto_approve_tools中,也不在require_approval_tools中,则根据auto_approve_if_no_rules的值来决定是否需要审批。
    
            
            Args:
                auto_approve_tools (Optional[List[Type[Tool]]], optional): 自动批准的工具列表. Defaults to None.
                require_approval_tools (Optional[List[Type[Tool]]], optional): 需要审批的工具列表. Defaults to None.
                auto_approve_if_no_rules (bool, optional): 当工具不在任何列表中时是否自动批准. Defaults to False.
            """
            self.auto_approve_tools = auto_approve_tools or []
            self.require_approval_tools = require_approval_tools or []
            self.auto_approve_if_no_rules = auto_approve_if_no_rules
            
        def approve(self, tool: Tool, parameters: Dict[str, Any], tool_call_id: str) -> bool:
            """
            根据工具类型和审批规则决定是否批准工具调用
            Args:
                tool (Tool): 被调用的工具实例
                parameters (Dict[str, Any]): 工具参数
                tool_call_id (str): 工具调用ID
            Returns:
                bool: 是否批准工具调用
            """
            if type(tool) in self.require_approval_tools:
                return self.do_approve(tool, parameters, tool_call_id)
            if type(tool) in self.auto_approve_tools:
                return True
            if self.auto_approve_if_no_rules == False:
                return self.do_approve(tool, parameters, tool_call_id)
            else:
                return True
    
        @abstractmethod
        def do_approve(self, tool: Tool, parameters: Dict[str, Any], tool_call_id: str) -> bool:
            """具体的审批逻辑,由子类实现
            
            Args:
                tool (Tool): 被调用的工具实例
                parameters (Dict[str, Any]): 工具参数
                tool_call_id (str): 工具调用ID
            """
            pass
    

    可以看到抽象方法改成了do_approve()自动批准指定工具、需要审批的指定工具的功能已在approve()实现,使得每个实现类都不用重复性地编写自动批准指定工具、需要审批的指定工具相关代码。我们再来看下子类(交互式命令行审批)代码:

    class DefaultApprovalTool(ApprovalTool):
        """默认审批工具"""
    
        def __init__(self,
                     auto_approve_tools: Optional[List[Type[Tool]]] = None,
                     require_approval_tools: Optional[List[Type[Tool]]] = None,
                     auto_approve_if_no_rules: bool = False,                  
                     max_attempts: int = 10):
            """初始化审批工具
            假设一个tool在auto_approve_tools和require_approval_tools中都存在,则以require_approval_tools为准,即需要审批。
            假设一个tool既不在auto_approve_tools中,也不在require_approval_tools中,则根据auto_approve_if_no_rules的值来决定是否需要审批。
    
            
            Args:
                auto_approve_tools (Optional[List[Type[Tool]]], optional): 自动批准的工具列表. Defaults to None.
                require_approval_tools (Optional[List[Type[Tool]]], optional): 需要审批的工具列表. Defaults to None.
                auto_approve_if_no_rules (bool, optional): 当工具不在任何列表中时是否自动批准. Defaults to False.
                max_attempts (int, optional): 最大尝试次数. Defaults to 10.
            """
            super().__init__(auto_approve_tools, require_approval_tools, auto_approve_if_no_rules)
            self.max_attempts = max_attempts
    
        def do_approve(self, tool: Tool, parameters: Dict[str, Any], tool_call_id: str) -> bool:
            if isinstance(tool, TerminalTool):
                current_dir = tool.current_dir
                prompt = f"{current_dir}> {parameters.get('command', '')}"
            else:
                prompt = f"工具名称: {tool.name}\n工具描述: {tool.description}\n工具参数: {parameters}"
            max_attempts = self.max_attempts
            while max_attempts > 0:
                is_approved = input(f"{prompt}\n是否批准该工具调用?(y/n): ")
                if is_approved.lower() == "y":
                    return True
                elif is_approved.lower() == "n":
                    return False
                else:
                    print(f"无效输入,请输入 'y' 或 'n'\n 剩余尝试次数: {max_attempts - 1}")
                    max_attempts -= 1
            print("超过最大尝试次数,默认拒绝该工具调用")
            return False
    

    可以看到子类只关心实现do_approve(),而自动批准指定工具、需要审批的指定工具不需要子类关心,这种重复的代码已在ApprovalTool类中的approve()方法编写。我们再来看下实际调用

    class ToolRegistry:
        """
        工具注册表,用于管理和调用工具
        """
        def __init__(self, approval_tool: Optional[ApprovalTool] = None):
            self._tools: Dict[str, Tool] = {}
            self.approval_tool = approval_tool
    
        def execute_tool(self, tool_call: Union[Dict[str, Any], ChatCompletionMessageFunctionToolCall]) -> Message:
            """
            执行工具
    
            Args:
                tool_call (Union[Dict[str, Any], ChatCompletionMessageFunctionToolCall]): 工具调用信息,llm返回的工具调用格式,包含工具名称、参数等信息
    
            Returns:
                Message: 工具执行结果封装的Message对象
            """
            
            # ...省略代码
            tool = self._tools[tool_name]
            if self.approval_tool:
                approved = self.approval_tool.approve(tool, parameters, tool_call_id)
                if not approved:
                    return Message(role="tool", content=f"❌ 工具调用未通过用户的审批,已被用户拒绝: {tool_name}", tool_call_id=tool_call_id)
    
            return tool.run(parameters, tool_call_id)
    

    参考文章

    模板模式 | 菜鸟教程

  • 相关阅读:
    2022.8.7-----leetcode.636
    2022年6月电子学会Python等级考试试卷(三级)答案解析
    GBase 8c V3.0.0数据类型——注释信息函数
    (五)CSS前端开发面试会问到的问题有哪些?
    深度学习论文阅读目标检测篇(七)中文版:YOLOv4《Optimal Speed and Accuracy of Object Detection》
    苹果“慌了”,中国客户不买账,这次要提供“折扣”可谓罕见
    案例!快看!电力行业利用IBPS低代码应用开发平台做好数据治理工作
    【题目讲解】等腰三角形
    鸿蒙应用开发-初见:ArkUI
    第五十九章 IIS 7 或更高版本的替代选项 (Windows) - 替代选项 2:将本机模块与 - 映射 InterSystems IRIS 文件扩展名
  • 原文地址:https://www.cnblogs.com/Yoshizawa-violet/p/20197311