工具系统：可插拔的工具注册机制

这篇笔记解决什么问题

一个能干活的 Agent，本质上就是"模型 + 一堆工具"。工具少的时候怎么写都行，但只要项目活得够久，工具一定会越来越多：查数据库、发请求、读写文件、操作浏览器……这时候最先崩溃的往往不是模型，而是你组织工具代码的方式。

最朴素的写法是把所有工具塞进一个 tools.py：函数实现、名字到函数的映射表、给模型看的 JSON Schema 描述，全在一个文件里。每加一个工具，你要在三个地方同步修改：

1. 写一个新的工具函数；
2. 把函数塞进映射字典；
3. 在描述列表里补一份 JSON Schema。

三处改动只要漏掉一处，就会出现"模型看得见工具但调不动"或"函数存在但模型不知道"这类诡异 bug。多人协作时更糟——所有人都在改同一个文件，冲突和遗漏是家常便饭。工具攒到十几个，这个文件轻松膨胀到几百上千行。

这篇笔记讲一种业界通用的解法：可插拔的工具注册机制。目标是把"加一个工具"的成本降到"往目录里丢一个文件"，主流程代码一行不用动。

核心思路：约定优于配置

自动发现工具，需要回答三个问题：扫描器怎么判断一个文件是工具？工具的元信息（名字、描述、参数）从哪读？拿到之后怎么执行？

答案不是写一套复杂的配置系统，而是定一组极简约定，所有工具文件都遵守它：

• 一个工具一个文件，统一放在专门的目录里（本文用 toolbox/）；
• 文件名带固定后缀（本文约定 _tool.py），扫描器只认这个后缀，目录里的辅助模块不会被误当成工具；
• 每个工具文件必须导出两个东西：

- DEFINITION：一个 dict，描述工具叫什么、干什么、参数长什么样（JSON Schema 格式）； - execute：一个函数，接收参数字典，返回字符串结果。

满足约定的文件，注册器扫到就能用；不满足的，跳过并打警告。这就是"约定优于配置"（Convention over Configuration）——用一条大家都遵守的规则，换掉一堆需要手工维护的注册代码。

DEFINITION 里通常同时放 id 和 name 两个字段。id 是系统内部的唯一索引，用来做配置、开关、日志；name 是暴露给模型的函数名。绝大多数时候两者相同，但拆开留了口子——比如以后想给同一个实现挂两个不同的对外名字，或者做灰度改名，都不用动索引。

工具文件的两件套：DEFINITION 与 execute

先看一个同步工具的完整例子。场景是一个运维小助手，第一个工具查磁盘用量：

# agent_app/toolbox/disk_usage_tool.py
import shutil

DEFINITION = {
    "id": "disk_usage",
    "name": "disk_usage",
    "description": (
        "查询指定磁盘分区的容量使用情况，返回总量、已用、剩余（GB）。"
        "适合回答“磁盘还剩多少空间”这类问题；不适合查询单个文件的大小。"
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "mount_point": {
                "type": "string",
                "description": "挂载点路径，例如 / 或 /data，缺省为 /",
            },
        },
    },
}


def execute(args: dict, **kwargs) -> str:
    mount = args.get("mount_point", "/")
    try:
        usage = shutil.disk_usage(mount)
    except FileNotFoundError:
        return f"Error: 挂载点不存在：{mount}"

    gb = 1024 ** 3
    return (
        f"{mount} 总容量 {usage.total / gb:.1f} GB，"
        f"已用 {usage.used / gb:.1f} GB，剩余 {usage.free / gb:.1f} GB"
    )

这个签名有两处值得注意的设计：

统一签名 `execute(args: dict, kwargs)。** args 是模型生成的参数字典，整个 dict 传进来，而不是展开成具名参数（args）。这样所有工具的调用方式完全一致，注册器不需要关心每个工具的参数长什么样。kwargs` 是留给主流程的扩展位：以后想把会话 ID、当前用户、工作目录之类的上下文传给工具，直接塞 kwargs 即可，用不到的工具忽略它，老工具一行不用改。

