"""工具基类""" from abc import ABC, abstractmethod from typing import Dict, Any, List, Optional, Callable, get_type_hints from pydantic import BaseModel import inspect import re def tool_action(name: str = None, description: str = None): """装饰器：标记一个方法为可展开的工具 action 用法: @tool_action("memory_add", "添加新记忆") def _add_memory(self, content: str, importance: float = 0.5) -> str: '''添加记忆 Args: content: 记忆内容 importance: 重要性分数 ''' ... Args: name: 工具名称（如果不提供，从方法名自动生成） description: 工具描述（如果不提供，从 docstring 提取） """ def decorator(func: Callable): func._is_tool_action = True func._tool_name = name func._tool_description = description return func return decorator class ToolParameter(BaseModel): """工具参数定义""" name: str type: str description: str required: bool = True default: Any = None class Tool(ABC): """工具基类 支持两种使用模式： 1. 普通模式：工具作为单一实体使用 2. 可展开模式：工具可以展开为多个独立的子工具（每个子工具对应一个功能） 展开模式支持两种实现方式： - 手动定义子工具类（传统方式） - 使用 @tool_action 装饰器自动生成（推荐） """ def __init__(self, name: str, description: str, expandable: bool = False): """初始化工具 Args: name: 工具名称 description: 工具描述 expandable: 是否可展开为多个子工具 """ self.name = name self.description = description self.expandable = expandable @abstractmethod def run(self, parameters: Dict[str, Any]) -> str: """执行工具""" pass @abstractmethod def get_parameters(self) -> List[ToolParameter]: """获取工具参数定义""" pass def get_expanded_tools(self) -> Optional[List['Tool']]: """获取展开后的子工具列表 默认实现：自动从标记了 @tool_action 的方法生成子工具 子类可以重写此方法提供自定义的展开逻辑 Returns: 如果工具支持展开，返回子工具列表；否则返回 None """ if not self.expandable: return None # 自动从装饰器标记的方法生成工具 tools = [] for name, method in inspect.getmembers(self, predicate=inspect.ismethod): if hasattr(method, '_is_tool_action'): tool = AutoGeneratedTool( parent=self, method=method, name=method._tool_name, description=method._tool_description ) tools.append(tool) return tools if tools else None def validate_parameters(self, parameters: Dict[str, Any]) -> bool: """验证参数""" required_params = [p.name for p in self.get_parameters() if p.required] return all(param in parameters for param in required_params) def to_dict(self) -> Dict[str, Any]: """转换为字典格式""" return { "name": self.name, "description": self.description, "parameters": [param.dict() for param in self.get_parameters()] } def to_openai_schema(self) -> Dict[str, Any]: """转换为 OpenAI function calling schema 格式 用于 FunctionCallAgent，使工具能够被 OpenAI 原生 function calling 使用 Returns: 符合 OpenAI function calling 标准的 schema """ parameters = self.get_parameters() # 构建 properties properties = {} required = [] for param in parameters: # 基础属性定义 prop = { "type": param.type, "description": param.description } # 如果有默认值，添加到描述中（OpenAI schema 不支持 default 字段） if param.default is not None: prop["description"] = f"{param.description} (默认: {param.default})" # 如果是数组类型，添加 items 定义 if param.type == "array": prop["items"] = {"type": "string"} # 默认字符串数组 properties[param.name] = prop # 收集必需参数 if param.required: required.append(param.name) return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": { "type": "object", "properties": properties, "required": required } } } def __str__(self) -> str: return f"Tool(name={self.name})" def __repr__(self) -> str: return self.__str__() class AutoGeneratedTool(Tool): """自动生成的工具 - 从方法签名和 docstring 自动提取参数""" def __init__(self, parent: Tool, method: Callable, name: str = None, description: str = None): """初始化自动生成的工具 Args: parent: 父工具实例 method: 被装饰的方法 name: 工具名称（如果为 None，从方法名生成） description: 工具描述（如果为 None，从 docstring 提取） """ self.parent = parent self.method = method # 生成工具名称 if name is None: # 从方法名生成：_add_memory -> parent_name_add_memory method_name = method.__name__.lstrip('_') name = f"{parent.name}_{method_name}" # 提取描述 if description is None: description = self._extract_description_from_docstring() super().__init__(name=name, description=description) # 自动解析参数 self._parameters = self._parse_parameters() def _extract_description_from_docstring(self) -> str: """从 docstring 提取描述""" doc = inspect.getdoc(self.method) if not doc: return f"执行 {self.method.__name__}" # 提取第一行作为描述 lines = doc.split('\n') for line in lines: line = line.strip() if line and not line.startswith('Args:') and not line.startswith('Returns:'): return line return f"执行 {self.method.__name__}" def _parse_parameters(self) -> List[ToolParameter]: """从方法签名和 docstring 自动提取参数""" sig = inspect.signature(self.method) type_hints = get_type_hints(self.method) docstring = inspect.getdoc(self.method) or "" # 从 docstring 解析参数描述 param_descriptions = self._parse_param_descriptions(docstring) parameters = [] for param_name, param in sig.parameters.items(): if param_name == 'self': continue # 获取类型 param_type_hint = type_hints.get(param_name, str) param_type = self._python_type_to_tool_type(param_type_hint) # 判断是否必需 required = param.default == inspect.Parameter.empty default = None if required else param.default # 获取描述 description = param_descriptions.get(param_name, f"参数 {param_name}") parameters.append(ToolParameter( name=param_name, type=param_type, description=description, required=required, default=default )) return parameters def _parse_param_descriptions(self, docstring: str) -> Dict[str, str]: """从 docstring 解析参数描述 支持格式: Args: param_name: 参数描述 another_param: 另一个参数描述 """ descriptions = {} # 查找 Args: 部分 args_match = re.search(r'Args:\s*\n(.*?)(?:\n\s*\n|Returns:|$)', docstring, re.DOTALL) if not args_match: return descriptions args_section = args_match.group(1) # 解析每个参数 # 匹配格式: param_name: 描述 或 param_name (type): 描述 param_pattern = r'^\s*(\w+)(?:\s*\([^)]+\))?\s*:\s*(.+?)(?=^\s*\w+\s*(?:\([^)]+\))?\s*:|$)' matches = re.finditer(param_pattern, args_section, re.MULTILINE | re.DOTALL) for match in matches: param_name = match.group(1).strip() param_desc = match.group(2).strip() # 清理描述中的多余空白 param_desc = re.sub(r'\s+', ' ', param_desc) descriptions[param_name] = param_desc return descriptions def _python_type_to_tool_type(self, py_type) -> str: """将 Python 类型转换为工具类型字符串""" # 处理泛型类型 origin = getattr(py_type, '__origin__', None) if origin is not None: if origin is list: return "array" elif origin is dict: return "object" # 处理基本类型 type_map = { str: "string", int: "integer", float: "number", bool: "boolean", list: "array", dict: "object", } return type_map.get(py_type, "string") def get_parameters(self) -> List[ToolParameter]: """获取参数列表""" return self._parameters def run(self, parameters: Dict[str, Any]) -> str: """执行方法""" return self.method(**parameters)