Skip to content

第三章 工具系统

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",
    )
字段类型说明
namestr工具名称(唯一标识)
toolsetstr所属工具集
schemadictOpenAI function calling schema
handlerCallable工具执行函数
check_fnCallable可用性检查函数(如检查环境变量)
requires_envlist需要的环境变量列表
is_asyncbool是否为异步工具
descriptionstr工具描述
emojistr显示用 emoji
max_result_size_charsint结果最大字符数限制

注册 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 imported

AST 检测逻辑

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__.pyregistry.pymcp_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 hookspre_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 result

3.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 tools

3.5 核心工具

terminal_tool — 终端执行

支持 6 种后端,通过配置切换执行环境:

后端说明环境变量/配置
local本地执行(默认)
dockerDocker 容器内执行TERMINAL_BACKEND=docker
modalModal 云端执行MODAL_APP_NAME
ssh远程 SSH 执行SSH_HOST, SSH_KEY
singularitySingularity 容器SINGULARITY_IMAGE
daytonaDaytona 开发环境DAYTONA_API_KEY

file_tools — 文件操作

  • read_file — 读取文件内容(支持行范围)
  • write_file — 写入文件
  • search_files — 文件内容搜索(正则/文本)
  • patch — 精确文件修补(diff-based)

web_tools — Web 工具

多后端搜索引擎支持:

后端说明
firecrawlFirecrawl API
exaExa 搜索
parallel并行多源搜索
tavilyTavily 搜索

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 核心循环下一章第四章 系统提示词工程

基于 MIT 许可发布