工具注册表与参数 Schema

智能体的「手」就是工具:模型说「查天气」,程序就得真去查。但模型不认识 Python 函数,只认 JSON——它需要一份「工具说明书」(参数 Schema)才知道能调什么、参数怎么写。本章做一个工具注册表:用 @tool 装饰器把普通函数登记成工具,自动生成 Schema 列表 TOOL_SCHEMAS,再用 call_tool(name, args) 统一分发、把任何返回值都变成字符串。这份 tools.py 是第 6~8 章的公用地基。

1. 工具要解决的三件事

  1. 描述:告诉模型「有什么工具、干什么用」(name + description);
  2. 参数 Schema:告诉模型「参数叫什么、什么类型、是否必填」,模型据此生成 JSON 参数;
  3. 分发:程序拿到工具名和参数,安全地调用真正的函数并返回结果。

2. 完整代码 tools.py

"""tools.py —— 工具注册表:@tool 登记函数,自动生成 Schema,call_tool 统一分发。"""
import inspect
import json
from datetime import datetime
from typing import Any, Callable, Dict, List, Optional

TOOL_REGISTRY: Dict[str, Callable] = {}    # 工具名 -> 函数
TOOL_SCHEMAS: List[Dict[str, Any]] = []    # OpenAI function calling 格式的工具列表

# Python 基础类型 -> JSON Schema 类型
_PY_TO_JSON: Dict[Any, str] = {
    int: "integer", float: "number", str: "string", bool: "boolean",
}


def _build_schema(func: Callable, name: str, description: str,
                  param_docs: Optional[Dict[str, str]]) -> Dict[str, Any]:
    """用 inspect 读函数签名,自动生成参数 Schema(描述缺省用参数名)。"""
    sig = inspect.signature(func)
    properties: Dict[str, Any] = {}
    required: List[str] = []
    for pname, param in sig.parameters.items():
        if pname in ("self", "cls"):
            continue
        jtype = _PY_TO_JSON.get(param.annotation, "string")  # 未注解一律按 string
        properties[pname] = {
            "type": jtype,
            "description": (param_docs or {}).get(pname, f"参数 {pname}"),
        }
        if param.default is inspect.Parameter.empty:   # 没有默认值 => 必填
            required.append(pname)
    return {
        "type": "function",
        "function": {
            "name": name,
            "description": description,
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required,
            },
        },
    }


def tool(name: Optional[str] = None,
         description: Optional[str] = None,
         params: Optional[Dict[str, str]] = None) -> Callable:
    """装饰器:登记一个工具,返回原函数(函数本身不受影响)。"""
    def decorator(func: Callable) -> Callable:
        tool_name = name or func.__name__
        desc = description                      # 注意:不能直接给闭包外的
        if desc is None:                        # description 赋值,否则会遮蔽外层参数
            doc = inspect.getdoc(func) or ""
            desc = doc.splitlines()[0] if doc else tool_name
        TOOL_REGISTRY[tool_name] = func
        TOOL_SCHEMAS.append(_build_schema(func, tool_name, desc, params))
        return func
    return decorator


@tool(description="计算两个整数的和",
      params={"a": "第一个整数", "b": "第二个整数"})
def add(a: int, b: int) -> int:
    return a + b


@tool(description="获取当前本地时间(年-月-日 时:分:秒)")
def get_current_time() -> str:
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


_CITY_WEATHER = {"北京": "晴,24℃", "上海": "多云,27℃", "深圳": "阵雨,29℃"}


@tool(description="查询指定城市的天气(演示用模拟数据,第 9 章换成真实接口)",
      params={"city": "城市名,如 北京"})
def get_weather(city: str) -> str:
    return _CITY_WEATHER.get(city, f"暂无 {city} 的天气数据")


def call_tool(name: str, args: Dict[str, Any]) -> str:
    """统一分发入口:按名字找到函数并调用,任何结果都转成字符串返回。

    约定:以"错误:"开头的字符串表示调用失败,模型看到后可自行纠正。
    """
    func = TOOL_REGISTRY.get(name)
    if func is None:
        return f"错误:未知工具 {name},可用工具:{', '.join(TOOL_REGISTRY)}"
    try:
        result = func(**args)
    except TypeError as e:
        return f"错误:调用 {name} 时参数不正确({e}),请核对参数名与类型"
    except Exception as e:      # 工具内部异常也必须吞掉,把错误交还给模型
        return f"错误:工具 {name} 执行失败:{type(e).__name__}: {e}"
    if isinstance(result, str):
        return result
    return json.dumps(result, ensure_ascii=False)   # dict/list 序列化成字符串


if __name__ == "__main__":
    # 自测:打印注册表内容,再试几次分发
    print("已注册工具:")
    for schema in TOOL_SCHEMAS:
        fn = schema["function"]
        print(f"- {fn['name']}: {fn['description']}  参数={list(fn['parameters']['properties'])}")
    print()
    print(call_tool("add", {"a": 12, "b": 30}))          # 输出:42
    print(call_tool("get_current_time", {}))             # 输出:2026-01-01 12:00:00(随当前时间变化)
    print(call_tool("get_weather", {"city": "北京"}))    # 输出:晴,24℃
    print(call_tool("get_weather", {"city": "东京"}))    # 输出:暂无 东京 的天气数据
    print(call_tool("fly", {}))                          # 输出:错误:未知工具 fly ...
    print(call_tool("add", {"a": "x"}))                  # 输出:错误:调用 add 时参数不正确...

3. 运行与观察

python tools.py

重点看两个生成物:TOOL_REGISTRY(程序内部用)和 TOOL_SCHEMAS(发给模型用)。TOOL_SCHEMAS 里每个工具的 parameters 结构,正是 OpenAI 函数调用规范要求的格式——第 6 章会把它原样塞进请求。

4. 设计要点

  • 自动生成 Schemainspect.signature 拿到参数名、类型注解、默认值,自动填 type 与 required;参数说明写在装饰器 params={...} 里,比从 docstring 解析更直观。
  • 类型注解要写基础类型:int/float/str/bool 会被正确映射,没注解或注解成 Optional[int] 等复合类型的一律按 string 处理——给模型用的 Schema 保持简单最重要。
  • 返回值统一成 str:模型只能把 Observation 当文本读,所以 call_tool 把 dict/list 用 json.dumps 序列化、异常转成「错误:」开头的字符串。以「错误:」开头是约定信号,第 6~8 章的循环都据此让模型自行修正。
  • 装饰器不影响原函数:被 @tool 包裹后,add(1, 2) 仍可直接调用,方便自己写单元测试。 小结:@tool 装饰器 + inspect 自动生成 Schema,让「加一个新工具」只需写一个普通函数加一行装饰器;call_tool 统一分发并保证返回字符串。第 6 章把 TOOL_SCHEMAS 交给模型,智能体就「长出手」了。
笔记加载中…