事件钩子

Hermes 提供了两个钩子系统，可在关键生命周期节点运行自定义代码：

系统	注册方式	运行环境	使用场景
网关钩子	`HOOK.yaml` + `handler.py` 存放在 `~/.hermes/hooks/` 目录下	仅网关	日志记录、告警、Webhook
插件钩子	在插件中通过 `ctx.register_hook()` 注册	CLI + 网关	工具拦截、指标收集、安全策略

两个系统均为非阻塞模式 —— 任何钩子中的错误都会被捕获并记录，绝不会导致代理崩溃。

网关事件钩子

网关钩子会在网关运行期间（Telegram、Discord、Slack、WhatsApp）自动触发，且不会阻塞主代理流程。

创建钩子

每个钩子是一个位于 ~/.hermes/hooks/ 下的目录，包含两个文件：

~/.hermes/hooks/
└── my-hook/
    ├── HOOK.yaml      # Declares which events to listen for
    └── handler.py     # Python handler function

HOOK.yaml

name: my-hook
description: Log all agent activity to a file
events:
  - agent:start
  - agent:end
  - agent:step

events 列表决定了哪些事件会触发你的处理器。你可以订阅任意组合的事件，包括通配符如 command:*。

handler.py

import json
from datetime import datetime
from pathlib import Path

LOG_FILE = Path.home() / ".hermes" / "hooks" / "my-hook" / "activity.log"

async def handle(event_type: str, context: dict):
    """Called for each subscribed event. Must be named 'handle'."""
    entry = {
        "timestamp": datetime.now().isoformat(),
        "event": event_type,
        **context,
    }
    with open(LOG_FILE, "a") as f:
        f.write(json.dumps(entry) + "\n")

处理器规则：

必须命名为 handle
接收 event_type（字符串）和 context（字典）
可以是 async def 或普通 def —— 两者均有效
错误会被捕获并记录，绝不会导致代理崩溃

可用事件

事件	触发时机	上下文键
`gateway:startup`	网关进程启动时	`platforms`（当前激活的平台名称列表）
`session:start`	新的消息会话创建时	`platform`、`user_id`、`session_id`、`session_key`
`session:end`	会话结束（重置前）	`platform`、`user_id`、`session_key`
`session:reset`	用户执行 `/new` 或 `/reset` 时	`platform`、`user_id`、`session_key`
`agent:start`	代理开始处理消息时	`platform`、`user_id`、`session_id`、`message`
`agent:step`	每次工具调用循环的迭代	`platform`、`user_id`、`session_id`、`iteration`、`tool_names`
`agent:end`	代理完成处理时	`platform`、`user_id`、`session_id`、`message`、`response`
`command:*`	任意斜杠命令执行时	`platform`、`user_id`、`command`、`args`

通配符匹配

注册为 command:* 的处理器将对所有 command: 事件（如 command:model、command:reset 等）触发。通过单一订阅即可监控所有斜杠命令。

示例

启动检查清单（BOOT.md）—— 内置功能

网关自带一个内置的 boot-md 钩子，会在每次启动时检查 ~/.hermes/BOOT.md 文件是否存在。如果文件存在，代理将在后台会话中执行其指令。无需安装 —— 只需创建该文件即可。

创建 ~/.hermes/BOOT.md：

# Startup Checklist

Check if any cron jobs failed overnight — run `hermes cron list`
Send a message to Discord #general saying "Gateway restarted, all systems go"
Check if /opt/app/deploy.log has any errors from the last 24 hours

代理会在后台线程中运行这些指令，因此不会阻塞网关启动。如果无需处理任何事项，代理将回复 [SILENT]，且不会发送任何消息。

提示

没有 BOOT.md？该钩子会静默跳过 —— 无任何开销。需要启动自动化时创建文件，不需要时删除即可。

长任务时发送 Telegram 告警

当代理执行超过 10 步时，向自己发送一条消息：

# ~/.hermes/hooks/long-task-alert/HOOK.yaml
name: long-task-alert
description: Alert when agent is taking many steps
events:
  - agent:step

# ~/.hermes/hooks/long-task-alert/handler.py
import os
import httpx

