第三章 工具系统
Hermes Agent 的工具系统是其核心能力的基础。通过注册中心 + 自动发现 + 编排层的三层架构,实现了灵活、可扩展的工具管理。
3.1 注册中心 registry.py
tools/registry.py 是工具系统的核心,实现了 ToolRegistry 单例模式,所有工具在模块导入时自动注册。
ToolEntry 数据结构
python
class ToolEntry:
"""注册的工具元数据"""
__slots__ = (
"name", "toolset", "schema", "handler", "check_fn",
"requires_env", "is_async", "description", "emoji",
"max_result_size_chars",
)| 字段 | 类型 | 说明 |
|---|---|---|
name | str | 工具名称(唯一标识) |
toolset | str | 所属工具集 |
schema | dict | OpenAI function calling schema |
handler | Callable | 工具执行函数 |
check_fn | Callable | 可用性检查函数(如检查环境变量) |
requires_env | list | 需要的环境变量列表 |
is_async | bool | 是否为异步工具 |
description | str | 工具描述 |
emoji | str | 显示用 emoji |
max_result_size_chars | int | 结果最大字符数限制 |
注册 API
python
class ToolRegistry:
"""Singleton registry"""
def register(
self,
name: str,
toolset: str,
schema: dict,
handler: Callable,
check_fn: Callable = None,
requires_env: list = None,
is_async: bool = False,
description: str = "",
emoji: str = "",
max_result_size_chars: int | float | None = None,
):
"""注册一个工具。每个工具文件在模块导入时调用。"""
with self._lock: # threading.RLock 保证线程安全
...线程安全设计
python
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, ToolEntry] = {}
self._lock = threading.RLock()
def _snapshot_state(self) -> tuple[List[ToolEntry], Dict[str, Callable]]:
"""返回注册表的一致性快照"""
with self._lock:
return list(self._tools.values()), dict(self._toolset_checks)注册表使用 threading.RLock(可重入锁)保护所有读写操作。读取时通过 _snapshot_state() 返回一致性快照,避免 MCP 动态刷新时的并发问题。
3.2 自动发现
Hermes Agent 的工具发现机制非常巧妙——通过 AST 扫描在不导入模块的情况下检测哪些文件包含工具注册。
discover_builtin_tools()
python
def discover_builtin_tools(tools_dir: Optional[Path] = None) -> List[str]:
"""Import built-in self-registering tool modules and return their module names."""
tools_path = Path(tools_dir) if tools_dir else 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 扫描检测
]
imported = []
for mod_name in module_names:
importlib.import_module(mod_name) # 导入触发 register()
imported.append(mod_name)
return importedAST 检测逻辑
python
def _is_registry_register_call(node: ast.AST) -> bool:
"""检测节点是否为 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 扫描不触发导入
- 安全过滤 — 排除
__init__.py、registry.py、mcp_tool.py等非工具文件
3.3 编排层 model_tools.py
model_tools.py 是工具系统的编排层,负责 schema 收集、分发调用、类型转换和异步桥接。
核心函数
| 函数 | 功能 |
|---|---|
get_tool_definitions() | 从 Registry 收集所有活跃工具的 schema |
handle_function_call() | 接收 LLM 返回的 tool_call,分发到对应 handler |
coerce_tool_args() | 将字符串参数转换为正确类型 |
_run_async() | 在同步上下文中桥接异步工具 |
| Plugin hooks | pre_tool_call / post_tool_call 钩子 |
分发流程
python
def handle_function_call(tool_name: str, tool_args: dict) -> str:
"""工具分发核心逻辑"""
entry = registry.get_entry(tool_name)
# 可用性检查
if not registry._evaluate_toolset_check(entry.toolset, entry.check_fn):
return f"Tool '{tool_name}' is not available in current environment"
# 类型转换
tool_args = coerce_tool_args(entry.schema, tool_args)
# Pre-hook
run_plugin_hook("pre_tool_call", tool_name, tool_args)
# 执行(同步/异步)
if entry.is_async:
result = _run_async(entry.handler, tool_args)
else:
result = entry.handler(**tool_args)
# 结果截断
if entry.max_result_size_chars and len(result) > entry.max_result_size_chars:
result = result[:entry.max_result_size_chars] + "\n...[truncated]"
# Post-hook
run_plugin_hook("post_tool_call", tool_name, tool_args, result)
return result3.4 工具集 toolsets.py
toolsets.py 定义了工具集的组合关系,支持递归包含——一个工具集可以包含其他工具集。
TOOLSETS 字典
python
_HERMES_CORE_TOOLS = [
"web_search", "web_extract", # Web
"terminal", "process", # Terminal
"read_file", "write_file", "patch", # File
"vision_analyze", "image_generate", # Vision
"browser_navigate", "browser_click", # Browser
"text_to_speech", # TTS
"todo", "memory", # Planning
"execute_code", "delegate_task", # Code + Delegation
"send_message", # Cross-platform
...
]
TOOLSETS = {
"web": {
"description": "Web research and content extraction tools",
"tools": ["web_search", "web_extract"],
"includes": []
},
"full_stack": {
"description": "Full stack development",
"tools": [...],
"includes": ["web", "terminal", "file"]
},
...
}递归解析
python
def resolve_toolset(name: str, visited: set = None) -> list:
"""递归解析工具集,返回所有工具名称"""
visited = visited or set()
if name in visited:
return [] # 防止循环引用
defn = TOOLSETS.get(name)
if not defn:
return []
tools = list(defn.get("tools", []))
for included in defn.get("includes", []):
tools.extend(resolve_toolset(included, visited | {name}))
return tools3.5 核心工具
terminal_tool — 终端执行
支持 6 种后端,通过配置切换执行环境:
| 后端 | 说明 | 环境变量/配置 |
|---|---|---|
local | 本地执行(默认) | 无 |
docker | Docker 容器内执行 | TERMINAL_BACKEND=docker |
modal | Modal 云端执行 | MODAL_APP_NAME |
ssh | 远程 SSH 执行 | SSH_HOST, SSH_KEY |
singularity | Singularity 容器 | SINGULARITY_IMAGE |
daytona | Daytona 开发环境 | DAYTONA_API_KEY |
file_tools — 文件操作
read_file— 读取文件内容(支持行范围)write_file— 写入文件search_files— 文件内容搜索(正则/文本)patch— 精确文件修补(diff-based)
web_tools — Web 工具
多后端搜索引擎支持:
| 后端 | 说明 |
|---|---|
firecrawl | Firecrawl API |
exa | Exa 搜索 |
parallel | 并行多源搜索 |
tavily | Tavily 搜索 |
delegate_tool — 子 Agent 委派
子 Agent 隔离执行,支持单任务和批处理模式。详见第九章。
mcp_tool — MCP 协议
支持 Model Context Protocol,可动态连接外部工具服务器。
browser_tool — 浏览器自动化
基于 CDP (Chrome DevTools Protocol) 的浏览器自动化,支持导航、点击、截图等操作。
3.6 自定义工具示例
以下是如何创建一个自定义工具的完整步骤:
步骤 1:创建工具文件
在 tools/ 目录下创建新文件 tools/my_custom_tool.py:
python
"""My Custom Tool -- 示例自定义工具"""
import json
from tools.registry import registry
def _my_handler(query: str, max_results: int = 10) -> str:
"""执行自定义操作"""
# 你的业务逻辑
results = perform_custom_operation(query, max_results)
return json.dumps(results, ensure_ascii=False)
def _check_available() -> bool:
"""检查工具是否可用(如依赖是否安装)"""
return True # 始终可用
# 模块级注册 — discover_builtin_tools() 会自动检测此调用
registry.register(
name="my_custom_tool",
toolset="custom",
schema={
"type": "function",
"function": {
"name": "my_custom_tool",
"description": "执行自定义操作",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "查询内容"
},
"max_results": {
"type": "integer",
"description": "最大结果数",
"default": 10
}
},
"required": ["query"]
}
}
},
handler=_my_handler,
check_fn=_check_available,
description="自定义工具示例",
emoji="🔧",
)步骤 2:无需其他操作
由于 AST 自动发现机制,新工具会在下次启动时自动被检测和注册。
步骤 3(可选):添加到工具集
python
# 在 toolsets.py 中添加
TOOLSETS["custom"] = {
"description": "自定义工具集",
"tools": ["my_custom_tool"],
"includes": []
}上一章:第二章 Agent 核心循环下一章:第四章 系统提示词工程