深度源码解析:Hermes Agent 工具体系的注册、加载与执行机制
本文基于 hermes-agent 开源项目源码,深入剖析一个工业级 AI Agent 的工具系统是如何从零构建的。涵盖工具注册、自动发现、按需加载、运行时分发、Agent Loop 拦截等核心机制,并对比原生 Function Calling 的局限,阐述 Handler 存在的必要性。
一、从一个问题开始
当你使用 OpenAI 的 Function Calling 时,只需要定义一个 JSON Schema:
{
"name": "get_weather",
"description": "Get the current weather",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "City name"}
},
"required": ["city"]
}
}
LLM 看到这个 Schema,输出 tool_call: {name: "get_weather", arguments: {city: "Beijing"}},然后你自己在代码里处理执行逻辑。
这套机制够用了——当你的工具只有 3~5 个的时候。
但 Hermes Agent 有 30+ 个内置工具、支持 MCP 服务器动态扩展、还要跑在 CLI / Telegram / Discord / Slack 等多种平台上。如果每加一个工具就去改主循环的分发代码,系统会迅速腐化。
Hermes 怎么解决的?答案是一个 自注册 + 自动发现 + 分层分发 的工具体系。
二、整体架构:三层分离
┌─────────────────────────────────────────────────────┐
│ Schema 层 │
│ LLM 看到的 function 定义(JSON Schema) │
│ 决定 LLM 知道什么工具可用、怎么调用 │
├─────────────────────────────────────────────────────┤
│ Registry 层 │
│ 工具注册中心(ToolRegistry 单例) │
│ 存储 schema + handler + check_fn 的绑定关系 │
├─────────────────────────────────────────────────────┤
│ Handler 层 │
│ 实际执行逻辑(Python 函数) │
│ 接收 LLM 参数 + 运行时注入参数,返回执行结果 │
└─────────────────────────────────────────────────────┘
Schema 给 LLM 看,Handler 给框架用,Registry 做桥接。 这就是核心设计。
三、工具注册:模块导入时自动触发
3.1 注册入口
每个工具文件在底部调用 registry.register(),以 tools/session_search_tool.py 为例:
# tools/session_search_tool.py 文件底部
from tools.registry import registry, tool_error
registry.register(
name="session_search",
toolset="session_search",
schema=SESSION_SEARCH_SCHEMA, # LLM 看到的 JSON Schema
handler=lambda args, **kw: session_search(
query=args.get("query") or "",
role_filter=args.get("role_filter"),
limit=args.get("limit", 3),
db=kw.get("db"), # 运行时注入,不在 schema 里
current_session_id=kw.get("current_session_id"), # 同上
),
check_fn=check_session_search_requirements, # 可用性检查
emoji="🔍",
)
这段代码在模块级别执行——Python 导入这个文件时,registry.register() 就会运行,把工具信息注册到全局 Registry 中。
3.2 自动发现:AST 扫描
Hermes 不需要维护一个手工 import 列表。discover_builtin_tools() 用 AST 分析自动发现哪些文件包含工具注册:
# tools/registry.py
def discover_builtin_tools(tools_dir=None):
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 扫描,检查是否有 registry.register() 调用
]
for mod_name in module_names:
importlib.import_module(mod_name) # ← 导入触发注册
return imported
_module_registers_tools() 用 AST 解析源文件,检查模块顶层是否有 registry.register(...) 调用:
def _is_registry_register_call(node):
"""检查 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"
)
为什么要用 AST 而不是直接 import 再 try/except? 因为有些辅助模块可能也包含 registry.register() 调用但不是独立工具(比如在函数内部调用)。AST 只检查模块顶层语句,更精确。
3.3 完整发现链
model_tools.py 被导入
│
├─ discover_builtin_tools() ← 扫描 tools/*.py,触发 registry.register()
├─ discover_mcp_tools() ← 加载外部 MCP 服务器工具
└─ discover_plugins() ← 加载 user/project/pip 插件工具
新增工具只需要:1) 在 tools/ 下新建 .py 文件;2) 在文件内调用 registry.register()。 框架自动发现,无需修改任何其他文件。
四、按需加载:Toolset 过滤机制
4.1 Toolset 定义
不是所有工具都适合所有平台。Hermes 用 Toolset(工具集)概念做粗粒度分组:
# toolsets.py
_HERMES_CORE_TOOLS = [
"web_search", "web_extract",
"terminal", "process",
"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",
]
TOOLSETS = {
"hermes-cli": {"tools": _HERMES_CORE_TOOLS, "includes": []},
"hermes-telegram": {"tools": _HERMES_CORE_TOOLS, "includes": []},
"hermes-acp": {"tools": [...], "includes": []}, # 编辑器集成,无消息/音频工具
"safe": {"tools": [], "includes": ["web", "vision", "image_gen"]}, # 无终端
# ... 更多平台
}
Toolset 支持 组合(includes 字段),resolve_toolset() 会递归展开:
def resolve_toolset(name, visited=None):
if name in {"all", "*"}:
# 特殊别名:返回所有工具的并集
for toolset_name in get_toolset_names():
all_tools.update(resolve_toolset(toolset_name))
return sorted(all_tools)
toolset = get_toolset(name)
tools = set(toolset.get("tools", []))
for included_name in toolset.get("includes", []):
tools.update(resolve_toolset(included_name)) # 递归展开
return sorted(tools)
4.2 双重过滤:Toolset + check_fn
get_tool_definitions() 是 schema 的最终提供者,做了两层过滤:
def get_tool_definitions(enabled_toolsets, disabled_toolsets, quiet_mode):
# 第一层:Toolset 白/黑名单
if enabled_toolsets:
for ts in enabled_toolsets:
tools_to_include.update(resolve_toolset(ts))
elif disabled_toolsets:
# 全量 - 排除黑名单
...
# 第二层:check_fn 可用性检查
filtered_tools = registry.get_definitions(tools_to_include, quiet=quiet_mode)
return filtered_tools
registry.get_definitions() 内部会调用每个工具的 check_fn:
# registry.py 内部逻辑
for entry in entries:
if entry.check_fn and not entry.check_fn():
continue # check_fn 返回 False → schema 不传给 LLM
result.append({"type": "function", "function": schema_with_name})
以 session_search 为例,它的 check_fn 检查数据库是否存在:
def check_session_search_requirements() -> bool:
try:
from hermes_state import DEFAULT_DB_PATH
return DEFAULT_DB_PATH.parent.exists()
except ImportError:
return False
没有数据库 → check_fn 返回 False → schema 不传给 LLM → LLM 不会调用一个用不了的工具。
4.3 启动时确定,运行时不变
# AIAgent.__init__
self.tools = get_tool_definitions(
enabled_toolsets=enabled_toolsets,
disabled_toolsets=disabled_toolsets,
quiet_mode=self.quiet_mode,
)
self.tools 在 Agent 初始化时确定,整个 Session 生命周期不变。每次 API 调用都是:
api_call(messages, tools=self.tools) # 全量传入
五、运行时分发:两条路径
当 LLM 返回 tool_call 后,Hermes 有两条分发路径:
5.1 普通工具:Registry Dispatch
大多数工具走通用路径,registry.dispatch() 自动找到对应 handler 并执行:
# model_tools.py → handle_function_call()
result = registry.dispatch(
function_name, function_args,
task_id=task_id,
user_task=user_task,
)
# registry.py → dispatch()
def dispatch(self, name, args, **kwargs):
entry = self.get_entry(name)
if entry.is_async:
return _run_async(entry.handler(args, **kwargs))
return entry.handler(args, **kwargs)
handler 接收两类参数:
| 参数 | 来源 | 例子 |
|---|---|---|
args |
LLM 的 tool_call.arguments |
{"query": "docker", "limit": 3} |
**kwargs |
框架运行时注入 | task_id, user_task |
5.2 特殊工具:Agent Loop 拦截
部分工具需要 Agent 级别的状态对象(数据库连接、内存存储等),Registry 的通用 dispatch 管不了。这些工具被列入拦截名单:
# model_tools.py
_AGENT_LOOP_TOOLS = {"todo", "memory", "session_search", "delegate_task"}
handle_function_call() 遇到它们直接拒绝:
if function_name in _AGENT_LOOP_TOOLS:
return json.dumps({"error": f"{function_name} must be handled by the agent loop"})
真正的执行路径在 run_agent.py 的 _invoke_tool() 方法里,由 Agent Loop 直接调用业务函数并注入运行时状态:
def _invoke_tool(self, function_name, function_args, effective_task_id, tool_call_id):
if function_name == "todo":
return _todo_tool(todos=..., store=self._todo_store)
elif function_name == "session_search":
return _session_search(
query=function_args.get("query", ""),
db=self._session_db, # ← 注入 SQLite 连接
current_session_id=self.session_id, # ← 注入当前会话 ID
)
elif function_name == "memory":
return _memory_tool(action=..., store=self._memory_store)
elif function_name == "delegate_task":
return _delegate_task(goal=..., parent_agent=self) # ← 注入 Agent 自身
else:
return handle_function_call(...) # 普通工具走 Registry
为什么 db 和 current_session_id 不放进 Schema? 因为 LLM 不需要也不应该知道这些——它连你用什么数据库都不知道,怎么可能传一个 SQLite 连接对象?这些是运行时基础设施,只能由框架注入。
六、完整调用链路:以 session_search 为例
当用户说"我之前做过 Docker 相关的事"时,完整的调用链路如下:
用户: "我之前做过 Docker 相关的事"
│
▼
Agent Loop 构建 messages,调用 LLM API(携带 tools schema)
│
▼
LLM 返回: tool_call: {name: "session_search", arguments: {query: "docker"}}
│
▼
_execute_tool_calls() → _invoke_tool("session_search", {query: "docker"}, ...)
│
├─ function_name == "session_search" → 命中 Agent Loop 拦截
│
▼
直接调用 session_search_tool.session_search(
query="docker",
db=self._session_db, ← 运行时注入
current_session_id="abc123", ← 运行时注入
)
│
├─ query 非空 → 关键词搜索模式
│
▼
1. db.search_messages(query="docker") ← SQLite FTS5 全文检索
2. 按 session_id 分组,去重,排除当前会话 ← 避免搜到自己
3. db.get_messages_as_conversation(sid) ← 加载完整对话
4. _truncate_around_matches(text, "docker") ← 智能截断(10万字符窗口)
5. async_call_llm(text, "docker") ← Gemini Flash 生成摘要
6. 返回 JSON: {success, query, results: [{session_id, when, summary}]}
│
▼
Agent Loop 将 tool_result 追加到 messages,继续对话
其中 _truncate_around_matches() 的智能截断策略值得一提——它不是简单从头截断,而是:
- 先找精确短语匹配位置
- 没有则找所有关键词 200 字符窗口内的共现位置
- 再没有则找单个词出现位置
- 选择覆盖匹配点最多的 10 万字符窗口
这样确保摘要 LLM 能看到最相关的上下文片段。
七、Handler 的必要性:对比原生 Function Calling
很多人会问:原生 Function Calling 只需要 Schema 就行,Handler 有什么用?
没有 Handler 的世界
假设不用 Registry + Handler,你的 Agent Loop 得这样写:
if function_name == "session_search":
result = session_search(query=function_args.get("query", ""), db=self._session_db, ...)
elif function_name == "read_file":
result = read_file(path=function_args.get("path"))
elif function_name == "web_search":
result = web_search(query=function_args.get("query"))
elif function_name == "terminal":
result = terminal(command=function_args.get("command"), task_id=effective_task_id)
# ... 30+ 个 elif ...
每加一个工具,就要改 Agent Loop 的核心代码。 参数映射、执行逻辑、错误处理全部耦合在一个巨型函数里。
有了 Handler:工具自治
每个工具文件自己声明"我叫什么、怎么调我",Agent Loop 只需一行通用分发:
result = registry.dispatch(function_name, function_args, task_id=task_id)
新增工具不需要改框架一行代码——放个文件就能用,删个文件就消失。
三个角色的比喻
Schema = 给 LLM 看的菜单("我有什么菜")
Handler = 给框架用的厨师("这道菜怎么做")
check_fn = 门口的招牌("今天有没有这道菜")
八、Prompt Cache:缓解 Tokens 压力但未解决根因
8.1 问题的本质
工具 Schema 一旦传入,不管缓存不缓存,它都实实在在占据 Context Window:
每个工具 Schema 约 200~500 tokens(名称 + 描述 + 参数定义)
30 个工具 ≈ 6,000~15,000 tokens 常驻占用
Prompt Cache 只解决"钱"的问题,不解决"长度"的问题。 缓存命中后读取成本降低 ~75%,但 Context Window 的可用空间不会因此增加。
8.2 Hermes 的缓存策略
Hermes 对 Anthropic Claude 模型自动启用 Prompt Caching(system_and_3 策略):
# AIAgent.__init__
is_openrouter = self._is_openrouter_url()
is_claude = "claude" in self.model.lower()
is_native_anthropic = self.api_mode == "anthropic_messages"
self._use_prompt_caching = (is_openrouter and is_claude) or is_native_anthropic
在系统提示 + 最近 3 条消息上打 cache_control 标记,工具 Schema 作为系统提示的一部分被缓存:
第 1 次调用:全量 tokens 计费(写缓存)
第 2 次调用:系统提示 + 工具 Schema 从缓存读,成本 ≈ -75%
第 N 次调用:同上,只有新消息计费
此外,Hermes 还用确定性 call_id 代替随机 UUID,避免每次调用导致缓存失效:
seed = f"{fn_name}:{arguments}:{index}"
digest = hashlib.sha256(seed.encode()).hexdigest()[:12]
return f"call_{digest}"
8.3 Context Compressor 兜底
当 token 数接近上下文窗口阈值时,自动压缩历史消息:
_preflight_tokens = estimate_request_tokens_rough(messages, system_prompt, tools=self.tools)
if _preflight_tokens >= self.context_compressor.threshold_tokens:
# 自动压缩历史消息,腾出空间
但压缩的是历史消息,不是工具 Schema。 工具 Schema 是固定占用,不可压缩。
8.4 未解决的问题
Hermes 目前的架构是静态工具集 + 启动时 toolset 白名单,运行时没有动态 tool selection。在工具数量很多的场景(比如 MCP 接了几十个服务器),会明显感受到上下文压缩效应。业界可能的演进方向:
| 方案 | 思路 | 代价 |
|---|---|---|
| Tool Selection Layer | 每轮先用小模型判断需要哪些工具 | 多一次推理开销 |
| Tool Routing Agent | 专门 router agent 决定调哪个工具 | 架构复杂 |
| 外部 Tool Registry | 工具 Schema 放外部,LLM 先搜索再调用 | 实现最复杂 |
九、总结
Hermes 的工具体系可以归结为以下设计决策:
| 设计决策 | 解决的问题 | 代价 |
|---|---|---|
自注册(registry.register()) |
新增工具无需改框架代码 | 工具文件需遵守注册协议 |
| 自动发现(AST 扫描) | 无需维护 import 列表 | AST 解析有少量启动开销 |
| Toolset 分组 | 不同平台加载不同工具 | 粗粒度,运行时无法动态调整 |
| check_fn 过滤 | LLM 不会调用不可用的工具 | 需要每个工具自己实现检查逻辑 |
| Agent Loop 拦截 | 注入运行时状态(DB、Store、Agent) | 拦截名单需手工维护 |
| Handler lambda 包装 | 分离 LLM 参数和运行时参数 | 多一层间接调用 |
| Prompt Caching | 多轮对话降低输入成本 | 不减少 Context Window 占用 |
核心思想:工具自治、框架做分发、Schema 和实现分离。 这让 Hermes 在 30+ 工具、多平台、可扩展的约束下,依然保持了代码的模块化和可维护性。
本文基于 hermes-agent 源码分析,如有疏漏欢迎指正。
更多推荐

所有评论(0)