构建一个 Hermes 插件
本指南将带你从零开始构建一个完整的 Hermes 插件。完成之后,你将拥有一个功能齐全的插件,包含多个工具、生命周期钩子、已打包的数据文件以及一个内置技能文件——涵盖了插件系统支持的所有功能。
你将构建的内容
一个 计算器 插件,包含两个工具:
calculate—— 计算数学表达式(2**16,sqrt(144),pi * 5**2)unit_convert—— 单位转换(100 F → 37.78 C,5 km → 3.11 mi)
此外还包括一个在每次工具调用时记录日志的钩子,以及一个打包的技能文件。
第一步:创建插件目录
mkdir -p ~/.hermes/plugins/calculator
cd ~/.hermes/plugins/calculator
第二步:编写清单文件
创建 plugin.yaml:
name: calculator
version: 1.0.0
description: Math calculator — evaluate expressions and convert units
provides_tools:
- calculate
- unit_convert
provides_hooks:
- post_tool_call
这告诉 Hermes:“我是一个名为 calculator 的插件,我提供工具和钩子。”provides_tools 和 provides_hooks 字段列出了插件注册的内容。
可选字段(你可以添加):
author: Your Name
requires_env: # gate loading on env vars; prompted during install
- SOME_API_KEY # simple format — plugin disabled if missing
- name: OTHER_KEY # rich format — shows description/url during install
description: "Key for the Other service"
url: "https://other.com/keys"
secret: true
第三步:编写工具 Schema
创建 schemas.py —— 这是 LLM 用来决定何时调用你的工具的依据:
"""Tool schemas — what the LLM sees."""
CALCULATE = {
"name": "calculate",
"description": (
"Evaluate a mathematical expression and return the result. "
"Supports arithmetic (+, -, *, /, **), functions (sqrt, sin, cos, "
"log, abs, round, floor, ceil), and constants (pi, e). "
"Use this for any math the user asks about."
),
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "Math expression to evaluate (e.g., '2**10', 'sqrt(144)')",
},
},
"required": ["expression"],
},
}
UNIT_CONVERT = {
"name": "unit_convert",
"description": (
"Convert a value between units. Supports length (m, km, mi, ft, in), "
"weight (kg, lb, oz, g), temperature (C, F, K), data (B, KB, MB, GB, TB), "
"and time (s, min, hr, day)."
),
"parameters": {
"type": "object",
"properties": {
"value": {
"type": "number",
"description": "The numeric value to convert",
},
"from_unit": {
"type": "string",
"description": "Source unit (e.g., 'km', 'lb', 'F', 'GB')",
},
"to_unit": {
"type": "string",
"description": "Target unit (e.g., 'mi', 'kg', 'C', 'MB')",
},
},
"required": ["value", "from_unit", "to_unit"],
},
}
Schema 为何重要:
description 字段是 LLM 决定是否使用你的工具的关键。请明确描述其功能以及使用场景。parameters 定义了 LLM 传递给工具的参数。
第四步:编写工具处理器
创建 tools.py —— 这是当 LLM 调用工具时实际执行的代码:
"""Tool handlers — the code that runs when the LLM calls each tool."""
import json
import math
# Safe globals for expression evaluation — no file/network access
_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:
"""Evaluate a math expression safely.
Rules for handlers:
1. Receive args (dict) — the parameters the LLM passed
2. Do the work
3. Return a JSON string — ALWAYS, even on error
4. Accept **kwargs for forward compatibility
"""
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}"})
# Conversion tables — values are in base units
_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):
# Normalize to Celsius
c = {"F": (value - 32) * 5/9, "K": value - 273.15}.get(from_u, value)
# Convert to target
return {"F": c * 9/5 + 32, "K": c + 273.15}.get(to_u, c)
def unit_convert(args: dict, **kwargs) -> str:
"""Convert between units."""
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:
# Temperature
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}"})
# Ratio-based conversions
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 未来可能会传递额外上下文。
第五步:编写注册逻辑
创建 __init__.py —— 这是将 Schema 与处理器连接起来的文件:
"""Calculator plugin — registration."""
import logging
from . import schemas, tools
logger = logging.getLogger(__name__)
# Track tool usage via hooks
_call_log = []
def _on_post_tool_call(tool_name, args, result, task_id, **kwargs):
"""Hook: runs after every tool call (not just ours)."""
_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):
"""Wire schemas to handlers and register hooks."""
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)
# This hook fires for ALL tool calls, not just ours
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 仍能正常运行
第六步:测试插件
启动 Hermes:
hermes
你应该在启动横幅的工具列表中看到 calculator: calculate, unit_convert。
尝试以下提示:
What's 2 to the power of 16?
Convert 100 fahrenheit to celsius
What's the square root of 2 times pi?
How many gigabytes is 1.5 terabytes?
检查插件状态:
/plugins
输出结果:
Plugins (1):
✓ calculator v1.0.0 (2 tools, 1 hooks)
你的插件最终结构
~/.hermes/plugins/calculator/
├── plugin.yaml # "I'm calculator, I provide tools and hooks"
├── __init__.py # Wiring: schemas → handlers, register hooks
├── schemas.py # What the LLM reads (descriptions + parameter specs)
└── tools.py # What runs (calculate, unit_convert functions)
四个文件,职责清晰分离:
- 清单文件:声明插件的身份
- Schema 文件:向 LLM 描述工具
- 处理器文件:实现实际逻辑
- 注册文件:连接所有组件
插件还能做什么?
打包数据文件
将任意文件放入插件目录,并在导入时读取:
# In tools.py or __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.md 文件,并在注册时安装:
import shutil
from pathlib import Path
def _install_skill():
"""Copy our skill to ~/.hermes/skills/ on first load."""
try:
from hermes_cli.config import get_hermes_home
dest = get_hermes_home() / "skills" / "my-plugin" / "SKILL.md"
except Exception:
dest = Path.home() / ".hermes" / "skills" / "my-plugin" / "SKILL.md"
if dest.exists():
return # don't overwrite user edits
source = Path(__file__).parent / "skill.md"
if source.exists():
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, dest)
def register(ctx):
ctx.register_tool(...)
_install_skill()
基于环境变量启用/禁用
如果插件需要 API 密钥:
# plugin.yaml — simple format (backwards-compatible)
requires_env:
- WEATHER_API_KEY
如果未设置 WEATHER_API_KEY,插件将被禁用,并显示清晰提示。不会崩溃,也不会导致代理出错——只会显示“Plugin weather disabled (missing: WEATHER_API_KEY)”。
当用户运行 hermes plugins install 时,系统会 交互式提示 输入任何缺失的 requires_env 变量。值会自动保存到 .env 文件中。
为了获得更好的安装体验,可以使用带描述和注册链接的丰富格式:
# plugin.yaml — rich format
requires_env:
- name: WEATHER_API_KEY
description: "API key for OpenWeather"
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 = tool hidden from model
)
注册多个钩子
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)
钩子参考
每个钩子的完整文档请参见 事件钩子参考 —— 包括回调签名、参数表、触发时机以及示例。以下是摘要:
| 钩子 | 触发时机 | 回调签名 | 返回值 |
|---|---|---|---|
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 | 忽略 |
pre_api_request | 每次向 LLM 提供商发起 HTTP 请求前 | method: str, url: str, headers: dict, body: dict | 忽略 |
post_api_request | 每次从 LLM 提供商收到 HTTP 响应后 | method: str, url: str, status_code: int, response: dict | 忽略 |
大多数钩子都是“触发即丢弃”的观察者——它们的返回值被忽略。例外是 pre_llm_call,它可以向对话中注入上下文。
所有回调都应接受 **kwargs 以保证向前兼容性。如果某个钩子回调崩溃,它会被记录并跳过,其他钩子和代理将继续正常运行。
pre_llm_call 上下文注入
这是唯一一个返回值有意义的钩子。当 pre_llm_call 回调返回一个包含 "context" 键的字典(或一个非空字符串)时,Hermes 会将该文本注入到当前轮次的用户消息中。这是实现记忆插件、RAG 集成、安全护栏以及任何需要向模型提供额外上下文的插件的机制。
返回格式
# Dict with context key
return {"context": "Recalled memories:\n- User prefers dark mode\n- Last project: hermes-agent"}
# Plain string (equivalent to the dict form above)
return "Recalled memories:\n- User prefers dark mode"
# Return None or don't return → no injection (observer-only)
return None
任何非 None、非空的返回值,若包含 "context" 键(或为非空字符串),都会被收集并附加到当前轮次的用户消息中。
注入机制说明
注入的上下文是附加到用户消息,而非系统提示。这是有意为之的设计选择:
- 提示缓存保留 —— 系统提示在各轮次中保持一致。Anthropic 和 OpenRouter 会缓存系统提示前缀,保持其稳定可节省多轮对话中 75% 以上的输入 token。如果插件修改了系统提示,每轮都会导致缓存未命中。
- 瞬时性 —— 注入仅在 API 调用时发生。对话历史中的原始用户消息永远不会被修改,也不会持久化到会话数据库。
- 系统提示属于 Hermes 的领域 —— 系统提示包含模型特定的指导、工具强制规则、人格指令和缓存的技能内容。插件应与用户输入一同提供上下文,而非通过修改代理的核心指令。
示例:记忆召回插件
"""Memory plugin — recalls relevant context from a vector store."""
import httpx
MEMORY_API = "https://your-memory-api.example.com"
def recall_context(session_id, user_message, is_first_turn, **kwargs):
"""Called before each LLM turn. Returns recalled memories."""
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 # nothing to inject
text = "Recalled context from previous sessions:\n"
text += "\n".join(f"- {m['text']}" for m in memories)
return {"context": text}
except Exception:
return None # fail silently, don't break the agent
def register(ctx):
ctx.register_hook("pre_llm_call", recall_context)
示例:安全护栏插件
"""Guardrails plugin — enforces content policies."""
POLICY = """You MUST follow these content policies for this session:
- Never generate code that accesses the filesystem outside the working directory
- Always warn before executing destructive operations
- Refuse requests involving personal data extraction"""
def inject_guardrails(**kwargs):
"""Injects policy text into every turn."""
return {"context": POLICY}
def register(ctx):
ctx.register_hook("pre_llm_call", inject_guardrails)
示例:仅观察型钩子(无注入)
"""Analytics plugin — tracks turn metadata without injecting context."""
import logging
logger = logging.getLogger(__name__)
def log_turn(session_id, user_message, model, is_first_turn, **kwargs):
"""Fires before each LLM call. Returns None — no context injected."""
logger.info("Turn: session=%s model=%s first=%s msg_len=%d",
session_id, model, is_first_turn, len(user_message or ""))
# No return → no injection
def register(ctx):
ctx.register_hook("pre_llm_call", log_turn)
多个插件同时返回上下文
当多个插件从 pre_llm_call 返回上下文时,它们的输出将通过双换行符连接,并一同附加到用户消息中。顺序遵循插件发现顺序(按插件目录名称的字母顺序)。
注册 CLI 命令
插件可以添加自己的 hermes <plugin> 子命令树:
def _my_command(args):
"""Handler for hermes my-plugin <subcommand>."""
sub = getattr(args, "my_command", None)
if sub == "status":
print("All good!")
elif sub == "config":
print("Current config: ...")
else:
print("Usage: hermes my-plugin <status|config>")
def _setup_argparse(subparser):
"""Build the argparse tree for hermes my-plugin."""
subs = subparser.add_subparsers(dest="my_command")
subs.add_parser("status", help="Show plugin status")
subs.add_parser("config", help="Show plugin config")
subparser.set_defaults(func=_my_command)
def register(ctx):
ctx.register_tool(...)
ctx.register_cli_command(
name="my-plugin",
help="Manage my plugin",
setup_fn=_setup_argparse,
handler_fn=_my_command,
)
注册后,用户可以运行 hermes my-plugin status、hermes my-plugin config 等命令。
记忆提供者插件采用基于约定的方式:在插件的 cli.py 文件中添加 register_cli(subparser) 函数。记忆插件发现系统会自动找到它——无需调用 ctx.register_cli_command()。详情请参见 记忆提供者插件指南。
激活提供者控制:记忆插件的 CLI 命令仅在配置中设置为活动 memory.provider 时才会显示。如果用户未配置你的提供者,你的 CLI 命令不会出现在帮助输出中,避免干扰。
通过 pip 发布
对于公开共享插件,请在您的 Python 包中添加入口点:
# pyproject.toml
[project.entry-points."hermes_agent.plugins"]
my-plugin = "my_plugin_package"
pip install hermes-plugin-calculator
# Plugin auto-discovered on next hermes startup
常见错误
处理器未返回 JSON 字符串:
# Wrong — returns a dict
def handler(args, **kwargs):
return {"result": 42}
# Right — returns a JSON string
def handler(args, **kwargs):
return json.dumps({"result": 42})
处理器签名中缺少 **kwargs:
# Wrong — will break if Hermes passes extra context
def handler(args):
...
# Right
def handler(args, **kwargs):
...
处理器抛出异常:
# Wrong — exception propagates, tool call fails
def handler(args, **kwargs):
result = 1 / int(args["value"]) # ZeroDivisionError!
return json.dumps({"result": result})
# Right — catch and return error 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)})
模式描述过于模糊:
# Bad — model doesn't know when to use it
"description": "Does stuff"
# Good — model knows exactly when and how
"description": "Evaluate a mathematical expression. Use for arithmetic, trig, logarithms. Supports: +, -, *, /, **, sqrt, sin, cos, log, pi, e."