构建 Hermes 插件
本指南从零开始,带你构建一个完整的 Hermes 插件。完成后你将拥有一个包含多个工具、生命周期 Hook、附带数据文件和内嵌 Skill 的工作插件 — 涵盖插件系统的所有功能。
你要构建什么
一个计算器插件,包含两个工具:
calculate— 计算数学表达式(2**16、sqrt(144)、pi * 5**2)unit_convert— 单位换算(100 F → 37.78 C、5 km → 3.11 mi)
还有一个记录每次工具调用的 Hook,以及一个内嵌的 Skill 文件。
第 1 步:创建插件目录
mkdir -p ~/.hermes/plugins/calculator
cd ~/.hermes/plugins/calculator
第 2 步:编写清单文件
创建 plugin.yaml:
name: calculator
version: 1.0.0
description: 数学计算器 — 计算表达式和单位换算
provides_tools:
- calculate
- unit_convert
provides_hooks:
- post_tool_call
这告诉 Hermes:"我是一个叫 calculator 的插件,我提供工具和 Hook。" provides_tools 和 provides_hooks 字段是插件注册内容的列表。
可选字段:
author: 你的名字
requires_env: # 根据环境变量控制加载;安装时提示
- SOME_API_KEY # 简单格式 — 缺少时插件禁用
- name: OTHER_KEY # 富格式 — 安装时显示描述/URL
description: "Other 服务的密钥"
url: "https://other.com/keys"
secret: true
第 3 步:编写工具 Schema
创建 schemas.py — 这是 LLM 读取以决定何时调用你的工具的内容:
"""工具 Schema — LLM 看到的内容。"""
CALCULATE = {
"name": "calculate",
"description": (
"计算数学表达式并返回结果。"
"支持算术运算(+、-、*、/、**)、函数(sqrt、sin、cos、"
"log、abs、round、floor、ceil)和常量(pi、e)。"
"用于用户询问的任何数学计算。"
),
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "要计算的数学表达式(如 '2**10'、'sqrt(144)')",
},
},
"required": ["expression"],
},
}
UNIT_CONVERT = {
"name": "unit_convert",
"description": (
"在不同单位间换算值。支持长度(m、km、mi、ft、in)、"
"重量(kg、lb、oz、g)、温度(C、F、K)、数据(B、KB、MB、GB、TB)"
"和时间(s、min、hr、day)。"
),
"parameters": {
"type": "object",
"properties": {
"value": {
"type": "number",
"description": "要换算的数值",
},
"from_unit": {
"type": "string",
"description": "源单位(如 'km'、'lb'、'F'、'GB')",
},
"to_unit": {
"type": "string",
"description": "目标单位(如 'mi'、'kg'、'C'、'MB')",
},
},
"required": ["value", "from_unit", "to_unit"],
},
}
为什么 Schema 很重要: description 字段是 LLM 决定何时使用你工具的依据。要具体说明它做什么和何时使用。parameters 定义 LLM 传递的参数。
第 4 步:编写工具处理器
创建 tools.py — 这是 LLM 调用你的工具时实际执行的代码:
"""工具处理器 — LLM 调用每个工具时运行的代码。"""
import json
import math
# 表达式计算的安全全局变量 — 无文件/网络访问
_SAFE_MATH = {
"abs": abs, "round": round, "min": min, "max": max,
"pow": pow, "sqrt": math.sqrt, "sin": math.sin, "cos": math.cos,
"tan": math.tan, "log": math.log, "log2": math.log2, "log10": math.log10,
"floor": math.floor, "ceil": math.ceil,
"pi": math.pi, "e": math.e,
"factorial": math.factorial,
}
def calculate(args: dict, **kwargs) -> str:
"""安全地计算数学表达式。
处理器规则:
1. 接收 args(dict)— LLM 传递的参数
2. 执行计算
3. 返回 JSON 字符串 — 始终如此,即使出错
4. 接受 **kwargs 以保持前向兼容
"""
expression = args.get("expression", "").strip()
if not expression:
return json.dumps({"error": "No expression provided"})
try:
result = eval(expression, {"__builtins__": {}}, _SAFE_MATH)
return json.dumps({"expression": expression, "result": result})
except ZeroDivisionError:
return json.dumps({"expression": expression, "error": "Division by zero"})
except Exception as e:
return json.dumps({"expression": expression, "error": f"Invalid: {e}"})
# 换算表 — 值以基本单位表示
_LENGTH = {"m": 1, "km": 1000, "mi": 1609.34, "ft": 0.3048, "in": 0.0254, "cm": 0.01}
_WEIGHT = {"kg": 1, "g": 0.001, "lb": 0.453592, "oz": 0.0283495}
_DATA = {"B": 1, "KB": 1024, "MB": 1024**2, "GB": 1024**3, "TB": 1024**4}
_TIME = {"s": 1, "ms": 0.001, "min": 60, "hr": 3600, "day": 86400}
def _convert_temp(value, from_u, to_u):
# 标准化为摄氏度
c = {"F": (value - 32) * 5/9, "K": value - 273.15}.get(from_u, value)
# 转换为目标单位
return {"F": c * 9/5 + 32, "K": c + 273.15}.get(to_u, c)
def unit_convert(args: dict, **kwargs) -> str:
"""单位间换算。"""
value = args.get("value")
from_unit = args.get("from_unit", "").strip()
to_unit = args.get("to_unit", "").strip()
if value is None or not from_unit or not to_unit:
return json.dumps({"error": "Need value, from_unit, and to_unit"})
try:
# 温度
if from_unit.upper() in {"C","F","K"} and to_unit.upper() in {"C","F","K"}:
result = _convert_temp(float(value), from_unit.upper(), to_unit.upper())
return json.dumps({"input": f"{value} {from_unit}", "result": round(result, 4),
"output": f"{round(result, 4)} {to_unit}"})
# 基于比率的换算
for table in (_LENGTH, _WEIGHT, _DATA, _TIME):
lc = {k.lower(): v for k, v in table.items()}
if from_unit.lower() in lc and to_unit.lower() in lc:
result = float(value) * lc[from_unit.lower()] / lc[to_unit.lower()]
return json.dumps({"input": f"{value} {from_unit}",
"result": round(result, 6),
"output": f"{round(result, 6)} {to_unit}"})
return json.dumps({"error": f"Cannot convert {from_unit} → {to_unit}"})
except Exception as e:
return json.dumps({"error": f"Conversion failed: {e}"})
处理器关键规则:
- 签名:
def my_handler(args: dict, **kwargs) -> str - 返回值: 始终是 JSON 字符串。成功和错误都是。
- 永不抛出异常: 捕获所有异常,改为返回错误 JSON。
- 接受
**kwargs: Hermes 未来可能传递额外的上下文。
第 5 步:编写注册代码
创建 __init__.py — 这将 Schema 连接到处理器:
"""Calculator 插件 — 注册。"""
import logging
from . import schemas, tools
logger = logging.getLogger(__name__)
# 通过 Hook 跟踪工具使用
_call_log = []
def _on_post_tool_call(tool_name, args, result, task_id, **kwargs):
"""Hook:在每次工具调用后运行(不仅限于我们的工具)。"""
_call_log.append({"tool": tool_name, "session": task_id})
if len(_call_log) > 100:
_call_log.pop(0)
logger.debug("Tool called: %s (session %s)", tool_name, task_id)
def register(ctx):
"""将 Schema 连接到处理器并注册 Hook。"""
ctx.register_tool(name="calculate", toolset="calculator",
schema=schemas.CALCULATE, handler=tools.calculate)
ctx.register_tool(name="unit_convert", toolset="calculator",
schema=schemas.UNIT_CONVERT, handler=tools.unit_convert)
# 此 Hook 对所有工具调用触发,不仅限于我们的
ctx.register_hook("post_tool_call", _on_post_tool_call)
register() 做什么:
- 启动时仅调用一次
ctx.register_tool()将你的工具放入注册表 — 模型立即可见ctx.register_hook()订阅生命周期事件ctx.register_cli_command()注册 CLI 子命令(如hermes my-plugin <subcommand>)- 如果此函数崩溃,插件被禁用但 Hermes 正常继续
第 6 步:测试
启动 Hermes:
hermes
你应该在 Banner 的工具列表中看到 calculator: calculate, unit_convert。
尝试这些提示:
2 的 16 次方是多少?
将 100 华氏度转换为摄氏度
2 的平方根乘以 pi 是多少?
1.5 太字节是多少千兆字节?
检查插件状态:
/plugins
输出:
Plugins (1):
✓ calculator v1.0.0 (2 tools, 1 hooks)
插件的最终结构
~/.hermes/plugins/calculator/
├── plugin.yaml # "我是 calculator,我提供工具和 Hook"
├── __init__.py # 连接:Schema → 处理器,注册 Hook
├── schemas.py # LLM 读取的内容(描述 + 参数规格)
└── tools.py # 运行的代码(calculate、unit_convert 函数)
四个文件,清晰分离:
- 清单文件声明插件是什么
- Schema 为 LLM 描述工具
- 处理器实现实际逻辑
- 注册代码连接一切
插件还能做什么?
附带数据文件
将任何文件放在插件目录中,在导入时读取:
# 在 tools.py 或 __init__.py 中
from pathlib import Path
_PLUGIN_DIR = Path(__file__).parent
_DATA_FILE = _PLUGIN_DIR / "data" / "languages.yaml"
with open(_DATA_FILE) as f:
_DATA = yaml.safe_load(f)
内嵌 Skill
插件可以附带 Skill 文件,Agent 通过 skill_view("plugin:skill") 加载。在 __init__.py 中注册:
~/.hermes/plugins/my-plugin/
├── __init__.py
├── plugin.yaml
└── skills/
├── my-workflow/
│ └── SKILL.md
└── my-checklist/
└── SKILL.md
from pathlib import Path
def register(ctx):
skills_dir = Path(__file__).parent / "skills"
for child in sorted(skills_dir.iterdir()):
skill_md = child / "SKILL.md"
if child.is_dir() and skill_md.exists():
ctx.register_skill(child.name, skill_md)
Agent 现在可以用带命名空间的名称加载你的 Skill:
skill_view("my-plugin:my-workflow") # → 插件的版本
skill_view("my-workflow") # → 内置版本(不变)
关键特性:
- 插件 Skill 是只读的 — 它们不会进入
~/.hermes/skills/,不能通过skill_manage编辑。 - 插件 Skill 不列在系统提示的
<available_skills>索引中 — 它们是选择性显式加载的。 - 裸 Skill 名称不受影响 — 命名空间防止与内置 Skill 冲突。
- 当 Agent 加载插件 Skill 时,会预置一个 Bundle 上下文横幅,列出同一插件的同级 Skill。
旧的 shutil.copy2 模式(将 Skill 复制到 ~/.hermes/skills/)仍然有效,但存在与内置 Skill 名称冲突的风险。新插件推荐使用 ctx.register_skill()。
基于环境变量控制加载
如果你的插件需要 API Key:
# plugin.yaml — 简单格式(向后兼容)
requires_env:
- WEATHER_API_KEY
如果 WEATHER_API_KEY 未设置,插件被禁用并显示清晰消息。不会崩溃,Agent 不会报错 — 只显示 "Plugin weather disabled (missing: WEATHER_API_KEY)"。
当用户运行 hermes plugins install 时,会交互式提示输入缺少的 requires_env 变量。值会自动保存到 .env。
为了更好的安装体验,使用带描述和注册 URL 的富格式:
# plugin.yaml — 富格式
requires_env:
- name: WEATHER_API_KEY
description: "OpenWeather 的 API Key"
url: "https://openweathermap.org/api"
secret: true
| 字段 | 必需 | 说明 |
|---|---|---|
name | 是 | 环境变量名称 |
description | 否 | 安装提示时向用户显示 |
url | 否 | 获取凭据的位置 |
secret | 否 | 如果为 true,输入时隐藏(类似密码字段) |
两种格式可以在同一列表中混用。已设置的变量会被静默跳过。
条件性工具可用性
对于依赖可选库的工具:
ctx.register_tool(
name="my_tool",
schema={...},
handler=my_handler,
check_fn=lambda: _has_optional_lib(), # False = 工具对模型隐藏
)
注册多个 Hook
def register(ctx):
ctx.register_hook("pre_tool_call", before_any_tool)
ctx.register_hook("post_tool_call", after_any_tool)
ctx.register_hook("pre_llm_call", inject_memory)
ctx.register_hook("on_session_start", on_new_session)
ctx.register_hook("on_session_end", on_session_end)
Hook 参考
每个 Hook 在 事件 Hook 参考 中有完整文档 — 回调签名、参数表、确切的触发时机和示例。以下是摘要:
| Hook | 触发时机 | 回调签名 | 返回值 |
|---|---|---|---|
pre_tool_call | 任何工具执行前 | tool_name: str, args: dict, task_id: str | 忽略 |
post_tool_call | 任何工具返回后 | tool_name: str, args: dict, result: str, task_id: str | 忽略 |
pre_llm_call | 每轮一次,工具调用循环之前 | session_id: str, user_message: str, conversation_history: list, is_first_turn: bool, model: str, platform: str | 上下文注入 |
post_llm_call | 每轮一次,工具调用循环之后(仅成功轮次) | session_id: str, user_message: str, assistant_response: str, conversation_history: list, model: str, platform: str | 忽略 |
on_session_start | 新会话创建时(仅首轮) | session_id: str, model: str, platform: str | 忽略 |
on_session_end | 每次 run_conversation 调用结束 + CLI 退出 | session_id: str, completed: bool, interrupted: bool, model: str, platform: str | 忽略 |
on_session_finalize | CLI/Gateway 拆除活跃会话 | session_id: str | None, platform: str | 忽略 |
on_session_reset | Gateway 切换新会话键(/new、/reset) | session_id: str, platform: str | 忽略 |
大多数 Hook 是即发即弃的观察者 — 返回值被忽略。例外是 pre_llm_call,它可以注入上下文到对话中。
所有回调应接受 **kwargs 以保持前向兼容。如果 Hook 回调崩溃,会被记录并跳过。其他 Hook 和 Agent 正常继续。
pre_llm_call 上下文注入
这是唯一返回值有意义的 Hook。当 pre_llm_call 回调返回包含 "context" 键的 dict(或纯字符串)时,Hermes 将该文本注入当前轮次的用户消息。这是记忆插件、RAG 集成、安全护栏和任何需要向模型提供额外上下文的插件的机制。
返回格式
# 带 context 键的 dict
return {"context": "召回的记忆:\n- 用户偏好深色模式\n- 上一个项目:hermes-agent"}
# 纯字符串(等同于上面的 dict 形式)
return "召回的记忆:\n- 用户偏好深色模式"
# 返回 None 或不返回 → 无注入(仅观察)
return None
任何非 None、非空且包含 "context" 键(或非空纯字符串)的返回值会被收集并附加到当前轮次的用户消息。
注入如何工作
注入的上下文附加到用户消息,而非系统提示。这是一个有意的设计选择:
- 提示缓存保留 — 系统提示在轮次间保持不变。Anthropic 和 OpenRouter 缓存系统提示前缀,因此保持稳定可在多轮对话中节省 75%+ 的输入 Token。如果插件修改系统提示,每轮都会是缓存未命中。
- 临时性 — 注入仅在 API 调用时发生。对话历史中的原始用户消息永远不会被修改,也不会有任何内容持久化到会话数据库。
- 系统提示是 Hermes 的领域 — 它包含模型特定的指导、工具执行规则、个性指令和缓存的 Skill 内容。插件通过与用户输入并行的方式贡献上下文,而非更改 Agent 的核心指令。
示例:记忆召回插件
"""记忆插件 — 从向量存储召回相关上下文。"""
import httpx
MEMORY_API = "https://your-memory-api.example.com"
def recall_context(session_id, user_message, is_first_turn, **kwargs):
"""在每次 LLM 轮次前调用。返回召回的记忆。"""
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 = "从之前会话召回的上下文:\n"
text += "\n".join(f"- {m['text']}" for m in memories)
return {"context": text}
except Exception:
return None # 静默失败,不中断 Agent
def register(ctx):
ctx.register_hook("pre_llm_call", recall_context)
示例:安全护栏插件
"""安全护栏插件 — 执行内容策略。"""
POLICY = """你必须遵循此会话的内容策略:
- 永远不要生成访问工作目录外文件系统的代码
- 执行破坏性操作前始终警告
- 拒绝涉及个人数据提取的请求"""
def inject_guardrails(**kwargs):
"""在每轮注入策略文本。"""
return {"context": POLICY}
def register(ctx):
ctx.register_hook("pre_llm_call", inject_guardrails)
示例:仅观察 Hook(无注入)
"""分析插件 — 跟踪轮次元数据但不注入上下文。"""
import logging
logger = logging.getLogger(__name__)
def log_turn(session_id, user_message, model, is_first_turn, **kwargs):
"""在每次 LLM 调用前触发。返回 None — 无上下文注入。"""
logger.info("Turn: session=%s model=%s first=%s msg_len=%d",
session_id, model, is_first_turn, len(user_message or ""))
# 无返回 → 无注入
def register(ctx):
ctx.register_hook("pre_llm_call", log_turn)
多个插件返回上下文
当多个插件从 pre_llm_call 返回上下文时,它们的输出以双换行符连接并一起附加到用户消息。顺序遵循插件发现顺序(按插件目录名字母排列)。
注册 CLI 命令
插件可以添加自己的 hermes <plugin> 子命令树:
def _my_command(args):
"""hermes my-plugin <subcommand> 的处理器。"""
sub = getattr(args, "my_command", None)
if sub == "status":
print("一切正常!")
elif sub == "config":
print("当前配置: ...")
else:
print("用法: hermes my-plugin <status|config>")
def _setup_argparse(subparser):
"""构建 hermes my-plugin 的 argparse 树。"""
subs = subparser.add_subparsers(dest="my_command")
subs.add_parser("status", help="显示插件状态")
subs.add_parser("config", help="显示插件配置")
subparser.set_defaults(func=_my_command)
def register(ctx):
ctx.register_tool(...)
ctx.register_cli_command(
name="my-plugin",
help="管理我的插件",
setup_fn=_setup_argparse,
handler_fn=_my_command,
)
注册后,用户可以运行 hermes my-plugin status、hermes my-plugin config 等。
记忆 Provider 插件使用基于约定的方式:在插件的 cli.py 文件中添加 register_cli(subparser) 函数。记忆插件发现系统会自动找到它 — 无需 ctx.register_cli_command() 调用。详见记忆 Provider 插件指南。
活跃 Provider 控制: 记忆插件 CLI 命令仅在其 Provider 是 config 中的活跃 memory.provider 时出现。如果用户未设置你的 Provider,你的 CLI 命令不会出现在帮助输出中。
注册斜杠命令
插件可以注册会话内斜杠命令 — 用户在对话中输入的命令(如 /lcm status 或 /ping)。这些在 CLI 和 Gateway(Telegram、Discord 等)中都有效。
def _handle_status(raw_args: str) -> str:
"""/mystatus 的处理器 — 用命令名之后的所有内容调用。"""
if raw_args.strip() == "help":
return "用法: /mystatus [help|check]"
return "插件状态:一切系统正常"
def register(ctx):
ctx.register_command(
"mystatus",
handler=_handle_status,
description="显示插件状态",
)
注册后,用户可以在任何会话中输入 /mystatus。命令出现在自动补全、/help 输出和 Telegram Bot 菜单中。
签名: ctx.register_command(name: str, handler: Callable, description: str = "")
| 参数 | 类型 | 说明 |
|---|---|---|
name | str | 不带前导斜杠的命令名(如 "lcm"、"mystatus") |
handler | Callable[[str], str | None] | 用原始参数字符串调用。也可以是 async。 |
description | str | 在 /help、自动补全和 Telegram Bot 菜单中显示 |
与 register_cli_command() 的关键区别:
register_command() | register_cli_command() | |
|---|---|---|
| 调用方式 | 会话中的 /name | 终端中的 hermes name |
| 工作位置 | CLI 会话、Telegram、Discord 等 | 仅终端 |
| 处理器接收 | 原始参数字符串 | argparse Namespace |
| 使用场景 | 诊断、状态、快捷操作 | 复杂子命令树、设置向导 |
冲突保护: 如果插件尝试注册与内置命令冲突的名称(help、model、new 等),注册会被静默拒绝并记录日志警告。内置命令始终优先。
异步处理器: Gateway 调度自动检测并 await 异步处理器,因此你可以使用同步或异步函数:
async def _handle_check(raw_args: str) -> str:
result = await some_async_operation()
return f"检查结果: {result}"
def register(ctx):
ctx.register_command("check", handler=_handle_check, description="运行异步检查")
本指南涵盖通用插件(工具、Hook、斜杠命令、CLI 命令)。对于专用插件类型,参见:
- 记忆 Provider 插件 — 跨会话知识后端
- 上下文引擎插件 — 替代上下文管理策略
通过 pip 分发
要公开分享插件,在 Python 包中添加入口点:
# pyproject.toml
[project.entry-points."hermes_agent.plugins"]
my-plugin = "my_plugin_package"
pip install hermes-plugin-calculator
# 下次 hermes 启动时自动发现插件
常见错误
处理器没有返回 JSON 字符串:
# 错误 — 返回 dict
def handler(args, **kwargs):
return {"result": 42}
# 正确 — 返回 JSON 字符串
def handler(args, **kwargs):
return json.dumps({"result": 42})
处理器签名缺少 **kwargs:
# 错误 — Hermes 传递额外上下文时会出错
def handler(args):
...
# 正确
def handler(args, **kwargs):
...
处理器抛出异常:
# 错误 — 异常传播,工具调用失败
def handler(args, **kwargs):
result = 1 / int(args["value"]) # ZeroDivisionError!
return json.dumps({"result": result})
# 正确 — 捕获并返回错误 JSON
def handler(args, **kwargs):
try:
result = 1 / int(args.get("value", 0))
return json.dumps({"result": result})
except Exception as e:
return json.dumps({"error": str(e)})
Schema 描述太模糊:
# 差 — 模型不知道何时使用它
"description": "做些事情"
# 好 — 模型确切知道何时和如何使用
"description": "计算数学表达式。用于算术、三角函数、对数。支持:+、-、*、/、**、sqrt、sin、cos、log、pi、e。"
📝 本文由 AI 翻译,如有疑问请参考英文原版