第 2 期：工具注册表

核心概念：声明式工具发现 —— 每个工具文件自注册 schema + handler，框架自动扫描装载，运行时按名分发

概念讲解

第 1 期中，我们把工具硬编码在 TOOLS 列表和 _handle_tool_call() 的 if-elif 分支里。加一个新工具就得改两个地方 —— schema 和 handler。工具一多，这种模式会崩：70 个工具就是 70 个 elif 分支。

Hermes 的解决方案是注册表模式：

工具文件 A（import 时自动执行）        ToolRegistry（全局单例）
──────────────────────────           ─────────────────────
registry.register(                    _tools = {
  name="terminal",                      "terminal": ToolEntry(...),
  schema={...},         ──注册──→      "read_file": ToolEntry(...),
  handler=my_func,                      "calculator": ToolEntry(...),
)                                     }

工具文件 B（import 时自动执行）
──────────────────────────
registry.register(
  name="read_file",     ──注册──→
  schema={...},
  handler=read_fn,
)

使用时：

model_tools.py: "模型想调 terminal"
       │
       ▼
registry.dispatch("terminal", {"command": "ls"})
       │
       ▼
找到 _tools["terminal"].handler → 执行 → 返回 JSON 字符串

这是典型的控制反转（IoC）：框架不主动调用工具，而是工具自己往框架里注册。新增工具只需要写一个文件，调一次 register()，不改框架任何代码。

源码关键片段

文件：tools/registry.py:77-99 — ToolEntry 数据类

class ToolEntry:
    """单个已注册工具的元数据。"""

    __slots__ = (
        "name",        # 工具名，如 "terminal"
        "toolset",     # 所属工具集，如 "terminal"
        "schema",      # OpenAI 格式的 JSON Schema
        "handler",     # 可调用对象，接收 (args, **kwargs) 返回 JSON str
        "check_fn",    # 可用性检查函数（如检测 Docker 是否安装）
        "requires_env",# 需要的环境变量列表
        "is_async",    # 是否异步 handler
        "description", # 工具描述
        "emoji",       # 展示用的 emoji
        "max_result_size_chars",     # 结果最大字符数
        "dynamic_schema_overrides",  # 运行时动态修改 schema 的回调
    )

文件：tools/registry.py:151-168 — ToolRegistry 单例

class ToolRegistry:
    """单例注册表 —— 收集所有工具文件注册的 schema + handler。"""

    def __init__(self):
        self._tools: Dict[str, ToolEntry] = {}         # name → ToolEntry
        self._toolset_checks: Dict[str, Callable] = {} # toolset → check_fn
        self._toolset_aliases: Dict[str, str] = {}     # alias → canonical name
        self._lock = threading.RLock()  # 线程安全（MCP 动态刷新可能并发写）
        self._generation: int = 0       # 变更计数器，用于缓存失效

文件：tools/registry.py:234-290 — register() 核心方法

    def register(
        self,
        name: str,           # 工具名
        toolset: str,        # 所属工具集
        schema: dict,        # OpenAI function schema
        handler: Callable,   # 实际执行函数
        check_fn: Callable = None,   # 可用性检查
        requires_env: list = None,   # 需要的环境变量
        override: bool = False,      # 是否允许覆盖已注册的同名工具
    ):
        with self._lock:
            existing = self._tools.get(name)
            if existing and existing.toolset != toolset:
                if not override:
                    # 拒绝覆盖 —— 防止插件意外踩掉内置工具
                    logger.error("Tool registration REJECTED: '%s'", name)
                    return
            self._tools[name] = ToolEntry(...)
            self._generation += 1  # 触发下游缓存失效

文件：tools/registry.py:390-413 — dispatch() 分发执行

    def dispatch(self, name: str, args: dict, **kwargs) -> str:
        """按名称执行工具 handler。"""
        entry = self.get_entry(name)
        if not entry:
            return json.dumps({"error": f"Unknown tool: {name}"})
        try:
            if entry.is_async:
                from model_tools import _run_async
                return _run_async(entry.handler(args, **kwargs))
            return entry.handler(args, **kwargs)
        except Exception as e:
            return json.dumps({"error": f"Tool execution failed: {e}"})

文件：tools/registry.py:337-385 — get_definitions() 过滤出可用工具的 schema

    def get_definitions(self, tool_names: Set[str], quiet: bool = False) -> List[dict]:
        """返回 OpenAI 格式的工具 schema 列表。

        只包含 check_fn() 返回 True 的工具（如 Docker 未安装则终端工具不可用）。
        check_fn 结果有 30s TTL 缓存，避免每次都探测外部状态。
        """
        result = []
        for name in sorted(tool_names):
            entry = self._tools.get(name)
            if not entry:
                continue
            if entry.check_fn and not _check_fn_cached(entry.check_fn):
                continue  # 工具不可用，跳过
            schema_with_name = {**entry.schema, "name": entry.name}
            # 应用运行时动态 schema 覆盖（如 delegate_task 的并发上限描述）
            if entry.dynamic_schema_overrides:
                schema_with_name.update(entry.dynamic_schema_overrides())
            result.append({"type": "function", "function": schema_with_name})
        return result

