Hermes-Agent-工具系统源码全解析-微信公众号排版

Hermes Agent 工具系统源码全解析

注册表自注册、按需加载与动态 Schema 重写 [04]

[04] Hermes Agent 工具系统源码全解析：589 行的注册表如何驱动 70+ 工具自注册、28 个工具集按需加载与动态 Schema 重写

> TL;DR：Hermes Agent 的工具系统不是一个大 switch-case。tools/registry.py（589行）是一个单例注册表，每个工具在模块顶层调用 registry.register() 自注册；model_tools.py 再根据启用的工具集做过滤、缓存、动态 Schema 重写。这篇从 registry.register() 到 dispatch() 的完整链路拆一遍——自注册发现、check_fn TTL 缓存、跨工具集重名保护、动态 Schema 覆盖。

---

上篇拆了 System Prompt 组装，这篇拆 Agent 的「手和脚」——工具系统。

你可能觉得工具系统没什么好讲的：不就是一堆函数挂上去吗？但 Hermes 的工具系统有 4 个值得单独写一篇文章的设计：

1. 自注册发现——70+ 个工具文件不会在代码里显式 import，而是通过 AST 扫描自动发现 2. check_fn TTL 缓存——每个工具集的可用性检查不是每次请求都跑，而是 30 秒缓存 3. 跨工具集重名保护——MCP 插件注册的工具不能覆盖内置工具（除非主动 opt-in） 4. 动态 Schema 重写——execute_code、discord 等工具的 Schema 在运行时根据配置重写

下面从注册表源码逐层拆。

---

1. 注册表内核：ToolRegistry 单例

入口文件 tools/registry.py（589 行），最尾部创建了模块级单例：

# Module-level singleton

registry = ToolRegistry()

所有工具文件都从模块顶层导入这个 registry 并调用 registry.register()。这个单例是整个工具系统的唯一真相来源——model_tools.py、run_agent.py、cli.py 都从它这里查询，没有平行数据。

ToolRegistry 的数据结构很简单——一个 Dict[str, ToolEntry]：

class ToolEntry:

__slots__= (

"name","toolset", "schema", "handler", "check_fn",

"requires_env","is_async", "description", "emoji",

"max_result_size_chars","dynamic_schema_overrides",

)

关键字段：

图 1：复杂表格已转换为 PNG，降低公众号导入变形风险。

---

2. 自注册发现：不靠手动 import，靠 AST 扫描

传统做法是把所有工具模块写在一张 import 表里。Hermes 不做这个——它通过 AST 扫描来发现哪些模块是"工具"。

入口在 discover_builtin_tools()：

def discover_builtin_tools(tools_dir=None) -> List[str]:

tools_path= Path(tools_dir) if tools_dir is not None else Path(__file__).resolve().parent

module_names= [

f"tools.{path.stem}"

forpath in sorted(tools_path.glob("*.py"))

ifpath.name not in {"__init__.py", "registry.py", "mcp_tool.py"}

and_module_registers_tools(path)# ← AST 扫描

]

imported= []

formod_name in module_names:

try:

importlib.import_module(mod_name)

imported.append(mod_name)

exceptException as e:

logger.warning("Couldnot import tool module %s: %s", mod_name, e)

returnimported

_module_registers_tools() 通过 Python 的 ast 模块分析源码——它会真正解析语法树，检查脚本的模块级别是否存在 registry.register(...) 调用：

def _module_registers_tools(module_path: Path) -> bool:

try:

source= module_path.read_text(encoding="utf-8")

tree= ast.parse(source, filename=str(module_path))

except(OSError, SyntaxError):

returnFalse

returnany(_is_registry_register_call(stmt) for stmt in tree.body)

为什么是 AST 而不是试 import？因为 import 失败可能导致级联崩溃——某个工具依赖的库没装，import 就抛异常。AST 扫描无副作作用。

def _is_registry_register_call(node: ast.AST) -> bool:

ifnot isinstance(node, ast.Expr) or not isinstance(node.value, ast.Call):

returnFalse

func= node.value.func

return(

isinstance(func,ast.Attribute)

andfunc.attr == "register"

andisinstance(func.value, ast.Name)

andfunc.value.id == "registry"

)

这段代码检查的是：表达式是否是 registry.register(...) 的 AST 结构（Name('registry') → Attribute('register') → Call）。

> 扫描发现后，Hermes 会自动 import 匹配的模块。import 触发模块顶层的 registry.register() 调用，工具就注册了。

---

3. register()：从工具文件到注册表

每个工具文件的末尾（或模块顶层）都有一行类似这样的代码：

# 摘自 tools/browser_tool.py（示意，非真实代码）

from tools.registry import registry, tool_result

def _handle_browser_navigate(args):

...

registry.register(

name="browser_navigate",

toolset="browser",

schema={

"description":"导航到指定 URL",

"parameters":{

"type":"object",

"properties":{

"url":{"type": "string", "description": "目标 URL"}

},

"required":["url"]

}

},

handler=_handle_browser_navigate,

check_fn=_check_playwright_installed,#实际检查函数

requires_env=["PLAYWRIGHT_BROWSER_PATH"],

)

