Tool Registry 与发现

上一章把 prompt 的事说完了。这一章我们看 prompt 里那个 tools 数组里的每个元素 是怎么从一个 Python 函数文件变成给 LLM 看的 schema，再变成被 LLM 调用时的 dispatch。 这一整套机制叫"工具系统"。Hermes 的工具系统在 tools/registry.py 和 model_tools.py， 约 1500 行。

6.1一个工具的全貌

先看一个真实工具的完整文件。tools/clarify_tool.py——它的功能是"让 Agent 向用户提问"：

tools/clarify_tool.py（简化到核心 ~50 行）
import json
from typing import Optional, List, Callable
from tools.registry import registry, tool_error

MAX_CHOICES = 4

# ─── 1. Handler：实际执行的 Python 函数 ─────────────────
def clarify_tool(
    question: str,
    choices: Optional[List[str]] = None,
    callback: Optional[Callable] = None,
) -> str:
    """Ask the user a question; optionally with choices."""
    if not question.strip():
        return tool_error("Question text is required.")

    if callback is None:
        return json.dumps({"error": "Clarify not available in this context."})

    user_response = callback(question, choices)
    return json.dumps({
        "question": question,
        "user_response": user_response,
    })


# ─── 2. check_fn：判断工具是否可用 ────────────────────
def check_clarify_requirements() -> bool:
    return True   # 永远可用


# ─── 3. Schema：给 LLM 看的 OpenAI Function 描述 ──────
CLARIFY_SCHEMA = {
    "name": "clarify",
    "description": (
        "Ask the user a question when you need clarification or a "
        "decision. Two modes: 1) multiple choice (up to 4); "
        "2) open-ended. Use when ambiguity blocks progress."
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "question": {
                "type": "string",
                "description": "The question to present.",
            },
            "choices": {
                "type": "array",
                "items": {"type": "string"},
                "maxItems": MAX_CHOICES,
            },
        },
        "required": ["question"],
    },
}


# ─── 4. 注册：模块导入时自动执行 ───────────────────
registry.register(
    name="clarify",
    toolset="clarify",
    schema=CLARIFY_SCHEMA,
    handler=lambda args, **kw: clarify_tool(
        question=args.get("question", ""),
        choices=args.get("choices"),
        callback=kw.get("callback"),
    ),
    check_fn=check_clarify_requirements,
    emoji="❓",
)

记住这四件套——所有 Hermes 工具都长这样：

1. Handler：真正干活的 Python 函数。
2. check_fn：判断工具是否可用（如环境变量是否设、依赖是否装）。
3. Schema：给 LLM 的接口契约。
4. register 调用：模块顶层执行，把四件事都登记到全局 registry。

6.2Registry：单例注册中心

所有工具的注册都进同一个全局 dict。tools/registry.py:

tools/registry.py:151-200
import threading
from dataclasses import dataclass
from typing import Callable, Dict

@dataclass
class ToolEntry:
    name: str
    toolset: str
    schema: dict
    handler: Callable
    check_fn: Callable = None
    requires_env: list = None
    is_async: bool = False
    emoji: str = ""
    max_result_size_chars: int = None
    dynamic_schema_overrides: Callable = None

class ToolRegistry:
    """单例 — 收集所有 tool 的 schema + handler。"""

    def __init__(self):
        self._tools: Dict[str, ToolEntry] = {}
        self._toolset_checks: Dict[str, Callable] = {}
        self._lock = threading.RLock()
        # Generation counter — 每次变更 +1，给 cache 用
        self._generation: int = 0

register 方法的全貌