THRESHOLD = 10
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
CHAT_ID = os.getenv("TELEGRAM_HOME_CHANNEL")

async def handle(event_type: str, context: dict):
    iteration = context.get("iteration", 0)
    if iteration == THRESHOLD and BOT_TOKEN and CHAT_ID:
        tools = ", ".join(context.get("tool_names", []))
        text = f"⚠️ Agent has been running for {iteration} steps. Last tools: {tools}"
        async with httpx.AsyncClient() as client:
            await client.post(
                f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage",
                json={"chat_id": CHAT_ID, "text": text},
            )

命令使用日志记录器

记录哪些斜杠命令被使用：

# ~/.hermes/hooks/command-logger/HOOK.yaml
name: command-logger
description: Log slash command usage
events:
  - command:*

# ~/.hermes/hooks/command-logger/handler.py
import json
from datetime import datetime
from pathlib import Path

LOG = Path.home() / ".hermes" / "logs" / "command_usage.jsonl"

def handle(event_type: str, context: dict):
    LOG.parent.mkdir(parents=True, exist_ok=True)
    entry = {
        "ts": datetime.now().isoformat(),
        "command": context.get("command"),
        "args": context.get("args"),
        "platform": context.get("platform"),
        "user": context.get("user_id"),
    }
    with open(LOG, "a") as f:
        f.write(json.dumps(entry) + "\n")

会话开始时触发 Webhook

在新会话创建时向外部服务发送 POST 请求：

# ~/.hermes/hooks/session-webhook/HOOK.yaml
name: session-webhook
description: Notify external service on new sessions
events:
  - session:start
  - session:reset

# ~/.hermes/hooks/session-webhook/handler.py
import httpx

WEBHOOK_URL = "https://your-service.example.com/hermes-events"

async def handle(event_type: str, context: dict):
    async with httpx.AsyncClient() as client:
        await client.post(WEBHOOK_URL, json={
            "event": event_type,
            **context,
        }, timeout=5)

工作原理

网关启动时，HookRegistry.discover_and_load() 扫描 ~/.hermes/hooks/ 目录
每个包含 HOOK.yaml 和 handler.py 的子目录会被动态加载
处理器根据声明的事件进行注册
在每个生命周期节点，hooks.emit() 会触发所有匹配的处理器
任何处理器中的错误都会被捕获并记录 —— 一个损坏的钩子绝不会导致代理崩溃

信息

网关钩子仅在网关（Telegram、Discord、Slack、WhatsApp）中触发。CLI 不会加载网关钩子。如需在所有环境中工作的钩子，请使用插件钩子。

插件钩子

插件可通过在插件的 register() 函数中调用 ctx.register_hook() 来注册钩子，这些钩子会在 CLI 和网关 会话中均触发。

def register(ctx):
    ctx.register_hook("pre_tool_call", my_tool_observer)
    ctx.register_hook("post_tool_call", my_tool_logger)
    ctx.register_hook("pre_llm_call", my_memory_callback)
    ctx.register_hook("post_llm_call", my_sync_callback)
    ctx.register_hook("on_session_start", my_init_callback)
    ctx.register_hook("on_session_end", my_cleanup_callback)

所有钩子的通用规则：

回调函数接收 关键字参数。始终接受 **kwargs 以保证向前兼容 —— 未来版本可能会添加新参数，但不会破坏你的插件。
如果回调函数崩溃，它会被记录并跳过。其他钩子和代理将继续正常运行。行为异常的插件绝不会导致代理崩溃。
所有钩子均为 一次性观察者，其返回值被忽略 —— 除了 pre_llm_call，它可以注入上下文。

快速参考

钩子	触发时机	返回值
`pre_tool_call`	在任何工具执行前	忽略
`post_tool_call`	在任何工具返回后	忽略
`pre_llm_call`	每轮对话开始时，工具调用循环之前	上下文注入
`post_llm_call`	每轮对话结束时，工具调用循环之后	忽略
`on_session_start`	新会话创建时（仅第一轮）	忽略
`on_session_end`	会话结束时	忽略

`pre_tool_call`

在每次工具执行之前立即触发——包括内置工具和插件工具。