register() 的内部逻辑做了三件事：

def register(self, name, toolset, schema, handler, check_fn=None,

requires_env=None,is_async=False, description="",

emoji="",max_result_size_chars=None,

dynamic_schema_overrides=None,override=False):

withself._lock:

existing= self._tools.get(name)

ifexisting and existing.toolset != toolset:

#跨工具集重名保护

both_mcp= (

existing.toolset.startswith("mcp-")

andtoolset.startswith("mcp-")

)

ifboth_mcp:

pass#MCP 同名工具间的覆盖是合法的

elifoverride:

pass#显式 opt-in 覆盖

else:

logger.error("Toolregistration REJECTED: '%s' ...")

return#❌ 拒绝注册

#注册工具

self._tools[name]= ToolEntry(...)

ifcheck_fn and toolset not in self._toolset_checks:

self._toolset_checks[toolset]= check_fn

self._generation+= 1# ← 版本号递增，通知缓存失效

_generation 是一个单调递增的计数器。每次注册/注销/别名变更都会递增。model_tools.py 的缓存会把这个 generation 作为 cache key 的一部分——generation 不变说明注册表状态没变，可以直接返回缓存。

跨工具集重名保护的场景：

---

4. check_fn TTL 缓存：30 秒弹性

check_fn 是工具集可用性检测函数。例如 terminal 工具集的 check_fn 探测 Docker daemon、Modal SDK 是否存在；browser 工具集的 check_fn 探测 playwright 是否安装。

如果每次获取工具定义都重新跑一遍这些检测，启动时还好，但在一个长驻进程里（Gateway）就浪费了。Hermes 用了 30 秒 TTL 缓存：

_CHECK_FN_TTL_SECONDS = 30.0

_check_fn_cache: Dict[Callable, tuple[float, bool]] = {}

def _check_fn_cached(fn: Callable) -> bool:

now= time.monotonic()

with_check_fn_cache_lock:

cached= _check_fn_cache.get(fn)

ifcached is not None:

ts,value = cached

ifnow - ts < _CHECK_FN_TTL_SECONDS:

returnvalue# ← 命中缓存

try:

value= bool(fn())

exceptException:

value= False

with_check_fn_cache_lock:

_check_fn_cache[fn]= (now, value)

returnvalue

为什么 30 秒？注释里写了理由：

> 30s TTL chosen so env-var changes (hermes tools enable foo) still take effect in near-real-time without forcing a full cache flush on every call.

太短（如 5 秒）→ 每次 get_definitions() 都重新探测，浪费。太长（如 300 秒）→ 用户 hermes tools enable browser 后要等 5 分钟才能生效。30 秒是实测后的折中值。

同时还有一个单次调用的内存缓存（check_results: Dict[Callable, bool] = {}）——在单次 get_definitions() 调用内，同一个 check_fn 不会被重复执行。

---

5. get_definitions()：从注册表到模型的工具 JSON

model_tools.py 中的 get_tool_definitions() 是连接注册表和 Agent Loop 的桥梁。它的核心流程：

5.1 传入工具集列表，解析出工具名

def _compute_tool_definitions(enabled_toolsets, disabled_toolsets, ...):

tools_to_include= set()

ifenabled_toolsets is not None:

fortoolset_name in effective_enabled_toolsets:

ifvalidate_toolset(toolset_name):

resolved= resolve_toolset(toolset_name)

tools_to_include.update(resolved)

else:

#默认：加载所有工具集

forts_name in get_all_toolsets():

tools_to_include.update(resolve_toolset(ts_name))

#禁用工具集作为减法

ifdisabled_toolsets:

fortoolset_name in disabled_toolsets:

resolved= resolve_toolset(toolset_name)

tools_to_include.difference_update(resolved)

注意 kanban worker 的特殊处理：

if os.environ.get("HERMES_KANBAN_TASK") and "kanban" not in effective_enabled_toolsets:

effective_enabled_toolsets.append("kanban")

Dispatcher 启动了 kanban worker，即使 Profile 配置中没有 kanban 工具集，Worker 也必须有 kanban 工具集——否则 Worker 无法上报进度、无法完成/阻塞任务。这是强制注入，不受 Profile 配置影响。

5.2 动态 Schema 重写

工具注册时的 Schema 是静态的，但某些工具的 Schema 必须在运行时根据配置动态调整。

execute_code 是典型例子。它的 Schema 中有一个 sandbox_allowed_tools 数组，列出在 execute_code 沙箱中可用的工具。但这个列表必须实时反映实际启用的工具——如果 web_search 工具集被禁用了，沙箱中也必须不可见：

if "execute_code" in available_tool_names:

fromtools.code_execution_tool import SANDBOX_ALLOWED_TOOLS, build_execute_code_schema

sandbox_enabled= SANDBOX_ALLOWED_TOOLS & available_tool_names

dynamic_schema= build_execute_code_schema(sandbox_enabled, mode=_get_execution_mode())

fori, td in enumerate(filtered_tools):

iftd.get("function", {}).get("name") == "execute_code":

filtered_tools[i]= {"type": "function", "function": dynamic_schema}