失败返回 Error: 开头的字符串，而不是抛异常。 工具跑在 Agent 循环里，异常一路抛上去会把整个循环掀翻；而一条 "Error: 挂载点不存在：/mnt/backup" 会作为工具结果回传给模型，模型看到错误信息后可以自己纠正参数重试，或者换一条路。把错误变成模型能读的数据，Agent 才有自愈能力。

目录最终长这样：

agent_app/
├── cli.py                  # 命令行入口
├── loop.py                 # Agent 主循环
└── toolbox/
    ├── __init__.py         # 注册器
    ├── disk_usage_tool.py
    ├── now_tool.py
    └── ping_url_tool.py    # 加新工具 = 丢一个新文件进来

注册器：扫描目录、动态导入、软失败

注册器写在 toolbox/__init__.py 里，核心就三步：glob 找文件、importlib 动态导入、校验约定后入表。先看不含异步支持的基础版：

# agent_app/toolbox/__init__.py（基础版）
import importlib
from pathlib import Path

_PKG_DIR = Path(__file__).parent
_TOOLBOX: dict[str, dict] | None = None  # 懒加载缓存：id -> {"spec", "runner"}


def _scan() -> dict[str, dict]:
    global _TOOLBOX
    if _TOOLBOX is not None:      # 只扫一次，之后走缓存
        return _TOOLBOX

    registry: dict[str, dict] = {}
    for entry in sorted(_PKG_DIR.glob("*_tool.py")):
        dotted = f"{__name__}.{entry.stem}"
        try:
            mod = importlib.import_module(dotted)
        except Exception as exc:
            print(f"[toolbox] 跳过 {entry.name}：导入失败 {exc}")
            continue

        spec = getattr(mod, "DEFINITION", None)
        runner = getattr(mod, "execute", None)
        if not isinstance(spec, dict) or "id" not in spec or not callable(runner):
            print(f"[toolbox] 跳过 {entry.name}：不符合 DEFINITION/execute 约定")
            continue

        registry[spec["id"]] = {"spec": spec, "runner": runner}

    _TOOLBOX = registry
    return registry


def tool_specs() -> list[dict]:
    """所有工具的 DEFINITION，供主循环拼装 tools 参数。"""
    return [item["spec"] for item in _scan().values()]


def run_tool(tool_id: str, args: dict, **kwargs) -> str:
    """统一执行入口：找不到工具、工具抛异常，一律返回 Error: 字符串。"""
    item = _scan().get(tool_id)
    if item is None:
        return f"Error: 没有名为 '{tool_id}' 的工具"
    try:
        return item["runner"](args, **kwargs)
    except Exception as exc:
        return f"Error: 工具 '{tool_id}' 执行出错：{exc}"

几个关键设计：

• 软失败（soft-fail）：某个工具文件有语法错误、缺依赖、少导出了 DEFINITION——都只是打一条警告然后 continue，其余工具照常加载。一颗老鼠屎不能坏一锅粥，这在多人往同一目录提交工具时尤其重要。
• 懒加载 + 缓存：_TOOLBOX 首次访问时才扫描，之后直接复用。进程启动不付扫描成本，Agent 第一次要用工具时才触发。
• 双保险的异常兜底：约定要求工具自己把错误转成 Error: 字符串，但总有人忘。run_tool 里的 try/except 是最后一道闸，任何漏网异常都会被拦下来转成字符串，保证 Agent 循环永远不会因为一个工具挂掉。

接入 Agent 主循环

注册器就绪后，主循环里不再出现任何具体工具的名字，只跟两个函数打交道：tool_specs() 拿描述，run_tool() 执行。

有一个小的格式转换要做：DEFINITION 存的是扁平的 name / description / parameters，而 OpenAI 兼容接口（DeepSeek、Kimi、GLM 等都遵循这套协议）要求外面再包一层 {"type": "function", "function": {...}}。转换逻辑放在主循环这边：

