backend/backend_app/helloAgents/tools/base.py

"""工具基类"""

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)