工具系统是 Hermes Agent 最精巧的架构模块之一。它解决了三个问题:
1. 70+ 工具如何管理? 2. 新工具如何零配置加入? 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[str, str] = {} # name → toolsetdefregister(self, name, toolset, schema, handler, check_fn=None, requires_env=None): ...defdispatch(self, name, args, task_id=None) -> str: ...ToolDef 包含:
• name— 工具名(如read_file、web_search)• toolset— 归属工具集(如file、web)• 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. Handler 必须返回 JSON 字符串 —— 统一格式让上下游解析逻辑一致 2. check_fn可选 —— 用于 API Key 检查、平台依赖检查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) → (set, set) # (可用, 不可用)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 系统。
夜雨聆风