本文基于 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

为什么 dbcurrent_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() 的智能截断策略值得一提——它不是简单从头截断,而是:

  1. 先找精确短语匹配位置
  2. 没有则找所有关键词 200 字符窗口内的共现位置
  3. 再没有则找单个词出现位置
  4. 选择覆盖匹配点最多的 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 源码分析,如有疏漏欢迎指正。

Logo

葡萄城是专业的软件开发技术和低代码平台提供商,聚焦软件开发技术,以“赋能开发者”为使命,致力于通过表格控件、低代码和BI等各类软件开发工具和服务,一站式满足开发者需求,帮助企业提升开发效率并创新开发模式。

更多推荐