Hermes Agent 源码-工具系统

Hermes Agent 源码解析

第 3 讲：工具系统深度解析——ToolRegistry、ToolSet、自注册机制与 check_fn 探针

基于 Hermes Agent v0.16.0 源码 · 2026-06-16

一、工具系统：Agent 的"四肢"

前两讲我们建立了全局架构地图并深入了对话循环。这一讲我们聚焦 Hermes 最活跃的代码区域之一——工具系统（Tool System）。工具是 Agent 与外部世界交互的唯一通道：执行命令、读写文件、搜索网页、操控浏览器、调度定时任务……每一轮对话中，LLM 看到的"工具列表"就是 Agent 的能力边界。

📦 源码仓库

https://github.com/NousResearch/hermes-agent

本地源码：~/.hermes/hermes-agent/

本讲核心文件：

tools/registry.py（589 行）— 注册中心

toolsets.py（912 行）— 工具集定义

model_tools.py（1229 行）— 编排层

tools/tool_search.py（735 行）— 渐进式工具披露

二、整体架构：三层分离设计

Hermes 工具系统采用经典的三层架构，每层职责清晰、互不耦合：

┌──────────────────────────────────────────────────────────────┐
│                    第 1 层：工具实现层                          │
│                                                              │
│  tools/terminal_tool.py   ← 2684 行，终端执行                  │
│  tools/web_tools.py       ← 1569 行，网页搜索/抓取             │
│  tools/browser_tool.py    ← 浏览器自动化                       │
│  tools/file_tools.py      ← 文件读写/搜索                      │
│  tools/skills_tool.py     ← 技能系统                           │
│  tools/kanban_tools.py    ← 多 Agent 协调                      │
│  … (共 33+ 工具模块)                                          │
│                                                              │
│  每个模块在文件末尾调用 registry.register() 自注册              │
│  包含: schema (JSON Schema) + handler (处理函数) + check_fn    │
├──────────────────────────────────────────────────────────────┤
│                    第 2 层：注册中心 (Registry)                 │
│                                                              │
│  tools/registry.py — ToolRegistry 单例                         │
│                                                              │
│  • ToolEntry: 工具元数据容器 (name, toolset, schema, handler…) │
│  • discover_builtin_tools(): AST 扫描 → 动态导入               │
│  • check_fn TTL cache: 30s 缓存外部探针结果                     │
│  • _generation 计数器: 变更感知 + 缓存失效                      │
│  • dispatch(): 统一分发入口，async 桥接                        │
│  • get_definitions(): 按需产出 OpenAI 格式 schema               │
├──────────────────────────────────────────────────────────────┤
│                    第 3 层：编排层 (Orchestration)              │
│                                                              │
│  toolsets.py — 工具集定义与解析                                 │
│  • TOOLSETS dict: 60+ 预定义工具集 (hermes-cli, hermes-…)      │
│  • resolve_toolset(): 递归解析，含循环检测                      │
│  • _HERMES_CORE_TOOLS: 核心工具白名单 (~55 个)                 │
│                                                              │
│  model_tools.py — 公共 API + 分发流水线                         │
│  • get_tool_definitions(): 过滤 + 缓存 + 动态 schema 注入      │
│  • handle_function_call(): 参数修正 → 中间件 → 分发 → 后处理   │
│  • _run_async(): 线程安全的 async 桥接                         │
│  • coerce_tool_args(): LLM 参数类型修正                         │
│  • tool_search 集成: 渐进式工具披露                             │
└──────────────────────────────────────────────────────────────┘