回调签名：

def my_callback(tool_name: str, args: dict, task_id: str, **kwargs):

参数	类型	描述
`tool_name`	`str`	即将执行的工具名称（例如 `"terminal"`、`"web_search"`、`"read_file"`）
`args`	`dict`	模型传递给该工具的参数
`task_id`	`str`	会话/任务标识符。未设置时为空字符串。

触发位置： 在 model_tools.py 中的 handle_function_call() 内部，在工具处理器运行之前。每调用一次工具即触发一次——如果模型并行调用 3 个工具，则此钩子将触发 3 次。

返回值： 忽略。

使用场景： 日志记录、审计追踪、工具调用计数、阻止危险操作（打印警告）、速率限制。

示例 —— 工具调用审计日志：

import json, logging
from datetime import datetime

logger = logging.getLogger(__name__)

def audit_tool_call(tool_name, args, task_id, **kwargs):
    logger.info("TOOL_CALL session=%s tool=%s args=%s",
                task_id, tool_name, json.dumps(args)[:200])

def register(ctx):
    ctx.register_hook("pre_tool_call", audit_tool_call)

示例 —— 对危险工具发出警告：

DANGEROUS = {"terminal", "write_file", "patch"}

def warn_dangerous(tool_name, **kwargs):
    if tool_name in DANGEROUS:
        print(f"⚠ Executing potentially dangerous tool: {tool_name}")

def register(ctx):
    ctx.register_hook("pre_tool_call", warn_dangerous)

`post_tool_call`

在每次工具执行返回后立即触发。

回调签名：

def my_callback(tool_name: str, args: dict, result: str, task_id: str, **kwargs):

参数	类型	描述
`tool_name`	`str`	刚刚执行完毕的工具名称
`args`	`dict`	模型传递给该工具的参数
`result`	`str`	工具的返回值（始终为 JSON 字符串）
`task_id`	`str`	会话/任务标识符。未设置时为空字符串。

触发位置： 在 model_tools.py 中的 handle_function_call() 内部，在工具处理器返回之后。每调用一次工具即触发一次。如果工具抛出未处理的异常（异常被捕获并以错误 JSON 字符串形式返回），则 post_tool_call 仍会触发，且 result 参数为该错误字符串。

返回值： 忽略。

使用场景： 记录工具结果、指标收集、跟踪工具成功率/失败率、特定工具完成时发送通知。

示例 —— 跟踪工具使用指标：

from collections import Counter
import json

_tool_counts = Counter()
_error_counts = Counter()

def track_metrics(tool_name, result, **kwargs):
    _tool_counts[tool_name] += 1
    try:
        parsed = json.loads(result)
        if "error" in parsed:
            _error_counts[tool_name] += 1
    except (json.JSONDecodeError, TypeError):
        pass

def register(ctx):
    ctx.register_hook("post_tool_call", track_metrics)

`pre_llm_call`

每轮对话仅触发一次，在工具调用循环开始前触发。这是唯一一个返回值会被使用的钩子——它可以将上下文注入到当前轮次的用户消息中。

回调签名：

def my_callback(session_id: str, user_message: str, conversation_history: list,
                is_first_turn: bool, model: str, platform: str, **kwargs):

参数	类型	描述
`session_id`	`str`	当前会话的唯一标识符
`user_message`	`str`	当前轮次用户原始消息（在任何技能注入前）
`conversation_history`	`list`	完整消息列表的副本（OpenAI 格式：`[{"role": "user", "content": "..."}]`）
`is_first_turn`	`bool`	如果是新会话的第一轮，则为 `True`；后续轮次为 `False`
`model`	`str`	模型标识符（例如 `"anthropic/claude-sonnet-4.6"`）
`platform`	`str`	会话运行的平台：`"cli"`、`"telegram"`、`"discord"` 等。

触发位置： 在 run_agent.py 中的 run_conversation() 内部，上下文压缩之后、主 while 循环之前。每调用一次 run_conversation() 即触发一次（即每轮用户输入触发一次），而非在工具循环内的每次 API 调用时触发。