# agent_app/loop.py（节选）
import json
import os

from openai import OpenAI

from agent_app.toolbox import run_tool, tool_specs

client = OpenAI(
    base_url=os.environ.get("LLM_BASE_URL", "https://api.deepseek.com/v1"),
    api_key=os.environ["LLM_API_KEY"],
)
MODEL = os.environ.get("LLM_MODEL", "deepseek-chat")


def _as_openai_tools() -> list[dict]:
    return [
        {
            "type": "function",
            "function": {
                "name": s["name"],
                "description": s["description"],
                "parameters": s["parameters"],
            },
        }
        for s in tool_specs()
    ]


def run_turn(messages: list[dict]) -> str:
    while True:
        reply = client.chat.completions.create(
            model=MODEL, messages=messages, tools=_as_openai_tools(),
        )
        msg = reply.choices[0].message
        if not msg.tool_calls:               # 没有工具调用，直接回答
            return msg.content or ""

        messages.append(msg)
        for call in msg.tool_calls:
            params = json.loads(call.function.arguments or "{}")
            outcome = run_tool(call.function.name, params)
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": outcome,
            })

注意 run_tool(call.function.name, params) 这行：注册表按 id 索引，模型回传的是 name，所以两者保持一致最省事。真要分开，再建一张 name → id 的映射表即可。

到这里，验证"可插拔"只需要一个动作：新建一个 now_tool.py（返回当前日期时间的两行工具），什么都不用注册，重启后问一句"现在几点"，模型就能调到它。主循环、注册器，零改动。

同步与异步工具的统一执行

工具一多，迟早会遇到天然异步的实现：用 httpx.AsyncClient 发请求、用 Playwright 控制浏览器，接口本身就是 async def。注册器需要同时兼容两种写法，且工具作者不需要做任何额外声明。

改造分三步：

1. 注册时识别：用 inspect.iscoroutinefunction(runner) 判断 execute 是普通函数还是协程函数，把结果作为 is_async 存进注册表；
2. 执行时分发：同步工具直接调用；异步工具先调用得到 coroutine，再丢给后台事件循环跑完取结果；
3. 后台事件循环长期存活：不要每次 asyncio.run()。一是主程序将来跑在 Web 框架里时，当前线程往往已有事件循环，asyncio.run() 会直接报错；二是有些库（典型如 Playwright）把连接、进程状态绑在事件循环上，循环一销毁状态全丢。用一个 daemon 线程养一个 run_forever 的循环，所有异步工具都扔进去跑，状态跨调用保留。

补上这部分后的注册器增量代码：

# agent_app/toolbox/__init__.py（异步支持部分）
import asyncio
import inspect
import threading

_loop_holder: dict = {"loop": None}
_loop_guard = threading.Lock()


def _ensure_loop() -> asyncio.AbstractEventLoop:
    """惰性创建一个跑在 daemon 线程里的长期事件循环。"""
    with _loop_guard:
        loop = _loop_holder["loop"]
        if loop is None or loop.is_closed():
            loop = asyncio.new_event_loop()
            threading.Thread(target=loop.run_forever, daemon=True).start()
            _loop_holder["loop"] = loop
        return loop


def _await_in_background(coro) -> str:
    """把协程提交到后台循环，阻塞等待结果。"""
    task = asyncio.run_coroutine_threadsafe(coro, _ensure_loop())
    return task.result()

注册和执行处各改一行：

# _scan() 入表时多存一个标记
registry[spec["id"]] = {
    "spec": spec,
    "runner": runner,
    "is_async": inspect.iscoroutinefunction(runner),
}

# run_tool() 里按标记分发
if item["is_async"]:
    return _await_in_background(item["runner"](args, **kwargs))
return item["runner"](args, **kwargs)

然后写一个异步工具验证——探测网址连通性：

# agent_app/toolbox/ping_url_tool.py
import time

import httpx