导入链 (循环导入安全):
  tools/registry.py  (零外部依赖)
         ↑
  tools/*.py  (导入 registry.register)
         ↑
  model_tools.py  (导入 registry + 所有工具模块)
         ↑
  run_agent.py, cli.py, gateway/run.py

设计哲学：注册中心不依赖任何工具模块，工具模块只依赖注册中心，编排层最后导入一切。这个"V 字形"导入链彻底消除了循环导入——5000+ 行代码无需 if TYPE_CHECKING 或延迟导入。

三、自注册机制：AST 扫描 + 模块级注册

1. discover_builtin_tools() — 零配置发现

启动时，Hermes 不需要配置文件来声明"有哪些工具"。它通过 AST 静态分析 自动发现：

📄 tools/registry.py (第 29-74 行)

def _is_registry_register_call(node: ast.AST) -> bool:
    """检测 AST 节点是否为 registry.register(...) 调用"""
    if not isinstance(node, ast.Expr) or not isinstance(node.value, ast.Call):
        return False
    func = node.value.func
    return (
        isinstance(func, ast.Attribute)
        and func.attr == "register"
        and isinstance(func.value, ast.Name)
        and func.value.id == "registry"
    )

def _module_registers_tools(module_path: Path) -> bool:
    """只检查模块顶层语句，避免把函数内的 register 调用误判为注册"""
    source = module_path.read_text(encoding="utf-8")
    tree = ast.parse(source, filename=str(module_path))
    return any(_is_registry_register_call(stmt) for stmt in tree.body)

def discover_builtin_tools(tools_dir=None) -> List[str]:
    """扫描 tools/ 目录，AST 确认 → 动态导入 → 自动注册"""
    tools_path = Path(tools_dir) if tools_dir else Path(__file__).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 过滤
    ]
    for mod_name in module_names:
        importlib.import_module(mod_name)  # ← 导入即注册

关键设计：AST 扫描只检查模块顶层（tree.body），不进入函数内部。这意味着工具可以在函数内调用 registry.register()（如条件注册）而不会被误扫描。只有模块顶层的 registry.register() 才触发自动导入。

2. 工具自注册模式

每个工具模块在文件末尾调用 registry.register() 完成自注册。以 terminal_tool.py 为例：

📄 tools/terminal_tool.py (第 2616-2684 行)

from tools.registry import registry

TERMINAL_SCHEMA = {
    "name": "terminal",
    "description": "Execute commands on the VM...",
    "parameters": {
        "type": "object",
        "properties": {
            "command": {"type": "string", ...},
            "background": {"type": "boolean", ...},
            "timeout": {"type": "integer", ...},
            "workdir": {"type": "string", ...},
            "pty": {"type": "boolean", ...},
            "notify_on_complete": {"type": "boolean", ...},
            "watch_patterns": {"type": "array", ...},
        },
        "required": ["command"]
    }
}

def _handle_terminal(args, **kw):
    return terminal_tool(
        command=args.get("command"),
        background=args.get("background", False),
        timeout=args.get("timeout"),
        task_id=kw.get("task_id"),
        ...
    )

registry.register(
    name="terminal",
    toolset="terminal",
    schema=TERMINAL_SCHEMA,
    handler=_handle_terminal,
    check_fn=check_terminal_requirements,
    emoji="💻",
    max_result_size_chars=100_000,
)

注册签名完整参数：

参数	类型	说明
`name`	str	唯一标识，LLM 调用时使用
`toolset`	str	所属工具集名称
`schema`	dict	JSON Schema 定义
`handler`	Callable	处理函数，接收 args dict
`check_fn`	Callable\|None	可用性探针（详见下文）
`requires_env`	list	所需环境变量列表
`is_async`	bool	是否异步处理
`emoji`	str	UI 显示图标
`max_result_size_chars`	int\|None	结果截断阈值
`dynamic_schema_overrides`	Callable\|None	运行时动态 schema 覆盖
`override`	bool	允许覆盖已有工具（插件用）

3. ToolEntry — 工具元数据容器

ToolEntry 使用 __slots__ 减少内存占用——每个工具一个实例，长生命周期下值得优化：

📄 tools/registry.py (第 77-106 行)

class ToolEntry:
    __slots__ = (
        "name", "toolset", "schema", "handler", "check_fn",
        "requires_env", "is_async", "description", "emoji",
        "max_result_size_chars", "dynamic_schema_overrides",
    )

    def __init__(self, name, toolset, schema, handler, check_fn,
                 requires_env, is_async, description, emoji,
                 max_result_size_chars=None, dynamic_schema_overrides=None):
        ...
        # dynamic_schema_overrides: 零参数 callable，返回 dict 覆盖 schema
        # 用于运行时配置依赖字段（如 delegate_task 的描述需反映
        # 当前 delegation.max_concurrent_children / max_spawn_depth）

四、ToolRegistry 核心机制

1. 线程安全与快照隔离

MCP 动态刷新会在运行时修改注册表，而对话循环同时在读取工具元数据。Hermes 用 RLock + 快照模式解决：

📄 tools/registry.py (第 151-180 行)

class ToolRegistry:
    def __init__(self):
        self._tools: Dict[str, ToolEntry] = {}
        self._toolset_checks: Dict[str, Callable] = {}
        self._toolset_aliases: Dict[str, str] = {}
        self._lock = threading.RLock()          # ← 读写都加锁
        self._generation: int = 0               # ← 变更版本号

    def _snapshot_state(self):
        """返回一致的快照 — 锁内复制，锁外消费"""
        with self._lock:
            return list(self._tools.values()), dict(self._toolset_checks)

    def _snapshot_entries(self) -> List[ToolEntry]:
        return self._snapshot_state()[0]

    def _snapshot_toolset_checks(self) -> Dict[str, Callable]:
        return self._snapshot_state()[1]

设计要点：_generation 计数器在每次注册/注销/别名变更时递增。编排层的 get_tool_definitions() 将 generation 纳入缓存键——注册表一变，缓存自动失效，无需显式通知。

2. 注册冲突保护

防止插件/MCP 意外覆盖内置工具：

📄 tools/registry.py (第 234-305 行)

def register(self, name, toolset, schema, handler, check_fn=None, ...):
    with self._lock:
        existing = self._tools.get(name)
        if existing and existing.toolset != toolset:
            # MCP ↔ MCP 覆盖允许（服务器刷新/重叠工具名）
            both_mcp = existing.toolset.startswith("mcp-") and toolset.startswith("mcp-")
            if both_mcp:
                logger.debug("MCP overwrite: %s → %s", name, toolset)
            elif override:
                # 显式 opt-in：插件声明要替换内置工具
                logger.info("Tool '%s': override by '%s'", name, toolset)
            else:
                # 拒绝影子覆盖 — 防止意外替换
                logger.error("REJECTED: '%s' shadows existing tool", name)
                return  # ← 静默丢弃，不崩溃
        self._tools[name] = ToolEntry(...)
        self._generation += 1

三层保护策略：MCP 内部允许覆盖（刷新场景）→ 显式 override=True 允许替换 → 其他全部拒绝并记录 ERROR。插件开发者必须显式声明意图。

3. check_fn TTL 缓存 — 外部探针优化

check_fn 探针检测外部状态（Docker 守护进程、Modal SDK、Playwright 等）。长生命周期进程中每轮调用是纯浪费——Hermes 用 30 秒 TTL 缓存：

📄 tools/registry.py (第 109-148 行)

_CHECK_FN_TTL_SECONDS = 30.0
_check_fn_cache: Dict[Callable, tuple[float, bool]] = {}
_check_fn_cache_lock = threading.Lock()

def _check_fn_cached(fn: Callable) -> bool:
    """TTL 缓存 check_fn 结果，异常吞掉返回 False"""
    now = time.monotonic()
    with _check_fn_cache_lock:
        cached = _check_fn_cache.get(fn)
        if cached is not None:
            ts, value = cached
            if now - ts < _CHECK_FN_TTL_SECONDS:
                return value  # ← 缓存命中
    try:
        value = bool(fn())
    except Exception:
        value = False         # ← 异常 → 工具不可用
    with _check_fn_cache_lock:
        _check_fn_cache[fn] = (now, value)
    return value

def invalidate_check_fn_cache() -> None:
    """配置变更后调用（如 hermes tools enable foo）"""
    with _check_fn_cache_lock:
        _check_fn_cache.clear()

双层缓存架构：

check_fn 缓存 (双层设计)

第 1 层: _check_fn_cached() — 30s TTL
  • 键: check_fn 函数对象
  • 值: (timestamp, bool)
  • 目的: 跨轮次缓存，减少 Docker/Playwright 探测

第 2 层: get_definitions() 内的 check_results dict
  • 键: check_fn 函数对象
  • 值: bool (单次调用内)
  • 目的: 单次 get_definitions() 内重复探针去重
  • 寿命: 函数返回即丢弃

4. dispatch() — 统一分发入口

所有工具调用最终汇聚到 registry.dispatch()：

📄 tools/registry.py (第 390-416 行)

def dispatch(self, name: str, args: dict, **kwargs) -> str:
    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})

错误处理链路：异常 → 日志记录 → _sanitize_tool_error() 剥离 XML 标签/CDATA/代码围栏 → JSON 序列化返回。防止工具错误文本中的结构化标记混淆 LLM 的后续解析。

五、ToolSet 系统：工具分组与组合

1. TOOLSETS 字典 — 60+ 预定义工具集

toolsets.py 定义了一个庞大的工具集目录：

📄 toolsets.py (第 31-581 行)

# 核心工具白名单 (~55 个工具)
_HERMES_CORE_TOOLS = [
    "web_search", "web_extract",
    "terminal", "process", "read_terminal",
    "read_file", "write_file", "patch", "search_files",
    "vision_analyze", "image_generate",
    "skills_list", "skill_view", "skill_manage",
    "browser_navigate", "browser_snapshot", "browser_click", ...
    "todo", "memory", "session_search", "clarify",
    "execute_code", "delegate_task", "cronjob", "send_message",
    "ha_list_entities", "ha_get_state", ...
    "kanban_show", "kanban_list", ...
    "computer_use",
]

Webhook 安全工具集 (最小权限)

_HERMES_WEBHOOK_SAFE_TOOLS = [
    "web_search", "web_extract", "vision_analyze", "clarify",
]

TOOLSETS = {
    "web": {"tools": ["web_search", "web_extract"], "includes": []},
    "terminal": {"tools": ["terminal", "process"], "includes": []},
    "browser": {"tools": ["browser_navigate", ...], "includes": []},
    ...
    "debugging": {
        "tools": ["terminal", "process"],
        "includes": ["web", "file"],  # ← 组合其他工具集
    },
    "hermes-cli": {"tools": _HERMES_CORE_TOOLS, "includes": []},
    "hermes-gateway": {
        "tools": [],
        "includes": ["hermes-telegram", "hermes-discord", ...],
    },
}

2. resolve_toolset() — 递归解析 + 循环检测

工具集可以包含其他工具集，形成 DAG。解析时需要处理菱形依赖和循环：

📄 toolsets.py (第 636-707 行)

def resolve_toolset(name: str, visited: Set[str] = None) -> List[str]:
    if visited is None:
        visited = set()

    # 特殊别名: "all" / "*" → 解析所有工具集
    if name in {"all", "*"}:
        all_tools: Set[str] = set()
        for toolset_name in get_toolset_names():
            resolved = resolve_toolset(toolset_name, visited.copy())
            all_tools.update(resolved)
        return sorted(all_tools)

    # 循环检测 / 菱形依赖
    if name in visited:
        return []  # ← 安全退出，不报错

    visited.add(name)

    toolset = get_toolset(name)
    if not toolset:
        # 自动为插件平台生成工具集 (hermes-<name>)
        if name.startswith("hermes-"):
            platform_name = name[len("hermes-"):]
            if platform_registry.is_registered(platform_name):
                return list(set(_HERMES_CORE_TOOLS) | plugin_tools)
        return []

    # 收集直接工具 + 递归解析 includes
    tools = set(toolset.get("tools", []))
    for included_name in toolset.get("includes", []):
        included_tools = resolve_toolset(included_name, visited)
        tools.update(included_tools)
    return sorted(tools)

自动插件工具集：当 hermes-<platform> 不在 TOOLSETS 中但平台已注册时，自动组合 _HERMES_CORE_TOOLS + 插件注册的工具。这让新平台无需修改 core 代码即可获得完整工具集。

3. 工具集拓扑结构

ToolSet 组合拓扑 (部分)

hermes-gateway  ← 组合所有平台工具集
  ├── hermes-telegram  ← _HERMES_CORE_TOOLS
  ├── hermes-discord   ← _HERMES_CORE_TOOLS + discord + discord_admin
  ├── hermes-slack     ← _HERMES_CORE_TOOLS
  ├── hermes-feishu    ← _HERMES_CORE_TOOLS + feishu_doc + feishu_drive
  ├── hermes-webhook   ← _HERMES_WEBHOOK_SAFE_TOOLS (最小权限!)
  └── ... (18 个平台)

debugging  ← 组合场景工具集
  ├── terminal (terminal, process)
  ├── web (web_search, web_extract)
  └── file (read_file, write_file, patch, search_files)

coding  ← 姿态工具集 (posture=True)
  └── 自动在代码仓库中选中，不含 messaging/tts/image_gen/spotify
  └── 不会被自动恢复到平台配置 (non-configurable)

六、model_tools.py：编排层深度解析

1. get_tool_definitions() — 三级缓存

这是对话循环每轮调用的热路径，Hermes 用三级缓存优化：

📄 model_tools.py (第 243-347 行)

# 缓存键: (enabled_toolsets, disabled_toolsets, generation, config_fp, kanban, skip_assembly)
_tool_defs_cache: Dict[tuple, List[Dict[str, Any]]] = {}
_TOOL_DEFS_CACHE_MAX = 8  # LRU 上限

def get_tool_definitions(enabled_toolsets, disabled_toolsets,
                         quiet_mode=False, skip_tool_search_assembly=False):
    if quiet_mode:
        # 配置指纹: mtime_ns + size (config 编辑自动失效缓存)
        cfg_fp = (cfg_path.stat().st_mtime_ns, cfg_path.stat().st_size)
        cache_key = (
            frozenset(enabled_toolsets),
            frozenset(disabled_toolsets),
            registry._generation,     # ← 注册表变更感知
            cfg_fp,                   # ← 配置文件变更感知
            bool(os.environ.get("HERMES_KANBAN_TASK")),
            bool(skip_tool_search_assembly),
        )
        cached = _tool_defs_cache.get(cache_key)
        if cached is not None:
            return list(cached)  # ← 浅拷贝，防止下游污染

    result = _compute_tool_definitions(...)
    # LRU 淘汰 + 缓存
    if len(_tool_defs_cache) >= _TOOL_DEFS_CACHE_MAX:
        _tool_defs_cache.pop(next(iter(_tool_defs_cache)))
    _tool_defs_cache[cache_key] = result
    return list(result)

get_tool_definitions() 三级缓存

第 1 层: _tool_defs_cache (LRU, 最多 8 条)
  • 键: (toolsets, generation, config_fp, kanban)
  • 命中: ~7ms 节省/次
  • 失效: 注册表变更 / config 编辑 / kanban 切换
  • 仅 quiet_mode=True 时生效 (stdout 副作用路径不缓存)

第 2 层: registry.get_definitions() 内 check_results dict
  • 单次调用内 check_fn 去重
  • 寿命: 函数返回即丢弃

第 3 层: _check_fn_cached() (30s TTL)
  • 跨调用缓存外部探针结果
  • 键: check_fn 函数对象

2. 动态 Schema 注入

某些工具的 schema 依赖于运行时状态。Hermes 在 _compute_tool_definitions() 中动态注入：

📄 model_tools.py (第 426-482 行)

# execute_code: 重建 schema，只列出实际可用的沙箱工具
if "execute_code" in available_tool_names:
    sandbox_enabled = SANDBOX_ALLOWED_TOOLS & available_tool_names
    dynamic_schema = build_execute_code_schema(sandbox_enabled, mode=...)
    # 替换静态 schema
    for i, td in enumerate(filtered_tools):
        if td["function"]["name"] == "execute_code":
            filtered_tools[i] = {"type": "function", "function": dynamic_schema}

discord: 根据 bot 权限重建 schema

for discord_tool_name in {"discord": "get_dynamic_schema_core", ...}:
    dynamic = schema_fn()
    if dynamic is None:
        filtered_tools = [t for t in filtered_tools
                         if t["function"]["name"] != discord_tool_name]
    else:
        # 替换为动态 schema

browser_navigate: 剥离 web 工具交叉引用

if not {"web_search", "web_extract"} & available_tool_names:
    desc = desc.replace("prefer web_search or web_extract", "")

设计意图：模型永远只看到实际可用的工具。如果 web_search 因 API key 缺失被 check_fn 过滤，browser_navigate 的描述中就不会提到它——防止模型幻觉调用不存在的工具。

3. handle_function_call() — 完整分发流水线

工具调用的完整处理链路：

handle_function_call() 分发流水线 (model_tools.py L876-1200)

function_name + function_args
  ↓
┌─ 参数类型修正 (coerce_tool_args)
│   "42" → 42, "true" → True, 裸字符串 → [字符串]
│   基于 JSON Schema 声明的类型
  ↓
├─ Tool Search 桥接分发 (tool_search/tool_describe/tool_call)
│   桥接工具内联处理，tool_call 解包为真实工具名
│   ↓
├─ Agent 循环工具拦截 (_AGENT_LOOP_TOOLS)
│   todo/memory/session_search/delegate_task → 返回错误提示
│   ↓
├─ pre_tool_call 插件钩子 (可阻断)
│   返回 block_message → 立即返回错误 + post_tool_call 钩子
│   ↓
├─ ACP 编辑审批 (write_file/patch 前置检查)
│   ↓
├─ 非读/搜工具通知 (重置连续读取计数器)
│   ↓
├─ 分发计时开始 (monotonic clock)
│   ↓
│   ┌─ 审批上下文设置 (tools.approval)
│   │   ↓
│   │   ┌─ tool_request 中间件 (参数重写)
│   │   │   ↓
│   │   │   ┌─ registry.dispatch()
│   │   │   │   → handler(args, **kwargs)
│   │   │   │   → async: _run_async(handler(...))
│   │   │   │   → error: _sanitize_tool_error()
│   │   │   └─ tool_execution 中间件 (结果转换)
│   │   └─ 审批上下文清理
│   ↓
├─ 分发计时结束 (duration_ms)
│   ↓
├─ post_tool_call 插件钩子 (观测)
│   ↓
└─ transform_tool_result 插件钩子 (可修改结果)
    ↓
  返回 result (JSON string)

4. _run_async() — 线程安全的 async 桥接

工具处理器可能是 async 的，但调用方可能是同步的。Hermes 实现了完整的桥接：

📄 model_tools.py (第 47-173 行)

def _run_async(coro):
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = None

    if loop and loop.is_running():
        # 已在 async 上下文中 (gateway/RL env)
        # → 启动独立线程 + 独立 event loop
        # → 300s 超时 + 主动 cancel pending tasks
        pool = ThreadPoolExecutor(max_workers=1)
        future = pool.submit(_run_in_worker)
        return future.result(timeout=300)

    # Worker 线程 (delegate_task 并行执行)
    if threading.current_thread() is not threading.main_thread():
        worker_loop = _get_worker_loop()  # 线程局部持久 loop
        return worker_loop.run_until_complete(coro)

    # 主线程 (CLI)
    tool_loop = _get_tool_loop()  # 全局持久 loop
    return tool_loop.run_until_complete(coro)

三种路径：

场景	策略	原因
Gateway (已有运行中的 loop)	新线程 + 新 loop	不能阻塞事件循环
Worker 线程 (并行工具执行)	线程局部持久 loop	避免与主线程竞争
CLI (无运行中的 loop)	全局持久 loop	复用 httpx/AsyncOpenAI 缓存

关键洞察：持久化 loop 而非每次 asyncio.run()，是因为 httpx/AsyncOpenAI 客户端在 GC 时会尝试关闭 transport——如果 loop 已关闭，触发 RuntimeError: Event loop is closed。持久 loop 让缓存客户端保持有效。

5. coerce_tool_args() — LLM 参数类型修正

LLM 经常返回错误类型的参数（字符串代替数字/布尔值）。Hermes 在分发前自动修正：

📄 model_tools.py (第 619-700 行)

def coerce_tool_args(tool_name: str, args: Dict) -> Dict:
    schema = registry.get_schema(tool_name)
    properties = (schema.get("parameters") or {}).get("properties")

    for key, value in list(args.items()):
        prop_schema = properties.get(key)
        expected = prop_schema.get("type")

        # array: 裸值包装为单元素列表
        if expected == "array" and not isinstance(value, (list, tuple)):
            args[key] = [value]  # {"urls": "http://x.com"} → {"urls": ["http://x.com"]}
            continue

        # 类型修正 (仅当值是字符串时)
        if isinstance(value, str):
            if expected in {"integer", "number"}:
                args[key] = _coerce_number(value, ...)  # "42" → 42
            elif expected == "boolean":
                args[key] = _coerce_boolean(value)      # "true" → True
            elif expected == "null" and value == "null":
                args[key] = None
            # union type: 逐个尝试修正

七、Tool Search：渐进式工具披露

当 MCP/插件工具过多时，全量 schema 会占用大量上下文窗口。Hermes 的 Tool Search 机制将非核心工具"折叠"为三个桥接工具：

📄 tools/tool_search.py (第 1-46 行)

TOOL_SEARCH_NAME = "tool_search"
TOOL_DESCRIBE_NAME = "tool_describe"
TOOL_CALL_NAME = "tool_call"

设计约束:

• 核心工具 (toolsets._HERMES_CORE_TOOLS) 永不延迟

• 阈值门控: 可延迟工具 < threshold_pct% 上下文窗口 → 不激活

• 目录无状态: 每次重建，不跨轮次缓存 (避免注册表漂移)

• 桥接工具路由通过 handle_function_call — 所有钩子正常触发

• 显示/轨迹解包: 用户始终看到真实工具名

Tool Search 激活流程

get_tool_definitions() 最后一步:
  ↓
┌─ 计算可延迟工具 (MCP + 插件, 不含核心工具) 的 token 数
├─ 如果 < threshold_pct (默认 10%) → 不激活, 全量返回
├─ 否则:
│   ├─ 移除可延迟工具的 schema
│   ├─ 注入 3 个桥接工具:
│   │   tool_search:  搜索可用工具 (关键词匹配)
│   │   tool_describe: 查看工具详细 schema
│   │   tool_call:    通过桥接调用底层工具
│   └─ 桥接工具路由回 handle_function_call
│       → tool_call 解包为真实工具名
│       → 所有中间件/钩子对真实工具名生效
│       → 用户/轨迹显示真实工具名 (桥接透明)

八、工具系统完整数据流

工具系统完整数据流

启动时:
  tools/registry.py
    discover_builtin_tools()
      → AST 扫描 tools/*.py
      → importlib.import_module()
      → 每个模块顶层 registry.register() 触发
      → ToolRegistry._tools 填充 ~55 个工具

  model_tools.py 模块加载
    discover_builtin_tools()  ← 再次触发 (幂等)
    discover_plugins()        ← 插件工具注册
    TOOL_TO_TOOLSET_MAP = registry.get_tool_to_toolset_map()
    TOOLSET_REQUIREMENTS = registry.get_toolset_requirements()

每轮对话:
  AIAgent.__init__()
    → get_tool_definitions(enabled_toolsets, disabled_toolsets)
      → resolve_toolset() 展开工具集
      → registry.get_definitions() 过滤 + check_fn 探针
      → 动态 schema 注入 (execute_code, discord, browser_navigate)
      → schema 清洗 (llama.cpp 兼容)
      → Tool Search 组装 (如触发)
      → 缓存 (LRU 8 条, 6 维键)
    → self.tools = result

  对话循环 (conversation_loop.py):
    → LLM 返回 tool_calls
    → handle_function_call(name, args, ...)
      → coerce_tool_args() 类型修正
      → pre_tool_call 钩子
      → 中间件链
      → registry.dispatch() → handler()
      → post_tool_call 钩子
      → transform_tool_result 钩子
    → 结果追加到 messages
    → 回到 LLM API 调用

九、关键设计模式总结

工具系统设计模式

1. 自注册模式 (Self-Registration)
   每个工具模块在顶层调用 registry.register()
   AST 扫描自动发现 → 零配置

2. 注册中心单例 (Registry Singleton)
   全局 ToolRegistry 实例, 线程安全 (RLock)
   _generation 计数器实现缓存自动失效

3. 快照隔离 (Snapshot Isolation)
   读操作返回副本, 写操作加锁
   MCP 动态刷新不影响正在运行的对话

4. 双层缓存 (Two-Level Caching)
   30s TTL (跨轮次) + 单次调用内去重
   check_fn 探针结果不重复探测

5. 防御性覆盖 (Defensive Override)
   三层: MCP 内部允许 → override=True → 拒绝
   防止插件意外替换内置工具

6. 渐进式披露 (Progressive Disclosure)
   Tool Search 桥接: 核心工具全量, 非核心按需加载
   阈值门控: <10% 上下文窗口时不激活

7. 错误卫生 (Error Hygiene)
   _sanitize_tool_error() 剥离结构化标记
   防止工具错误文本注入混淆 LLM

8. 类型修正 (Type Coercion)
   coerce_tool_args() 自动修正 LLM 参数类型
   基于 JSON Schema, 失败时保留原值

十、下一讲预告

第 4 讲：插件系统 — PluginContext API、钩子系统 (pre_llm_call / pre_tool_call / post_tool_call / transform_tool_result)、平台适配器模式。我们将分析 hermes_cli/plugins.py 和 gateway/platforms/base.py，理解 Hermes 如何通过插件扩展能力而不膨胀核心代码。

📚 系列导航: 第 1 讲 · 第 2 讲 · 第 3 讲 (本篇) · 第 4-10 讲 (待续)

源码仓库: github.com/NousResearch/hermes-agent