返回值： 如果回调返回一个包含 "context" 键的字典，或一个非空字符串，则该文本将被追加到当前轮次的用户消息末尾。返回 None 表示不注入上下文。

# Inject context
return {"context": "Recalled memories:\n- User likes Python\n- Working on hermes-agent"}

# Plain string (equivalent)
return "Recalled memories:\n- User likes Python"

# No injection
return None

上下文注入位置： 始终注入到用户消息中，从不注入到系统提示中。这保留了提示缓存——系统提示在各轮之间保持一致，因此缓存的 token 可被复用。系统提示属于 Hermes 的范畴（模型引导、工具强制、个性、技能）。插件则在用户输入旁贡献上下文。

所有注入的上下文均为临时性——仅在 API 调用时添加。对话历史中的原始用户消息不会被修改，且不会持久化到会话数据库中。

当多个插件返回上下文时，它们的输出将按插件发现顺序（按目录名字母顺序）用双换行符连接。

使用场景： 记忆召回、RAG 上下文注入、安全护栏、每轮分析。

示例 —— 记忆召回：

import httpx

MEMORY_API = "https://your-memory-api.example.com"

def recall(session_id, user_message, is_first_turn, **kwargs):
    try:
        resp = httpx.post(f"{MEMORY_API}/recall", json={
            "session_id": session_id,
            "query": user_message,
        }, timeout=3)
        memories = resp.json().get("results", [])
        if not memories:
            return None
        text = "Recalled context:\n" + "\n".join(f"- {m['text']}" for m in memories)
        return {"context": text}
    except Exception:
        return None

def register(ctx):
    ctx.register_hook("pre_llm_call", recall)

示例 —— 安全护栏：

POLICY = "Never execute commands that delete files without explicit user confirmation."

def guardrails(**kwargs):
    return {"context": POLICY}

def register(ctx):
    ctx.register_hook("pre_llm_call", guardrails)

`post_llm_call`

每轮对话仅触发一次，在工具调用循环完成后，代理生成最终响应时触发。仅在成功完成的轮次中触发——若该轮被中断，则不会触发。

回调签名：

def my_callback(session_id: str, user_message: str, assistant_response: str,
                conversation_history: list, model: str, platform: str, **kwargs):

参数	类型	描述
`session_id`	`str`	当前会话的唯一标识符
`user_message`	`str`	当前轮次中用户的原始消息
`assistant_response`	`str`	代理在当前轮次的最终文本响应
`conversation_history`	`list`	当轮次完成后完整的消息列表副本
`model`	`str`	模型标识符
`platform`	`str`	会话运行的平台

触发时机： 在 run_agent.py 中的 run_conversation() 函数内，工具循环退出并生成最终响应后触发。受 if final_response and not interrupted 保护 —— 因此当用户在轮次中途中断或代理达到迭代限制但未生成响应时，不会触发。

返回值： 忽略。

使用场景： 将对话数据同步到外部记忆系统、计算响应质量指标、记录轮次摘要、触发后续操作。

示例 —— 同步到外部记忆系统：

import httpx

MEMORY_API = "https://your-memory-api.example.com"

def sync_memory(session_id, user_message, assistant_response, **kwargs):
    try:
        httpx.post(f"{MEMORY_API}/store", json={
            "session_id": session_id,
            "user": user_message,
            "assistant": assistant_response,
        }, timeout=5)
    except Exception:
        pass  # best-effort

def register(ctx):
    ctx.register_hook("post_llm_call", sync_memory)

示例 —— 跟踪响应长度：

import logging
logger = logging.getLogger(__name__)

def log_response_length(session_id, assistant_response, model, **kwargs):
    logger.info("RESPONSE session=%s model=%s chars=%d",
                session_id, model, len(assistant_response or ""))

def register(ctx):
    ctx.register_hook("post_llm_call", log_response_length)

`on_session_start`

在创建全新会话时仅触发一次。在会话续接时（用户发送第二条消息到现有会话）不会触发。

回调签名：

def my_callback(session_id: str, model: str, platform: str, **kwargs):

参数	类型	描述
`session_id`	`str`	新会话的唯一标识符
`model`	`str`	模型标识符
`platform`	`str`	会话运行的平台

