第五篇：Hook 与插件扩展
概述
Hermes Agent 的核心设计遵循"开闭原则"——对扩展开放，对修改关闭。Hook（钩子）系统和 Plugin（插件）系统是两大扩展机制。Hook 是轻量级的事件处理器，在 Agent 生命周期的关键节点被触发，适合横切关注点（如国际化、审计、通知）。Plugin 则是更重的功能模块，用于扩展 Agent 的底层能力（如记忆存储、上下文引擎、MCP Server）。
本篇将从源码级别深入分析 Hook 系统的实现机制，通过 i18n 国际化 Hook 这一真实案例展示完整的开发流程，并给出 Hook 与 Plugin 的选择指南。
1. Hook 系统详解
1.1 HookRegistry：发现、加载与触发
Hook 系统的核心是 HookRegistry 类（源码位于 gateway/hooks.py，第 34 行）：
class HookRegistry:
"""Discovers, loads, and fires event hooks.
Usage:
registry = HookRegistry()
registry.discover_and_load()
await registry.emit("agent:start", {"platform": "telegram", ...})
"""
def __init__(self):
# event_type -> [handler_fn, ...]
self._handlers: Dict[str, List[Callable]] = {}
self._loaded_hooks: List[dict] = [] # metadata for listing
HookRegistry 维护两个核心数据结构：
_handlers：事件类型到处理函数列表的映射，例如{"agent:start": [fn1, fn2], "session:end": [fn3]}_loaded_hooks：所有已加载 Hook 的元数据列表，用于调试和自省
1.2 discover_and_load()：发现与加载
discover_and_load() 方法（第 69 行）是 Hook 系统的入口点，在 Gateway 启动时被调用：
def discover_and_load(self) -> None:
"""Scan the hooks directory for hook directories and load their handlers.
Also registers built-in hooks that are always active.
Each hook directory must contain:
- HOOK.yaml with at least 'name' and 'events' keys
- handler.py with a top-level 'handle' function (sync or async)
"""
self._register_builtin_hooks() # 1. 先注册内置 Hook
if not HOOKS_DIR.exists(): # HOOKS_DIR = ~/.hermes/hooks/
return
for hook_dir in sorted(HOOKS_DIR.iterdir()): # 2. 扫描 hooks 目录
if not hook_dir.is_dir():
continue
manifest_path = hook_dir / "HOOK.yaml" # 3. 读取清单文件
handler_path = hook_dir / "handler.py" # 4. 定位处理模块
if not manifest_path.exists() or not handler_path.exists():
continue
try:
manifest = yaml.safe_load(manifest_path.read_text(encoding="utf-8"))
hook_name = manifest.get("name", hook_dir.name)
events = manifest.get("events", [])
if not events:
continue
# 5. 动态加载 handler 模块
spec = importlib.util.spec_from_file_location(
f"hermes_hook_{hook_name}", handler_path
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
handle_fn = getattr(module, "handle", None)
if handle_fn is None:
continue
# 6. 为每个声明的事件注册处理函数
for event in events:
self._handlers.setdefault(event, []).append(handle_fn)
self._loaded_hooks.append({
"name": hook_name,
"description": manifest.get("description", ""),
"events": events,
"path": str(hook_dir),
})
except Exception as e:
print(f"[hooks] Error loading hook {hook_dir.name}: {e}")
加载流程的关键步骤：
- 内置 Hook 注册：首先注册
boot-md等内置 Hook - 目录扫描：遍历
~/.hermes/hooks/下的所有子目录 - 清单解析：读取
HOOK.yaml获取 Hook 的名称和关注的事件列表 - 动态导入：使用
importlib.util.spec_from_file_location()动态加载handler.py - 函数提取：从模块中查找名为
handle的顶层函数 - 事件注册：将
handle函数注册到所有声明的事件类型上
1.3 emit()：事件触发
emit() 方法（第 138 行）负责在指定事件发生时调用所有已注册的处理函数：
async def emit(self, event_type: str, context: Optional[Dict[str, Any]] = None) -> None:
"""Fire all handlers registered for an event.
Supports wildcard matching: handlers registered for "command:*" will
fire for any "command:..." event.
"""
if context is None:
context = {}
# Collect handlers: exact match + wildcard match
handlers = list(self._handlers.get(event_type, []))
# Check for wildcard patterns
if ":" in event_type:
base = event_type.split(":")[0]
wildcard_key = f"{base}:*"
handlers.extend(self._handlers.get(wildcard_key, []))
for fn in handlers:
try:
result = fn(event_type, context)
# Support both sync and async handlers
if asyncio.iscoroutine(result):
await result
except Exception as e:
print(f"[hooks] Error in handler for '{event_type}': {e}")
关键设计：
- 通配符匹配：
command:*匹配所有command:前缀的事件（如command:reset、command:new） - 同步/异步兼容：通过
asyncio.iscoroutine()检测，同步函数直接调用结果，异步函数 await - 错误隔离：单个 Hook 的异常不会影响其他 Hook 或主流程（try/except 包裹）
- 非阻塞保证：Hook 失败只打印日志，永远不会阻断消息处理管线
1.4 生命周期事件
Hook 系统定义了以下生命周期事件：
| 事件 | 触发时机 | 典型用途 |
|---|---|---|
gateway:startup | Gateway 进程启动 | 执行启动检查清单、发送启动通知 |
session:start | 新会话创建（首次消息） | 初始化会话状态、加载用户偏好 |
session:end | 会话结束（/new 或 /reset） | 清理会话资源、保存会话摘要 |
session:reset | 会话重置完成 | 通知用户新会话已就绪 |
agent:start | Agent 开始处理消息 | 记录审计日志、设置计时器 |
agent:step | Agent 工具调用循环的每一步 | 步骤级监控、资源限制 |
agent:end | Agent 完成消息处理 | 计时统计、结果通知 |
command:* | 任何斜杠命令执行 | 命令审计、权限检查 |
1.5 Hook 目录结构
每个 Hook 是 ~/.hermes/hooks/ 下的一个目录，包含两个必需文件：
~/.hermes/hooks/
└── my-hook/ # Hook 目录（名称任意）
├── HOOK.yaml # 元数据清单
└── handler.py # 处理函数
HOOK.yaml 格式：
name: my-hook # Hook 名称（必填）
description: 一个示例 Hook # 描述（可选）
events: # 关注的事件列表（必填）
- agent:start
- agent:end
handler.py 格式：
async def handle(event_type: str, context: dict) -> None:
"""Hook handler — must be a top-level function named 'handle'.
Args:
event_type: 事件类型标识符（如 "agent:start"）
context: 事件上下文数据（dict）
"""
print(f"Hook triggered: {event_type}")
# 你的处理逻辑
处理函数的签名必须为 handle(event_type: str, context: dict)，可以是同步或异步函数。
2. 实战：i18n 国际化 Hook
以下通过一个真实的 i18n 国际化 Hook 案例展示完整的 Hook 开发流程。该 Hook 的目标是：当 Hermes 通过任何平台适配器发送消息时，自动将消息文本翻译为目标语言。
2.1 设计思路
i18n Hook 采用 monkey-patch 策略：在 gateway:startup 事件中，替换 BasePlatformAdapter.send() 方法，在原始发送逻辑之前插入翻译步骤。
BasePlatformAdapter.send() 的调用链：
原始: agent → adapter.send() → 平台 API
Hook: agent → adapter.send() → translate() → adapter._original_send() → 平台 API
2.2 HOOK.yaml 清单
name: i18n
description: Automatic message translation for multi-language support
events:
- gateway:startup
2.3 handler.py 完整代码分析
# ~/.hermes/hooks/i18n/handler.py
import logging
import pkgutil
import importlib
logger = logging.getLogger("hooks.i18n")
# --- 翻译字典（从 YAML 加载） ---
_translation_dict = {}
def _load_translations():
"""加载 YAML 翻译字典"""
global _translation_dict
try:
from pathlib import Path
import yaml
dict_path = Path(__file__).parent / "translations.yaml"
if dict_path.exists():
with open(dict_path, "r", encoding="utf-8") as f:
_translation_dict = yaml.safe_load(f) or {}
except Exception as e:
logger.warning("i18n: Failed to load translations: %s", e)
def translate(text: str, target_lang: str = "en") -> str:
"""翻译文本，支持精确匹配和子串匹配。
策略:
1. 精确匹配：text 在翻译字典中直接命中
2. 子串匹配：遍历字典中的 key，替换 text 中的匹配项
"""
if not text or not _translation_dict:
return text
lang_dict = _translation_dict.get(target_lang, {})
if not lang_dict:
return text
# 1. 精确匹配
if text in lang_dict:
return lang_dict[text]
# 2. 子串匹配 — 按长度降序排列，优先匹配更长的子串
translated = text
for source, target in sorted(lang_dict.items(), key=lambda x: -len(x[0])):
if source in translated:
translated = translated.replace(source, target)
return translated
async def handle(event_type: str, context: dict) -> None:
"""gateway:startup handler — monkey-patch BasePlatformAdapter.send()"""
# 1. 加载翻译字典
_load_translations()
logger.info("i18n: Loaded %d translation entries", len(_translation_dict))
# 2. 使用 pkgutil.iter_modules 遍历 gateway.platforms 的所有子模块
# 自动发现所有 BasePlatformAdapter 的子类
import gateway.platforms as platforms_pkg
from gateway.platforms.base import BasePlatformAdapter
patched_classes = set()
for importer, module_name, is_pkg in pkgutil.iter_modules(
platforms_pkg.__path__, platforms_pkg.__name__ + "."
):
try:
module = importlib.import_module(module_name)
except ImportError:
continue
# 3. 遍历模块中的所有类，查找 BasePlatformAdapter 的子类
for attr_name in dir(module):
attr = getattr(module, attr_name)
if (
isinstance(attr, type)
and issubclass(attr, BasePlatformAdapter)
and attr is not BasePlatformAdapter
and attr not in patched_classes
):
# 4. Monkey-patch send() 方法
_patch_send(attr)
patched_classes.add(attr)
logger.info("i18n: Patched %d adapter classes", len(patched_classes))
def _patch_send(adapter_class):
"""保存原始 send() 并替换为翻译版本"""
original_send = adapter_class.send
async def translated_send(self, chat_id, content, reply_to=None, metadata=None):
# 翻译消息内容
target_lang = getattr(self, "_target_lang", "en")
translated_content = translate(content, target_lang)
# 调用原始 send
return await original_send(self, chat_id, translated_content, reply_to, metadata)
adapter_class.send = translated_send
logger.debug("i18n: Patched %s.send()", adapter_class.__name__)
2.4 翻译字典（translations.yaml）
en:
"你好": "Hello"
"任务已完成": "Task completed"
"正在处理": "Processing"
"部署成功": "Deployment successful"
"发生错误": "An error occurred"
ja:
"你好": "こんにちは"
"任务已完成": "タスク完了"
2.5 关键技术点解析
pkgutil.iter_modules 遍历子类：使用 pkgutil.iter_modules() 动态遍历 gateway.platforms 包下的所有模块，然后通过 issubclass() 检测 BasePlatformAdapter 的子类。这意味着即使未来添加了新的平台适配器，i18n Hook 也能自动生效，无需修改。
translate() 的精确+子串匹配：翻译策略分两层。首先尝试精确匹配（整个文本作为一个 key），如果未命中则进行子串替换。子串替换按 key 长度降序排列，避免短 key 吞噬长 key 的情况（例如先匹配 "好" 会破坏 "你好" 的翻译）。
monkey-patch send()：将 BasePlatformAdapter.send() 替换为包装版本，在调用原始方法前插入翻译逻辑。这种方式对平台适配器完全透明——适配器代码无需任何改动。
3. 内置 Hook
3.1 boot-md：启动时执行 BOOT.md
boot-md 是 Hermes 唯一的内置 Hook（源码位于 gateway/builtin_hooks/boot_md.py），它在 Gateway 启动时检查 ~/.hermes/BOOT.md 文件是否存在，如果存在则执行其中的指令。
async def handle(event_type: str, context: dict) -> None:
"""Gateway startup handler — run BOOT.md if it exists."""
if not BOOT_FILE.exists():
return
content = BOOT_FILE.read_text(encoding="utf-8").strip()
if not content:
return
logger.info("Running BOOT.md (%d chars)", len(content))
# Run in a background thread so we don't block gateway startup.
thread = threading.Thread(
target=_run_boot_agent,
args=(content,),
name="boot-md",
daemon=True,
)
thread.start()
核心设计：
- 非阻塞：使用独立线程执行，不阻塞 Gateway 启动
- 静默模式：Agent 回复包含
[SILENT]则不发送通知 - 限制迭代：
max_iterations=20防止无限循环
def _run_boot_agent(content: str) -> None:
prompt = _build_boot_prompt(content)
agent = AIAgent(
quiet_mode=True,
skip_context_files=True,
skip_memory=True,
max_iterations=20,
)
result = agent.run_conversation(prompt)
response = result.get("final_response", "")
if response and "[SILENT]" not in response:
logger.info("boot-md completed: %s", response[:200])
BOOT.md 示例：
# Startup Checklist
1. Check if any cron jobs failed overnight
2. Send a status update to Discord #general
3. If there are errors in /opt/app/deploy.log, summarize them
内置 Hook 的注册发生在 _register_builtin_hooks() 方法中（第 54 行）：
def _register_builtin_hooks(self) -> None:
try:
from gateway.builtin_hooks.boot_md import handle as boot_md_handle
self._handlers.setdefault("gateway:startup", []).append(boot_md_handle)
self._loaded_hooks.append({
"name": "boot-md",
"description": "Run ~/.hermes/BOOT.md on gateway startup",
"events": ["gateway:startup"],
"path": "(builtin)",
})
except Exception as e:
print(f"[hooks] Could not load built-in boot-md hook: {e}")
4. 插件系统
4.1 插件与 Hook 的区别
Plugin 是 Hermes 的另一种扩展机制，与 Hook 有本质区别：
| 维度 | Hook | Plugin |
|---|---|---|
| 加载方式 | ~/.hermes/hooks/ 目录发现 | config.yaml 配置注册 |
| 触发机制 | 事件驱动（生命周期事件） | 接口实现（Provider 模式） |
| 执行时机 | 在主流程的关键节点 | 作为独立模块被调用 |
| 典型用途 | 消息拦截、审计、通知 | 记忆存储、上下文引擎、工具服务 |
| 代码侵入性 | monkey-patch（可选） | 接口实现（显式） |
| 错误处理 | 静默失败，不阻断主流程 | 需要自行处理，可能影响功能 |
4.2 Memory Provider 插件
Memory Provider 插件扩展 Agent 的记忆存储能力。不同的 Provider 实现不同的存储后端：
# config.yaml
memory:
provider: sqlite # 或 redis、chromadb 等
sqlite:
path: ~/.hermes/memory.db
Memory Provider 需要实现特定的接口（如 store()、retrieve()、search()），由 Agent 的记忆系统在运行时调用。
4.3 Context Engine 插件
Context Engine 插件控制 Agent 如何构建和管理上下文窗口。不同的 Engine 可以实现不同的策略：
- 滑动窗口：保留最近 N 轮对话
- 摘要压缩：对历史对话进行摘要，保留关键信息
- 向量检索：基于语义相似度检索相关历史
4.4 MCP Server 插件
MCP（Model Context Protocol）Server 插件为 Agent 提供外部工具和数据源。MCP 是一个标准化协议，允许 Agent 连接到各种外部服务：
# config.yaml
mcp_servers:
- name: filesystem
command: npx
args: ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/dir"]
- name: database
command: python
args: ["-m", "mcp_server_db"]
MCP Server 作为独立进程运行，通过 JSON-RPC 协议与 Agent 通信。
5. Hook vs Plugin 选择指南
5.1 何时使用 Hook
以下场景适合使用 Hook：
- 横切关注点：需要在多个平台适配器上统一生效的行为（如国际化、审计日志、消息过滤）
- 生命周期拦截：在 Agent 处理流程的特定节点插入逻辑（如 session 开始时初始化、结束时清理）
- 轻量级扩展：不需要引入新的存储后端或外部服务，只是修改现有行为
- 快速原型：可以通过 monkey-patch 快速验证想法，无需修改核心代码
典型 Hook 场景：
- 消息内容过滤/转换
- 用户权限检查
- 消息转发/通知
- 性能监控和日志
- 自定义命令拦截
5.2 何时使用 Plugin
以下场景适合使用 Plugin：
- 新能力接入：需要为 Agent 添加全新的能力（如新的记忆存储后端、外部工具服务）
- 数据持久化：需要独立的数据存储和管理（如向量数据库、图数据库）
- 外部服务集成：需要与外部服务建立长连接或复杂交互（如数据库、搜索引擎）
- 配置驱动：需要用户在 config.yaml 中配置参数
典型 Plugin 场景：
- 替换默认的 SQLite 记忆存储为 Redis/ChromaDB
- 添加新的 MCP 工具服务
- 自定义上下文压缩策略
- 接入自定义的 LLM Provider
5.3 决策流程图
需要修改现有行为还是添加新能力？
├── 修改现有行为 → 需要修改所有平台适配器？
│ ├── 是 → Hook（monkey-patch BasePlatformAdapter）
│ └── 否 → 修改特定适配器代码或 Hook 特定事件
├── 添加新能力 → 需要外部服务或持久化？
│ ├── 是 → Plugin
│ └── 否 → Hook 或直接修改核心代码
└── 只在生命周期节点执行 → Hook
5.4 最佳实践
Hook 最佳实践：
- 保持轻量：Hook 处理函数应快速返回，耗时操作使用后台线程或 asyncio.create_task()
- 错误隔离：虽然 HookRegistry 会捕获异常，但处理函数内部也应做好错误处理
- 幂等性：Hook 可能被多次触发（如 gateway 重启），确保逻辑是幂等的
- 避免循环：Hook 中调用 Agent 的方法可能再次触发事件，注意避免无限循环
- 日志记录：使用
logging模块记录关键操作，便于调试
Plugin 最佳实践：
- 接口一致性：严格实现 Provider 接口的所有方法
- 优雅降级：当外部服务不可用时，提供合理的回退行为
- 配置验证：在启动时验证所有必要的配置参数
- 资源管理：正确管理连接池、文件句柄等资源
- 测试覆盖：为 Plugin 编写单元测试和集成测试
调试指南
本节列出 Hook 与插件系统中最常见的问题及排查方法。
Hook 未加载
症状：Hook 目录存在但 Handler 未被触发，emit() 日志中看不到对应事件处理记录。
排查步骤：
# 检查 Hook 目录结构是否完整
ls -la ~/.hermes/hooks/<hook-name>/
# 应包含 HOOK.yaml 和 handler.py 两个文件
# 验证 HOOK.yaml 格式（必须有 name 和 events 字段）
cat ~/.hermes/hooks/<hook-name>/HOOK.yaml
# 检查 handler.py 中是否有顶层 handle 函数
grep -n "^def handle\|^async def handle" ~/.hermes/hooks/<hook-name>/handler.py
常见原因：缺少 HOOK.yaml 或 handler.py、HOOK.yaml 中 events 字段为空或拼写错误（如写成 event）、handler.py 中没有名为 handle 的顶层函数（必须是模块级函数，不是类方法）。检查 discover_and_load() 的日志输出，搜索 [hooks] Error loading hook 字样。
Hook 执行报错
症状：Hook 被加载了但运行时出错，主流程未受影响但 Hook 功能失效。
排查步骤：
# 查看 Hook 执行错误日志
journalctl --user -u hermes-gateway | grep "\[hooks\] Error in handler"
# 查看更详细的错误堆栈
grep -A 5 "Error in handler" ~/.hermes/logs/gateway.log | tail -30
常见原因：handler.py 中的代码抛出未捕获的异常（如 ImportError、KeyError、AttributeError）。HookRegistry 的 emit() 会捕获所有异常并打印日志，但不会中断主流程。检查错误信息中提到的具体异常类型和行号，修复后重启 Gateway 即可生效。
Monkey-patch 失效
症状：i18n Hook 等 monkey-patch 类 Hook 不生效，消息未被翻译或处理。
排查步骤：
# 检查 patch 日志（i18n Hook 会输出 patched 数量）
journalctl --user -u hermes-gateway | grep -i "patched\|patch"
# 确认平台适配器是否在 patch 之后才加载
grep -E "adapter|platform" ~/.hermes/logs/gateway.log | head -20
常见原因：如果某个平台适配器子类自己覆盖了 send() 方法，那么对 BasePlatformAdapter.send() 的 monkey-patch 不会影响该子类。解决方案是在 _patch_send() 中对每个发现的子类单独 patch，而不是只 patch 基类。检查 i18n Hook 的 handle() 函数中 patched_classes 集合的大小——如果为 0，说明未发现任何子类。
调试技巧
在 Hook 开发过程中，可以利用以下方法快速定位问题：
# 方法 1：在 handler.py 中加 print() 输出
# print() 的输出会出现在 journalctl 日志中
# 示例：在 handle() 函数开头添加
# print(f"[debug-hook] event={event_type}, context keys={list(context.keys())}")
# 查看输出
journalctl --user -u hermes-gateway -f | grep debug-hook
# 方法 2：使用 logging 模块（推荐用于生产）
# import logging; logger = logging.getLogger("hooks.my-hook")
# logger.info("...")
# 方法 3：检查 Hook 是否在加载列表中
# 在 Gateway 启动日志中搜索
journalctl --user -u hermes-gateway | grep -i "hook"
修改 handler.py 后需要重启 Gateway 才能生效（Hook 在启动时一次性加载）。开发阶段建议使用 hermes gateway run 前台运行，便于实时观察日志输出。
思考题
-
错误传播：Hook 系统的设计原则是"Hook 失败不阻断主流程"。但考虑一个审计 Hook，它需要在每次消息处理后记录审计日志。如果审计 Hook 失败了，消息仍然被发送，但审计记录丢失了。这个行为是否符合预期？你会如何设计一个更健壮的审计系统？
-
通配符匹配：当前
emit()实现了command:*匹配所有command:前缀的事件。如果未来需要支持更复杂的模式（如agent:*匹配agent:start和agent:step，但不匹配agent），你会如何修改匹配逻辑？ -
i18n Hook 的 monkey-patch：i18n Hook 通过替换
BasePlatformAdapter.send()实现翻译。这种方式有什么潜在风险？考虑以下场景：如果另一个 Hook 也 monkey-patch 了send()，会发生什么？你能想到更安全的实现方式吗？ -
boot-md Hook：boot-md 在独立线程中运行 Agent。为什么要用线程而不是 asyncio.create_task()？如果在主事件循环中运行，会有什么问题？
-
Hook 与 Plugin 的边界：假设你需要实现一个功能——在每次 Agent 回复后，自动将对话保存到 Elasticsearch 进行全文检索。这应该实现为 Hook 还是 Plugin？请分析两种方案的优劣。