乐于分享
好东西不私藏

Hermes Agent 源码解析(四):工具系统 —— 自注册、热插拔、按需加载

Hermes Agent 源码解析(四):工具系统 —— 自注册、热插拔、按需加载

工具系统是 Hermes Agent 最精巧的架构模块之一。它解决了三个问题:

  1. 1. 70+ 工具如何管理?
  2. 2. 新工具如何零配置加入?
  3. 3. 不在使用的工具如何不浪费资源?

三层架构

tools/registry.py        ← 注册中心:存储所有工具的 schema + handler + metadata       ↑tools/*.py               ← 每个工具独立文件,模块顶层调用 registry.register()       ↑model_tools.py           ← 编排层:触发自动发现,提供查询/派发接口

第一层:tools/registry.py

这是一个无外部依赖的纯注册中心(589 行)。核心数据结构:

class_Registry:def__init__(self):self._tools: Dict[str, ToolDef] = {}    # name → ToolDefself._toolset_map: Dict[strstr] = {}  # name → toolsetdefregister(self, name, toolset, schema, handler,                 check_fn=None, requires_env=None):        ...defdispatch(self, name, args, task_id=None) -> str:        ...

ToolDef 包含:

  • • name — 工具名(如 read_fileweb_search
  • • toolset — 归属工具集(如 fileweb
  • • schema — OpenAI function calling schema
  • • handler — 执行函数(必须返回 JSON 字符串)
  • • check_fn — 可用性检查(如 API Key 是否存在)
  • • requires_env — 需要的环境变量列表

AST 扫描自动发现

发现机制非常聪明——不是通过 import 列表,而是扫描文件系统

def_module_registers_tools(module_path: Path) -> bool:"""通过 AST 判断模块是否包含顶层 registry.register() 调用"""    source = module_path.read_text(encoding="utf-8")    tree = ast.parse(source)for node in tree.body:if _is_registry_register_call(node):returnTruereturnFalse

这意味着:在任何 tools/*.py 中加一个 registry.register() 调用,系统自动发现。不需要修改任何 import 列表。

第二层:tools/*.py —— 每个工具独立文件

一个典型工具文件的结构:

# tools/delegate_tool.pyimport jsonfrom tools.registry import registrydefcheck_requirements() -> bool:returnbool(os.getenv("SOME_API_KEY"))defmy_handler(param: str, task_id: str = None) -> str:    result = do_something(param)return json.dumps({"success"True"data": result})registry.register(    name="delegate_task",    toolset="delegation",    schema={"name""delegate_task","description""委派子任务给子 Agent","parameters": {"type""object","properties": {"goal": {"type""string"},"context": {"type""string"},            },        },    },    handler=lambda args, **kw: my_handler(        param=args.get("goal"""), task_id=kw.get("task_id")    ),    check_fn=check_requirements,    requires_env=["SOME_API_KEY"],)

三个关键约定

  1. 1. Handler 必须返回 JSON 字符串 —— 统一格式让上下游解析逻辑一致
  2. 2. check_fn 可选 —— 用于 API Key 检查、平台依赖检查
  3. 3. requires_env 可选 —— 声明所需环境变量,供 doctor 命令诊断

第三层:model_tools.py —— 编排层

923 行的编排层,提供核心 API:

# 获取当前会话的工具定义(已按工具集过滤)get_tool_definitions(enabled_toolsets, disabled_toolsets) → list[schema]# 派发函数调用handle_function_call(function_name, function_args, task_id) → str# 查询工具信息get_all_tool_names() → list[str]get_toolset_for_tool(name) → strcheck_tool_availability(quiet) → (setset)  # (可用, 不可用)

Sync → Async 桥接

工具 handler 可以是同步或异步。model_tools.py 的 _run_async() 函数是单点真相

_run_async(coro)  ├── 检查当前线程是否有 running loop  ├── 有 → 启新线程隔离执行(gateway 等异步环境)  ├── 无 → 用持久化 event loop(CLI 路径)  └── 持久化 loop 防止 "Event loop is closed" 错误

这个持久化 loop 的设计原因是:asyncio.run() 会在执行完后关闭 loop,但缓存的 httpx / AsyncOpenAI 客户端还绑定在已关闭的 loop 上,触发 GC 时会崩。

懒加载依赖

tools/lazy_deps.py 提供按需包安装:

# 用户选了某个 provider 才安装对应的包# 避免 pyproject.toml 中 dependencies 过于臃肿# 示例:anthropic 包不在核心依赖中,选了 Anthropic provider 才装

核心依赖只保留每个会话都必需的包(openai、httpx、pyyaml 等),其余进入 extras 或懒加载。这是供应链安全策略的一部分 —— 更小的 dependencies = 更小的攻击面。


欢迎一起讨论! AST 扫描自动发现这个方案你觉得聪明还是过度设计?你项目里用过类似的技巧吗?📢 关注我,第一时间收到更新推送。

下一篇讲这些工具如何按功能集分组 —— Toolsets 系统。