触发时机： 在 run_agent.py 中的 run_conversation() 函数内，新会话的第一轮中触发 —— 具体是在系统提示构建完成后、工具循环开始前。判断条件为 if not conversation_history（无先前消息 = 新会话）。

返回值： 忽略。

使用场景： 初始化会话范围的状态、预热缓存、向外部服务注册会话、记录会话启动日志。

示例 —— 初始化会话缓存：

_session_caches = {}

def init_session(session_id, model, platform, **kwargs):
    _session_caches[session_id] = {
        "model": model,
        "platform": platform,
        "tool_calls": 0,
        "started": __import__("datetime").datetime.now().isoformat(),
    }

def register(ctx):
    ctx.register_hook("on_session_start", init_session)

`on_session_end`

在每次 run_conversation() 调用的最末尾触发，无论结果如何。如果用户在代理处理过程中退出（如按 Ctrl+C 或输入 /exit），也会从 CLI 的退出处理器中触发。

回调签名：

def my_callback(session_id: str, completed: bool, interrupted: bool,
                model: str, platform: str, **kwargs):

参数	类型	描述
`session_id`	`str`	会话的唯一标识符
`completed`	`bool`	如果代理生成了最终响应则为 `True`，否则为 `False`
`interrupted`	`bool`	如果本轮被中断（用户发送新消息、输入 `/stop` 或退出）则为 `True`
`model`	`str`	模型标识符
`platform`	`str`	会话运行的平台

触发位置：

run_agent.py —— 每次 run_conversation() 调用结束后，所有清理操作完成后。无论本轮是否出错，都会触发。
cli.py —— 在 CLI 的 atexit 处理器中触发，但仅当代理处于处理中状态（_agent_running=True）时才触发。这会捕获在处理过程中按 Ctrl+C 或输入 /exit 的情况。此时 completed=False 且 interrupted=True。

返回值： 忽略。

使用场景： 刷新缓冲区、关闭连接、持久化会话状态、记录会话时长、清理在 on_session_start 中初始化的资源。

示例 —— 刷新并清理：

_session_caches = {}

def cleanup_session(session_id, completed, interrupted, **kwargs):
    cache = _session_caches.pop(session_id, None)
    if cache:
        # Flush accumulated data to disk or external service
        status = "completed" if completed else ("interrupted" if interrupted else "failed")
        print(f"Session {session_id} ended: {status}, {cache['tool_calls']} tool calls")

def register(ctx):
    ctx.register_hook("on_session_end", cleanup_session)

示例 —— 会话时长跟踪：

import time, logging
logger = logging.getLogger(__name__)

_start_times = {}

def on_start(session_id, **kwargs):
    _start_times[session_id] = time.time()

def on_end(session_id, completed, interrupted, **kwargs):
    start = _start_times.pop(session_id, None)
    if start:
        duration = time.time() - start
        logger.info("SESSION_DURATION session=%s seconds=%.1f completed=%s interrupted=%s",
                     session_id, duration, completed, interrupted)

def register(ctx):
    ctx.register_hook("on_session_start", on_start)
    ctx.register_hook("on_session_end", on_end)

有关完整指南，请参阅 构建插件指南，其中包含工具模式、处理器和高级钩子模式的详细说明。

网关事件钩子​

创建钩子​

HOOK.yaml​

handler.py​

可用事件​

通配符匹配​

示例​

启动检查清单（BOOT.md）—— 内置功能​

长任务时发送 Telegram 告警​

命令使用日志记录器​

会话开始时触发 Webhook​

工作原理​

插件钩子​

快速参考​

pre_tool_call​

post_tool_call​

pre_llm_call​

post_llm_call​

on_session_start​

on_session_end​

网关事件钩子

创建钩子

HOOK.yaml

handler.py

可用事件

通配符匹配

示例

启动检查清单（BOOT.md）—— 内置功能

长任务时发送 Telegram 告警

命令使用日志记录器

会话开始时触发 Webhook

工作原理

插件钩子

快速参考

`pre_tool_call`

`post_tool_call`

`pre_llm_call`

`post_llm_call`

`on_session_start`

`on_session_end`