break

discord 和 discord_admin 也有类似的动态重写——基于 Discord Bot 的实际 intents（从 GET /applications/@me 检测）隐藏不支持的操作：

_discord_schema_fns = {

"discord":"get_dynamic_schema_core",

"discord_admin":"get_dynamic_schema_admin",

}

for discord_tool_name in _discord_schema_fns:

ifdiscord_tool_name in available_tool_names:

fromtools import discord_tool as _dt

schema_fn= getattr(_dt, _discord_schema_fns[discord_tool_name])

dynamic= schema_fn()

ifdynamic:

#替换静态 Schema

除了这些硬编码的动态重写，Registry 还支持通用的 dynamic_schema_overrides 回调：

if entry.dynamic_schema_overrides is not None:

try:

overrides= entry.dynamic_schema_overrides()

ifisinstance(overrides, dict):

schema_with_name.update(overrides)

exceptException as exc:

logger.warning("dynamic_schema_overridesfor tool %s raised %s", name, exc)

这个回调在 get_definitions() 每次调用时执行，返回的 dict 会和静态 Schema 做浅合并。delegate_task 就利用这个机制——它的 max_concurrent_children 和 max_spawn_depth 参数描述必须反映当前配置值：

# tools/delegate_task.py（示意）

registry.register(

name="delegate_task",

toolset="agent",

dynamic_schema_overrides=_current_delegation_config,#运行时回调

...

)

5.3 缓存策略

整个 get_tool_definitions() 支持多层缓存：

# 最外层：model_tools 的内存缓存（以 registry generation + 配置 mtime 为 key）

if quiet_mode:

cache_key= (

frozenset(enabled_toolsets),

frozenset(disabled_toolsets),

registry._generation,#注册表版本

(cfg_path.stat().st_mtime_ns,cfg_path.stat().st_size),# 配置文件指纹

bool(os.environ.get("HERMES_KANBAN_TASK")),

)

cached= _tool_defs_cache.get(cache_key)

ifcached is not None:

returnlist(cached)

# 次外层：registry.get_definitions() 内部的 check_fn TTL 缓存

filtered_tools = registry.get_definitions(tools_to_include, quiet=quiet_mode)

两个缓存层级互不覆盖——外层按配置+版本缓存整个结果列表，内层只缓存 check_fn 的执行结果。

---

6. dispatch()：从模型调用到工具执行

当模型返回 tool_calls 时，Agent Loop 调用 registry.dispatch()：

def dispatch(self, name: str, args: dict) -> str:

entry= self.get_entry(name)

ifnot entry:

returnjson.dumps({"error": f"Unknown tool: {name}"})

try:

ifentry.is_async:

frommodel_tools import _run_async

return_run_async(entry.handler(args))

returnentry.handler(args)

exceptException as e:

logger.exception("Tool%s dispatch error: %s", name, e)

raw= f"Tool execution failed: {type(e).__name__}: {e}"

try:

frommodel_tools import _sanitize_tool_error

sanitized= _sanitize_tool_error(raw)

exceptException:

sanitized= raw

returnjson.dumps({"error": sanitized})

异常处理的细节：错误信息通过 _sanitize_tool_error() 过滤——防止 framing token、CDATA、反引号等结构性噪音污染模型的下一次推理。

每个工具的处理函数必须返回 JSON 字符串。注册表提供了两个通用辅助函数：

def tool_error(message, **extra) -> str:

result= {"error": str(message)}

ifextra:

result.update(extra)

returnjson.dumps(result, ensure_ascii=False)

def tool_result(data=None, **kwargs) -> str:

ifdata is not None:

returnjson.dumps(data, ensure_ascii=False)

returnjson.dumps(kwargs, ensure_ascii=False)

使用方式：

return tool_result(success=True, count=42)

# → '{"success": true, "count": 42}'

return tool_error("file not found", code=404)

# → '{"error": "file not found", "code": 404}'

---

7. 工具集模型：从注册到解析的完整链路

把整条链路串起来：

工具文件（tools/*.py）# 每个工具独立文件

│模块顶层 registry.register()

▼

ToolRegistry（tools/registry.py）# 单例注册表

│get_definitions() → 按 check_fn 过滤 → 缓存

▼

model_tools.get_tool_definitions()# 工具集解析 + 动态 Schema 重写

│Agent Loop 获取 → 注入 API 请求

▼

LLM 返回 tool_calls

│

▼

Agent Loop → registry.dispatch(name, args)# 执行工具

│异常 → _sanitize_tool_error() → 返回 JSON

▼

工具结果回填对话历史，进入下一轮迭代

---

8. 这篇文章的代码索引

图 2：复杂表格已转换为 PNG，降低公众号导入变形风险。

已注册的工具命名来源：每个工具文件就是一个工具，文件名即工具名（在 registry.register() 的 name 参数中指定）。

---

下一篇拆记忆矩阵——MEMORY.md、Hindsight 向量库、state.db SQLite 三层架构的协同与冲突。

---

*本系列基于 Hermes Agent v0.15.2 源码。工具系统文件：tools/registry.py（589 行）+ model_tools.py（1174 行）。*