DEFINITION = {
    "id": "ping_url",
    "name": "ping_url",
    "description": (
        "探测一个网址是否可访问，返回 HTTP 状态码和响应耗时（毫秒）。"
        "只做连通性检查，不抓取页面正文。"
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "url": {"type": "string", "description": "完整的 http(s) 地址"},
        },
        "required": ["url"],
    },
}


async def execute(args: dict, **kwargs) -> str:
    url = args.get("url", "")
    if not url.startswith(("http://", "https://")):
        return "Error: url 必须以 http:// 或 https:// 开头"

    started = time.perf_counter()
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.head(url, follow_redirects=True)
    except Exception as exc:
        return f"Error: 无法访问 {url}：{exc}"

    cost_ms = (time.perf_counter() - started) * 1000
    return f"{url} -> HTTP {resp.status_code}，耗时 {cost_ms:.0f} ms"

工具作者只是把 def 写成了 async def，其余什么都没做——识别、调度全由注册器兜住。这就是统一执行的价值：约定不因同步/异步而分叉。

手动注册还是动态扫描：按规模选

动态扫描不是免费的，它引入了一层约定（后缀、导出名、返回类型），新人接手要先学这层规则。所以值不值得，取决于规模：

经验法则：原型期直接写死，别过度设计；一旦工具数过了五个、或者第二个人开始提交工具，就切到注册器。 约定优于配置的收益是随规模复利的，规模不到，它就只是负担。

实践要点与常见坑

• description 是写给模型的接口文档。 工具明明注册成功，模型却不调用或乱传参，九成是 description 没写清。好的描述要覆盖：这个工具做什么、什么场景该用、什么场景不该用、每个参数的语义和边界。自己读了都拿不准何时该用的描述，模型也一样拿不准。
• 返回值永远是字符串。 不要返回 dict 或对象，工具结果最终会以文本形式进入对话历史；结构化数据就 json.dumps 之后再返回。
• 别在工具里 raise，也别依赖注册器的兜底。 兜底 try/except 是保险丝，不是设计。工具内部对可预期的失败（文件不存在、超时、参数缺失）主动返回带上下文的 Error: 信息，模型才有足够线索自我修正。
• 约定后缀要真的过滤。 扫描 *_tool.py 意味着目录里可以放 helpers.py、config.py 这类共享模块而不会被误注册——这是后缀约定除了"标识身份"之外的第二个作用。
• 缓存与热更新的取舍。 本文的注册表进程内只扫一次，改了工具要重启进程。开发期嫌麻烦，可以加一个 reload() 清空缓存重扫；生产环境反而推荐保持只扫一次，行为可预测。
• 后台事件循环是为未来铺路。 CLI 阶段直接 asyncio.run() 也能跑，但迁到 FastAPI 之类的异步 Web 框架时必炸。一开始就用长期存活的后台循环，迁移时工具层零改动。

这套"目录扫描 + 约定导出"的思路不止用于工具：本系列后面讲 Skills（把多步方法论打包给 Agent）和 MCP 客户端（把外部服务器的工具挂进来）时，会看到同样的注册模式反复出现——只是被发现的对象从"函数"换成了"流程"和"远程能力"。

动手清单

1. 从零搭一个 toolbox/ 目录，实现本文的注册器基础版，写两个同步工具（比如查当前时间、统计一段文本的字数），跑通"丢文件即注册"。
2. 故意制造一个坏工具：新建一个有语法错误的 broken_tool.py，确认注册器打警告跳过它，其余工具正常工作。
3. 给注册器补上异步支持，把 ping_url 类工具接进来，用日志确认 is_async=True 且结果正常返回。
4. 做一个对照实验：把某个工具的 description 改成一句含糊的话（如"处理数据"），观察模型的调用行为变差；再改回精确描述，体会 description 对 Agent 表现的影响。
5. 进阶：给 run_tool 的 **kwargs 传一个 caller="cli" 上下文，在某个工具里读取并打印它，体验 kwargs 扩展位的用法。

*本文为学习笔记，知识框架参考自叶小钗「生产级 Agent」系列课程；文字与代码示例为本站原创整理。*