导读: 上一篇我们拆解了 Agent Loop,但循环里的 run_tool() 还是个黑盒。当工具从 3 个涨到 50+ 个,再加 MCP 外部工具动态涌入,if/elif 分发链必然失控。本文用 10505 字节的真实源码,拆解 ToolRegistry 自注册机制——加新工具不用改循环,也不用改任何中心配置,只在文件末尾加一行 registry.register()

从 if/elif 说起:工具一多就失控

先看最直觉的实现。s01 的循环里,执行工具就一个 run_tool()

def run_tool(name, args):
    if name == "terminal":
        return handle_terminal(args)
    elif name == "read_file":
        return handle_read_file(args)
    elif name == "write_file":
        return handle_write_file(args)
    # 再加一个工具,这里就多一个 elif

三个工具还好。那五十个呢?一百个呢?再加 MCP 外部工具动态注册呢?

每加一个工具就要改分发函数,改完还要回归测试整个循环。这条路走不远。

关键洞察:加工具不应该改循环,也不应该改任何中心配置文件。

从黑盒到注册表:三个方法搞定一切

s02 的核心改动,是把 run_tool() 黑盒拆成一张表 + 三个方法。

先看表里的条目长什么样:

@dataclass
class ToolEntry:
    """A registered tool with its metadata and handler."""
    name: str
    toolset: str
    schema: dict
    handler: Callable

一个条目四件事:名字、所属工具集、OpenAI 格式的 schema、真正的处理函数。

注册表本体就是一个字典:

self._tools: dict[str, ToolEntry]

三个方法,各管一件事:

  • register(name, toolset, schema, handler):往字典里塞条目
  • dispatch(name, args, **kwargs):按名字查表执行,查不到返回 {"error": f"Unknown tool: {name}"}。注意 kwargs 是预留的——s03 传 conn,s09 传 session,现在不用管
  • get_definitions(enabled_toolsets=None):返回 OpenAI 格式的工具定义列表,可按 toolset 过滤。不同场景的 agent 只看到允许的工具

dispatch 的实现简单到让人怀疑:

def dispatch(self, name: str, args: dict, **kwargs) -> str:
    """Look up a tool by name and execute its handler."""
    entry = self._tools.get(name)
    if not entry:
        return json.dumps({"error": f"Unknown tool: {name}"})
    return entry.handler(args, **kwargs)

查表、执行、完事。没有 if/elif,没有 switch-case,就是一个字典的 get()

对比图:s01 if/elif 黑盒 vs s02 注册表分发

自注册模式:handler 和 register 紧挨着

光有注册表还不够,关键是工具怎么"自己走进"这张表。

模式很简单:每个工具 = 一个 handler 函数 + 紧跟其后的 registry.register() 调用。

看 terminal 工具的完整代码:

def handle_terminal(args, **kwargs):
    command = args.get("command", "")
    for blocked in BLOCKED_COMMANDS:
        if blocked in command:
            return json.dumps({"error": f"Blocked: {blocked}"})
    try:
        result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=30)
        output = result.stdout + result.stderr
        return output[:10000] if output else "(no output)"
    except subprocess.TimeoutExpired:
        return "(command timed out after 30s)"
    except Exception as exc:
        return f"(error: {exc})"

registry.register(
    name="terminal", toolset="terminal",
    schema={"name": "terminal", "description": "Run a shell command and return its output.", "parameters": {...}},
    handler=handle_terminal,
)

注意几个细节:

  • 黑名单命令直接拦截,返回 JSON 错误
  • 30 秒超时,防止 Agent 跑飞
  • 输出截断到 10000 字符,防止刷屏

这个仓库一共四个工具,分三个 toolset:

工具 toolset 关键行为
terminal terminal 黑名单 + 30s 超时 + 10000 截断
read_file file 读 100KB 上限,空文件返回 "(empty file)"
write_file file 自动建目录,返回写入字符数
web_search web stub 占位,接真实 API 只改这一个函数

信息图:四个工具一览,按 toolset 分组展示

web_search 的 stub 实现值得单独看一眼:

return json.dumps({
    "note": "web_search is a stub in this teaching version",
    "query": query,
})

要接真实搜索 API?只改这个函数,注册表、循环、其他工具一概不动。

导入即注册:Python 顶层代码的执行时机

现在问题来了:registry.register() 写在文件末尾,谁来执行它?

答案是 Python 的导入机制。import 一个模块时,模块里的顶层代码会立即执行registry.register(...) 就是顶层代码,只要模块被导入,注册就自动发生。