tools/registry.py:200-260 (摘要)
def register(self, name, toolset, schema, handler,
             check_fn=None, requires_env=None,
             is_async=False, override=False, ...):
    """Register a tool. Called at module-import time by each tool file."""
    with self._lock:
        existing = self._tools.get(name)

        # 防意外覆盖：除非显式 override=True
        if existing and existing.toolset != toolset:
            both_mcp = (existing.toolset.startswith("mcp-")
                        and toolset.startswith("mcp-"))
            if both_mcp:
                # MCP 之间的覆盖正常（server 重连刷新工具）
                logger.debug("MCP overwriting MCP for %s", name)
            elif override:
                logger.info("%s: '%s' override '%s' (explicit)",
                            name, toolset, existing.toolset)
            else:
                logger.error("Tool '%s' REJECTED: would shadow toolset '%s'",
                              name, existing.toolset)
                return

        self._tools[name] = ToolEntry(
            name=name, toolset=toolset, schema=schema, handler=handler,
            check_fn=check_fn, ...)

        if check_fn and toolset not in self._toolset_checks:
            self._toolset_checks[toolset] = check_fn

        self._generation += 1

# 模块级单例
registry = ToolRegistry()

四个值得深挖的设计决策：

① RLock 而不是普通 Lock

MCP server 可以在运行时刷新它的工具列表（比如 GitHub 加了新 API）。 Hermes 的 MCP 客户端会调 registry.register() 更新。 与此同时，主 Agent 循环可能正在调 registry.get_definitions() 读 schema。 读写竞争必须用锁。

用 RLock（可重入）是因为有些路径在持锁状态下又调了别的同样需要锁的方法。 普通 Lock 会死锁。

② _generation 计数器

外部缓存（比如 schema 的 OpenAI 兼容包装）需要知道 registry 啥时候变了。 每次 mutation 把 _generation += 1，外部就拿这个数当 cache key。 命中就是字典查找，不命中就重算。不需要外部主动 invalidate。

# 调用方代码（model_tools.py）
cache_key = (
    frozenset(enabled_toolsets),
    registry._generation,        ## ← 这一行
    config_mtime,
)
cached = _tool_defs_cache.get(cache_key)
if cached is not None:
    return cached
# 否则重算

③ override 显式打开

Plugin 想替换一个内置工具（比如换一个更强的 browser_navigate 实现）？ 必须传 override=True。否则 Hermes 拒绝，并在 log 里大声警告。 这条规则是防止 plugin 互相意外覆盖——一个用户装了两个声明 browser 工具的 plugin， 没有 override 强制，谁先装谁赢，bug 隐蔽且难调。

④ MCP 互相覆盖默认允许

两个 MCP server 都暴露 read_file？前面注册的会被后面注册的覆盖，但 log 是 debug 级。 原因是MCP 重连时会刷新整套工具——你不想每次重连都出 ERROR 日志。

6.3自动发现

问题：谁来 import 这些 tool 文件？让用户在某个 __init__.py 里维护 79 个 import？

Hermes 的回答：自动扫描，按文件名导入。在 model_tools.py 第 180 行：

model_tools.py:180
# 触发整个 tools/ 目录扫描，每个 .py 文件被 import 一次
discover_builtin_tools()

discover_builtin_tools() 内部做的事：

model_tools.py (节选)
def discover_builtin_tools():
    """扫描 tools/*.py 并 import，触发模块顶层的 registry.register() 调用。"""
    tools_dir = Path(__file__).parent / "tools"
    for tool_file in tools_dir.glob("*.py"):
        if tool_file.startswith("_"): continue          # skip _utils.py 等
        module_name = f"tools.{tool_file.stem}"
        try:
            importlib.import_module(module_name)
        except Exception as e:
            logger.warning("Failed to import %s: %s", module_name, e)
            # 一个 tool import 失败不该挂掉整个 Agent
            continue

关键设计：

• 一个 import 失败不影响其他：用 try/except，warning 即可。
• 下划线开头跳过：约定 _utils.py 这种辅助文件不是工具。
• 模块级 register：tool 文件被 import 的副作用就是注册。 你不用显式调 init 函数。

为什么这种"魔法"可以接受

"模块级副作用"在 Python 圈通常被视为反模式——容易出隐藏 bug。 但工具发现是少数合理的例外：