文件：tools/registry.py:42-70 — 自动发现机制（AST 扫描）

def _module_registers_tools(module_path: Path) -> bool:
    """用 AST 检查文件是否包含顶层 registry.register() 调用。"""
    source = module_path.read_text(encoding="utf-8")
    tree = ast.parse(source)
    return any(_is_registry_register_call(stmt) for stmt in tree.body)

def discover_builtin_tools(tools_dir=None) -> List[str]:
    """扫描 tools/ 目录，import 所有包含 register() 的文件。"""
    tools_path = Path(tools_dir) or Path(__file__).resolve().parent
    module_names = [
        f"tools.{path.stem}"
        for path in sorted(tools_path.glob("*.py"))
        if path.name not in {"__init__.py", "registry.py", "mcp_tool.py"}
        and _module_registers_tools(path)  # AST 静态检查，不 import 就能判断
    ]
    for mod_name in module_names:
        importlib.import_module(mod_name)  # import 触发模块顶层的 register()
    return module_names

文件：tools/registry.py:544 — 模块级单例

# 整个进程只有一个 registry 实例
registry = ToolRegistry()

设计决策

Hermes 的选择	为什么这样做	没选的替代方案	替代方案的代价（为什么不选）
AST 静态扫描 判断文件是否注册工具	不需要 import 就能过滤掉无关文件（如 ansi_strip.py），避免加载不必要的依赖	直接 import 所有 tools/*.py	某些工具依赖可选库（playwright、modal），全量 import 会在缺库时崩溃
模块级 register() 调用（import 即注册）	声明式、零样板代码；新增工具只写一个文件	在中心文件维护工具注册列表	改一个工具要改两个文件，70+ 工具时列表维护成噩梦
_generation 计数器做缓存失效	下游（get_tool_definitions）可以基于 generation 做 memo；MCP 动态刷新时自动失效	每次调用都重新计算	频繁调用 get_definitions 时浪费 ~7ms（registry walk + check_fn probe）
check_fn 有 30s TTL 缓存	检查函数可能探测 Docker/Modal/Playwright 等外部状态，不缓存会拖慢每次 API 调用	不缓存，每次实时探测	10 次工具调用 = 10 次 Docker ping，每次多 50-200ms
覆盖保护（override=False 默认拒绝同名注册）	防止插件意外覆盖内置工具；需要覆盖时必须显式 opt-in	最后注册者赢	调试地狱 —— 某个插件悄悄替换了 terminal 工具，排查半天
handler 必须返回 JSON 字符串	统一序列化格式；异常也包装成 {"error": "..."}	允许返回任意类型，框架做序列化	框架不知道怎么序列化自定义对象；错误处理不一致

复刻代码说明

本期复刻代码在 dongrealm-hermes-agent (day2 分支) 中迭代，功能说明：实现 ToolRegistry 类，将第 1 期的硬编码工具改为注册表驱动 —— 工具自注册 + 自动发现 + 按名分发。

行为和第 1 期完全一致（两个工具、交互式对话），但内部架构从硬编码变成了注册表驱动。

## 常见陷阱

1. **register() 必须在模块顶层调用** — 如果放在函数内部，`discover_builtin_tools()` 的 AST 扫描不会发现它（只检查 `tree.body` 顶层语句）。源码验证：`tools/registry.py:50`，`_is_registry_register_call` 只匹配 `ast.Expr` 类型的顶层节点。

2. **handler 返回值必须是 JSON 字符串** — 不是 dict，是 `json.dumps(...)` 后的字符串。源码中 `dispatch()` 直接 return handler 的返回值，不做二次序列化。如果你返回 dict，模型收到的是 Python repr 而非有效 JSON。

3. **线程安全不能忽略** — Gateway 模式下多个消息并发处理，MCP 服务器动态刷新也可能并发写。源码用 `threading.RLock()` 保护 `_tools` 字典。如果你的 Agent 只在单线程跑（CLI 模式），可以暂时不加锁。

4. **循环依赖陷阱** — `registry.py` 不能 import 任何工具文件或 `model_tools.py`。依赖链是单向的：`registry ← tools/* ← model_tools ← run_agent`。反向 import 会导致循环 import 死锁。

5. **`discover_builtin_tools()` 只跑一次** — 在 `model_tools.py` 模块加载时执行。之后动态新增的工具（如 MCP 服务器）通过 `register()` 直接注册，不走发现流程。