所以整个系统的导入链是单向的:

registry.py          (不导入任何工具)
     ^
tools/*.py           (导入 registry,注册自己)
     ^
model_tools.py       (导入所有 tools/*.py,触发注册)
     ^
run_agent.py         (导入 model_tools.py,使用接口)

注册表在最底层,不依赖任何工具。这是整个设计能工作的关键——依赖方向永远是从工具指向注册表,而不是反过来

编排层甚至不用写死导入语句,用 importlib 循环导入:

import importlib

_modules = [
    "tools.web_tools",       # import → register("web_search") 自动执行
    "tools.terminal_tool",   # import → register("terminal") 自动执行
    "tools.file_tools",      # import → register("read_file"/"write_file") 自动执行
    # ... 20+ 模块
]
for mod in _modules:
    importlib.import_module(mod)

模块名是字符串,放在列表里循环。加新工具?列表加一行字符串。

流程图:import → register 自动执行 → 注册表填充

循环怎么用:一行没改

回到 Agent Loop。s02 与 s01 的关键差异,在循环启动时和调用时:

ENABLED_TOOLSETS = ["terminal", "file", "web"]
tools = registry.get_definitions(ENABLED_TOOLSETS)   # 循环启动时取定义
...
output = registry.dispatch(tool_name, tool_args)      # 按名字查表执行

get_definitions 按 toolset 过滤,循环只看到允许的工具。dispatch 按名字查表,执行完返回字符串。

循环本身的骨架——消息历史、模型调用、结果回填——一行都没改。工具系统的演进,被完全封装在注册表后面。

架构图:三层导入链,registry 在最底层

真实 Hermes 的进阶设计

教学版只讲 schema / handler / dispatch 三件事。真实 Hermes 的注册表还多了两个标记:

is_async:异步工具桥接

异步工具不能直接 asyncio.run() 包一层——每次创建新循环然后关闭,工具内部缓存的 httpx / AsyncOpenAI 客户端绑定在旧循环上,循环关了客户端就废了。真实实现是注册表自动走持久化事件循环桥接,让异步工具和同步工具在同一个循环里共存。

check_fn:可用性检查

get_definitions() 只返回通过检查的工具。没有 BROWSERBASE_API_KEY?browser 工具就不会出现在 schema 列表里。Agent 根本不知道有这个工具存在,省得它试了又失败。

MCP 外部工具

编排层 discover_mcp_tools() 发现外部工具后,注册进同一个注册表。内部工具和外部工具在 Agent 眼里没有区别——都是 dispatch(name, args)。这块在 s16 展开。

避坑指南:理解设计的关键

为什么不反向依赖? 如果 registry 导入 tools,那加工具就得改 registry 的 import 语句,又变成中心配置了。现在 tools 导入 registry,加工具完全不动核心代码。

工具 vs 技能 vs MCP 的区别? 工具是原子操作(terminal、read_file),技能是工具的组合策略(s08 展开),MCP 是外部工具的统一入口(s16 展开)。三层概念,别混。

权限检查在哪做? s02 不涉及权限。黑名单只是最基础的防护,真正的权限体系在 s09。教学边界要清楚。

小结

工具系统从 s01 的黑盒演进为 s02 的注册表,核心就三件事:

  1. ToolEntry:四字段描述一个工具
  2. ToolRegistry:register / dispatch / get_definitions 三个方法
  3. 自注册模式:handler + register() 紧挨着,导入即注册

加新工具,只加一个文件,不动循环,不动配置。

下一篇,我们给 Agent 装上记忆——SQLite 会话存储,让对话跨重启存活,再加上提示词组装。到时候 kwargs 里预留的 conn 就派上用场了。


你在实际项目里遇到过工具分发失控的情况吗?是用了注册表模式,还是硬扛 if/elif 到底?评论区聊聊你的解法。

参考文献

  • Hermes Agent 教学仓库:agents/s02_tool_system.py(本文代码素材,10505 字节真实可运行)
  • Hermes Agent 教学仓库:docs/zh/s02-tool-system.md(导入链、自注册原理、is_async/check_fn 进阶设计)
  • 前篇参考:《百行 Agent Loop:让模型从"会说话"变成"会干活"》(Agent Loop 骨架)

📥 源码获取:如需本系列全部源码,请在以下链接克隆:
https://gitcode.com/ganxin7932508/learn-hermes-agent.git

Logo

这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容,并连接云服务、办公空间等稀缺资源,助你专注创造,无忧运营。

更多推荐