1. 规则简单：所有 tools/*.py 都会被 import，模块顶层调 register。
2. 不会引入循环依赖：tool 文件只依赖 tools.registry 和标准库。
3. 排查容易：log 里有 "Failed to import" 警告。
4. 替代方案更糟：维护一份 79 项的显式 import 列表，每次加工具都要改两处。

6.4Plugin 工具：另一条注册路径

上面是 内置工具的注册路径。Plugin 工具走另一条路： hermes_cli/plugins.py 里有 PluginManager.register_tool()。

hermes_cli/plugins.py:317-360
def register_tool(
    self,
    name: str,
    toolset: str,
    schema: dict,
    handler: Callable,
    check_fn: Callable = None,
    requires_env: list = None,
    is_async: bool = False,
    description: str = "",
    emoji: str = "",
    override: bool = False,
) -> None:
    """Register a tool in the global registry AND mark it as plugin-provided."""
    from tools.registry import registry

    registry.register(
        name=name, toolset=toolset, schema=schema, handler=handler,
        check_fn=check_fn, requires_env=requires_env,
        is_async=is_async, description=description, emoji=emoji,
        override=override,
    )
    self._manager._plugin_tool_names.add(name)   # 多记一个"这是 plugin 来的"

用户的 plugin 长这样：

~/.hermes/plugins/fortune/__init__.py
import json, random

def fortune_handler(args, **kw):
    quotes = ["Talk is cheap.", "Make it work, make it right..."]
    return json.dumps({"quote": random.choice(quotes)})

FORTUNE_SCHEMA = {
    "name": "fortune_cookie",
    "description": "Return a random programming fortune quote.",
    "parameters": {"type": "object", "properties": {}, "required": []},
}

# 注意：不是直接调 registry.register，而是通过 ctx
def register(ctx):
    ctx.register_tool(
        name="fortune_cookie",
        toolset="fortune",
        schema=FORTUNE_SCHEMA,
        handler=fortune_handler,
        emoji="🥠",
    )

插件的 register(ctx) 函数由 PluginManager 在加载 plugin 时调用。 ctx 提供 register_tool 等便捷方法。这样 Hermes 知道哪些工具来自 plugin， 方便诊断和卸载。

6.5get_definitions：给 LLM 准备 schema

注册完工具，下一步是把 schema 提供给 LLM。在 model_tools.py：

model_tools.py:264-326 (摘要)
def get_tool_definitions(
    enabled_toolsets: List[str] = None,
    disabled_toolsets: List[str] = None,
    quiet_mode: bool = False,
) -> List[Dict[str, Any]]:
    """Get tool definitions for LLM API calls with toolset-based filtering.

    All tools must belong to a toolset to be accessible. Tools whose
    check_fn() returns False are filtered out (e.g. terminal tool when
    Docker isn't running).
    """
    # 缓存机制：用 (toolsets, registry._generation, config_mtime) 做 key
    if quiet_mode:
        cache_key = _build_cache_key(enabled_toolsets, disabled_toolsets)
        cached = _tool_defs_cache.get(cache_key)
        if cached is not None:
            return list(cached)   # shallow copy 避免污染

    # 实际计算
    result = _compute_tool_definitions(enabled_toolsets, disabled_toolsets, quiet_mode)
    if quiet_mode:
        _tool_defs_cache[cache_key] = result
    return list(result)

_compute_tool_definitions 的逻辑：

1. 从 toolsets.py 的 TOOLSETS 字典展开 enabled_toolsets 到具体 tool name 集合。
2. 从 disabled_toolsets 中拿到要排除的名字。
3. 调 registry.get_definitions(tool_names)。
4. Registry 内部对每个 tool 调 check_fn()——返回 False 的剔掉。
5. 给每个 schema 应用 dynamic_schema_overrides()（如果有）。
6. 包装成 OpenAI {"type": "function", "function": {...}} 格式。

check_fn 的 TTL 缓存

check_fn 可能做实际的 I/O（"docker 在跑吗？"）。每次 get_definitions 都 fresh 调一次太贵。Registry 内部对 check_fn 结果有 30 秒 TTL 缓存：

@functools.lru_cache(maxsize=128)
def _check_fn_with_ttl(fn, ttl_bucket):
    return fn()

def _check_fn_cached(fn):
    # 把 TTL 切成 30 秒"桶"，同一桶内同一 fn 调用结果一致
    bucket = int(time.time()) // 30
    return _check_fn_with_ttl(fn, bucket)

30 秒选这个数字是权衡：太短，频繁 I/O；太长，用户改了配置（启用 terminal） 要等 60 秒才生效。30 秒让用户感觉"近实时"。

Dynamic Schema Overrides

有些工具的 schema 描述依赖运行时配置。例如 delegate_task 的描述里说"最多并发 N 个子 agent"，N 来自 config.yaml。Hermes 让工具注册 dynamic_schema_overrides 回调：

def _dynamic_delegate_schema():
    cfg = load_config()
    max_concurrent = cfg.get("delegation.max_concurrent_children", 3)
    return {
        "description": f"Delegate a subtask... max {max_concurrent} concurrent."
    }

registry.register(
    name="delegate_task",
    schema=STATIC_SCHEMA,
    dynamic_schema_overrides=_dynamic_delegate_schema,
    ...
)

每次 get_tool_definitions 被调用时，dynamic override 被求值。 cache key 包含 config_mtime，所以 config 改了 cache 自动失效。

6.6handle_function_call：执行管道

LLM 返回 tool_calls 后，model_tools.py 的 handle_function_call 接管：

model_tools.py:741-894 (摘要)
def handle_function_call(
    function_name: str,
    function_args: Dict[str, Any],
    task_id: Optional[str] = None,
    tool_call_id: Optional[str] = None,
    session_id: Optional[str] = None,
    enabled_tools: Optional[List[str]] = None,
    skip_pre_tool_call_hook: bool = False,
) -> str:
    """Main dispatcher — routes calls to the registry."""

    # 1. 参数类型转换（"42" → 42 等）
    function_args = coerce_tool_args(function_name, function_args)

    # 2. Agent-loop 内置工具（todo / memory）不走这里
    if function_name in _AGENT_LOOP_TOOLS:
        return json.dumps({"error": f"{function_name} must be handled by agent loop"})

    # 3. 调 pre_tool_call hook（plugin 可以阻止）
    if not skip_pre_tool_call_hook:
        block_message = get_pre_tool_call_block_message(
            function_name, function_args,
            task_id=task_id or "",
            session_id=session_id or "",
        )
        if block_message is not None:
            return json.dumps({"error": block_message}, ensure_ascii=False)

    # 4. ACP/Zed edit approval（GUI 集成时）
    edit_block = maybe_require_edit_approval(function_name, function_args)
    if edit_block is not None:
        return edit_block

    # 5. 实际 dispatch — 测时长
    _start = time.monotonic()
    result = registry.dispatch(
        function_name, function_args,
        task_id=task_id,
        ...
    )
    duration_ms = int((time.monotonic() - _start) * 1000)

    # 6. post_tool_call hook（plugin 观察用）
    invoke_hook("post_tool_call",
                tool_name=function_name, args=function_args,
                result=result, duration_ms=duration_ms,
                task_id=task_id or "")

    # 7. transform_tool_result hook（plugin 可以改写结果）
    hook_results = invoke_hook("transform_tool_result", ...)
    for hr in hook_results:
        if isinstance(hr, str):
            result = hr
            break

    return result

这 7 步可以分成三组：

registry.dispatch

registry.dispatch 是最里层的执行：

tools/registry.py:260-305
def dispatch(self, name: str, args: dict, **kwargs) -> str:
    """Execute a tool handler by name.

    * Async handlers bridged via _run_async()
    * All exceptions caught and returned as JSON error
    """
    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:
        logger.exception("Tool %s dispatch error: %s", name, e)
        raw = f"Tool execution failed: {type(e).__name__}: {e}"
        sanitized = _sanitize_tool_error(raw)
        return json.dumps({"error": sanitized})

三个细节：

1. 异常一定 catch。Agent 主循环不应该被工具异常打死。
2. 错误清洗：_sanitize_tool_error 把 <tool_call>、```、CDATA 这种"框架字符串"从异常消息里剥掉——免得错误信息里有 <tool_call> 让模型把它解析成嵌套工具调用。
3. async handler 自动桥接：用 _run_async 跑 coroutine。

6.7async 桥接的细节

Hermes 同时支持纯 sync 的 CLI 和 asyncio-based 的 Gateway。 工具可以写成 async（处理 HTTP/网络等），但调用方可能在两种环境里。 _run_async 是这个桥：

model_tools.py:84-173 (摘要)
_persistent_loops: Dict[int, asyncio.AbstractEventLoop] = {}

def _run_async(coro):
    """Run an async coroutine from a sync context."""
    try:
        running = asyncio.get_running_loop()
    except RuntimeError:
        running = None

    if running is not None:
        # 当前线程已经在 event loop 里 — 起一个临时线程跑 asyncio.run
        return _run_in_thread(coro)
    else:
        # 没 loop — 用持久 loop 跑（per-thread）
        tid = threading.get_ident()
        loop = _persistent_loops.get(tid)
        if loop is None or loop.is_closed():
            loop = asyncio.new_event_loop()
            _persistent_loops[tid] = loop
        return loop.run_until_complete(coro)

三种场景：

6.8Schema description 的艺术

LLM 选用哪个工具完全看 description 字段。写好它直接决定工具被 正确调用的频率。

反例

"description": "A tool for searching."     # 模糊到没用
"description": "Use this powerful tool to access advanced search capabilities..."
     # 营销话术，全是噪声

好的范例

看 todo_tool.py 的 description：

"description": (
    "Manage your task list for the current session. Use for complex "
    "tasks with 3+ steps or when the user provides multiple tasks. "
    "Call with no parameters to read the current list.\n\n"
    "Writing:\n"
    "- Provide 'todos' array to create/update items\n"
    "- merge=false (default): replace the entire list with a fresh plan\n"
    "- merge=true: update existing items by id, add any new ones\n\n"
    "Each item: {id, content, status: pending|in_progress|completed|cancelled}\n"
    "List order is priority. Only ONE item in_progress at a time.\n"
    "Mark items completed immediately when done."
)

这段 description 长 250 字，但每一字都"告诉模型何时该用、怎么用、有什么约束"。 注意几个写作技巧：

• 开头一句话总结用途：模型只读前 30 字就能判断"现在该不该考虑这个工具"。
• 触发条件具体："3+ steps or when the user provides multiple tasks"——比"复杂任务"清楚 100 倍。
• 参数语义说清楚：merge=true 和 false 的区别说在 description 而不是只在 parameters 里。LLM 主要读 description。
• 约束写明："Only ONE item in_progress at a time"——避免幻觉。

禁忌：跨 toolset 引用

Hermes 的一条规则：tool schema description 不能提到其他 toolset 的工具名。

反例：

"description": "Search the web. For local files, prefer read_file instead."

问题：如果 read_file 在当前 session 不可用（toolset 没启）， LLM 会被指向一个看不见的工具，可能幻觉调用。Hermes 在 get_tool_definitions 里用 dynamic override 动态加上这种跨引用， 而不是写死在 static schema 里。

6.9本章带走的

• 每个工具是四件套：handler + check_fn + schema + register 调用。模块级副作用注册。
• Registry 是线程安全的全局单例，用 _generation 计数器做外部缓存失效。
• override=True 是显式开关：plugin 想替换内置工具必须主动声明。
• 自动发现：discover_builtin_tools() 扫 tools/*.py 触发模块顶层注册。一个失败不挂全局。
• check_fn TTL 缓存 30 秒：平衡 I/O 成本和实时性。
• dynamic_schema_overrides：让 schema 反映运行时配置（如并发上限）。
• handle_function_call 7 步管道：预处理 → 拦截 → 执行 → 观察。
• async 桥接：CLI 用持久 loop，已在 loop 里就开线程。Per-thread 隔离。
• Schema description 是 prompt engineering。写清楚何时用、怎么用、约束。
• 避免跨 toolset 静态引用——用 dynamic override 动态加。