构建自己的 AI Agent · 第二篇：工具系统 — 从函数签名到工具链

"parameters": {
    "type": "object",
    "properties": {},  # ← 空！什么参数都不接受
    "required": []
}

def get_weather(city: str, date: str = "today") -> dict:
    """查询指定城市的天气信息。

    Args:
        city: 城市名称，支持中文或英文，如 '北京' 或 'Beijing'
        date: 日期，格式 YYYY-MM-DD，默认为今天
    """
    # ...

import inspect
import json
import functools
from typing import get_type_hints, Literal, get_origin, get_args
from dataclasses import dataclass, field

@dataclass
class ToolInfo:
    """工具的完整元数据"""
    name: str
    description: str
    fn: callable
    schema: dict
    param_types: dict

class ToolRegistry:
    """工具注册表 —— 管理所有可用工具"""
    def __init__(self):
        self._tools: dict[str, ToolInfo] = {}

    def register(self, fn=None, *, name=None, description=None):
        """@tool 装饰器的核心实现"""
        if fn is None:
            return lambda f: self.register(f, name=name,
                                           description=description)

        tool_name = name or fn.__name__
        tool_desc = description or (fn.__doc__ or "").split("\n")[0].strip()

        hints = get_type_hints(fn) if hasattr(fn, '__annotations__') else {}
        param_names = list(inspect.signature(fn).parameters.keys())

        properties = {}
        required = []

        for pname, ptype in hints.items():
            if pname == 'return':
                continue
            prop = self._type_to_schema(ptype)
            # 从 docstring 提取参数描述
            param_desc = self._extract_param_desc(fn.__doc__, pname)
            if param_desc:
                prop['description'] = param_desc
            properties[pname] = prop

            # 没有默认值 = 必填
            sig = inspect.signature(fn)
            if sig.parameters[pname].default is inspect.Parameter.empty:
                required.append(pname)

        schema = {
            "type": "function",
            "function": {
                "name": tool_name,
                "description": tool_desc,
                "parameters": {
                    "type": "object",
                    "properties": properties,
                    "required": required
                }
            }
        }

        self._tools[tool_name] = ToolInfo(
            name=tool_name,
            description=tool_desc,
            fn=fn,
            schema=schema,
            param_types=hints
        )
        return fn

    @staticmethod
    def _type_to_schema(ptype) -> dict:
        """核心：Python 类型 → JSON Schema"""
        origin = get_origin(ptype)
        args = get_args(ptype)

        # 处理 Optional[X] = Union[X, None]
        if origin is Union or (hasattr(ptype, '__origin__') and
                                str(ptype).startswith('typing.Optional')):
            non_none = [a for a in args if a is not type(None)]
            if non_none:
                return ToolRegistry._type_to_schema(non_none[0])

        # Literal["a", "b"] → enum
        if origin is Literal:
            return {"type": "string", "enum": list(args)}

        # list[T] → array
        if origin is list:
            return {"type": "array", "items": ToolRegistry._type_to_schema(args[0])}

        # 基础类型
        type_map = {
            str: "string", int: "integer", float: "number",
            bool: "boolean", dict: "object"
        }
        json_type = type_map.get(origin or ptype)
        if json_type:
            return {"type": json_type}

        return {"type": "string"}  # fallback

    @staticmethod
    def _extract_param_desc(docstring: str, param_name: str) -> str | None:
        """从 Google 风格的 docstring 提取参数描述"""
        if not docstring:
            return None
        for line in docstring.split('\n'):
            line = line.strip()
            if line.startswith(f'{param_name}:'):
                return line.split(':', 1)[1].strip()
        return None

    def get_schemas(self) -> list[dict]:
        """获取所有工具的 OpenAI tool calling schema"""
        return [t.schema for t in self._tools.values()]

    def execute(self, name: str, arguments: dict) -> str:
        """执行工具：校验参数 → 调用函数 → 序列化结果"""
        tool = self._tools.get(name)
        if not tool:
            return f"错误：未找到工具 '{name}'。可用工具：{list(self._tools.keys())}"

        # 参数校验
        try:
            validated = self._validate_params(tool, arguments)
        except ValueError as e:
            return f"参数错误：{e}\n工具签名：{tool.name}{inspect.signature(tool.fn)}"

        # 执行
        try:
            result = tool.fn(**validated)
            return json.dumps(result, ensure_ascii=False, default=str)
        except Exception as e:
            return f"工具执行失败：{type(e).__name__}: {e}\n请检查参数并重试。"

    def _validate_params(self, tool: ToolInfo, arguments: dict) -> dict:
        """参数校验与类型转换"""
        validated = {}
        sig = inspect.signature(tool.fn)

        for pname, ptype in tool.param_types.items():
            if pname == 'return':
                continue
            if pname in arguments:
                value = arguments[pname]
                # 类型转换
                origin = get_origin(ptype)
                if origin is list and not isinstance(value, list):
                    raise ValueError(f"参数 '{pname}' 应为列表类型")
                validated[pname] = value
            elif sig.parameters[pname].default is not inspect.Parameter.empty:
                validated[pname] = sig.parameters[pname].default
            else:
                raise ValueError(f"缺少必填参数 '{pname}'")

        return validated


# ── 使用示例 ──
registry = ToolRegistry()
tool = registry.register  # 别名

@tool
def get_weather(city: str, date: str = "today") -> dict:
    """查询指定城市的天气信息。

    Args:
        city: 城市名称，如 '北京' 或 'Shanghai'
        date: 日期，格式 YYYY-MM-DD，默认为今天
    """
    # 实际项目中调用天气 API
    return {"city": city, "date": date, "temp": 22, "weather": "晴"}

@tool
def search_code(query: str, file_pattern: str = "*.py") -> list[str]:
    """在代码库中搜索包含特定关键词的文件。

    Args:
        query: 搜索关键词
        file_pattern: 文件名匹配模式
    """
    return [f"src/main.py:42: {query} results"]

# 生成的 Schema：
print(json.dumps(registry.get_schemas(), indent=2, ensure_ascii=False))

class SimpleAgent:
    def __init__(self, system_prompt: str, registry: ToolRegistry,
                 max_iter: int = 10):
        self.client = OpenAI()
        self.system_prompt = system_prompt
        self.registry = registry     # ← 使用 ToolRegistry
        self.max_iter = max_iter

    def run(self, user_query: str) -> str:
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": user_query}
        ]
        schemas = self.registry.get_schemas()  # ← 自动生成的 schema

        for _ in range(self.max_iter):
            response = self.client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                tools=schemas if schemas else None
            )
            msg = response.choices[0].message

            if msg.tool_calls:
                for tc in msg.tool_calls:
                    fn_name = tc.function.name
                    fn_args = json.loads(tc.function.arguments)
                    # ← 使用 registry 执行（含校验+错误处理）
                    result = self.registry.execute(fn_name, fn_args)

                    messages.append({
                        "role": "assistant",
                        "content": None,
                        "tool_calls": [tc]
                    })
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": result
                    })
            else:
                return msg.content or ""

        return "达到最大迭代次数。"