hermes教程-构建一个 Hermes 插件

mkdir -p ~/.hermes/plugins/calculator
cd ~/.hermes/plugins/calculator

name: calculator
version: 1.0.0
description: 数学计算器 — 计算表达式和转换单位
provides_tools:
  - calculate
  - unit_convert
provides_hooks:
  - post_tool_call

author: 你的名字
requires_env:          # 根据环境变量控制加载；安装时提示
  - SOME_API_KEY       # 简单格式 — 如果缺失则禁用插件
  - name: OTHER_KEY    # 丰富格式 — 安装时显示描述/URL
    description: "其他服务的密钥"
    url: "https://other.com/keys"
    secret: true

"""工具模式 — 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"],
    },
}

"""工具处理器 — 当 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": "未提供表达式"})

    try:
        result = eval(expression, {"__builtins__": {}}, _SAFE_MATH)
        return json.dumps({"expression": expression, "result": result})
    except ZeroDivisionError:
        return json.dumps({"expression": expression, "error": "除以零"})
    except Exception as e:
        return json.dumps({"expression": expression, "error": f"无效: {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": "需要 value、from_unit 和 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"无法转换 {from_unit} → {to_unit}"})
    except Exception as e:
        return json.dumps({"error": f"转换失败: {e}"})

"""计算器插件 — 注册。"""

import logging

from . import schemas, tools

logger = logging.getLogger(__name__)
## 通过钩子跟踪工具使用情况
_call_log = []

def _on_post_tool_call(tool_name, args, result, task_id, **kwargs):
    """钩子：每次工具调用后运行（不仅限于我们的工具）。"""
    _call_log.append({"tool": tool_name, "session": task_id})
    if len(_call_log) > 100:
        _call_log.pop(0)
    logger.debug("工具被调用: %s (会话 %s)", tool_name, task_id)

def register(ctx):
    """将模式连接到处理器并注册钩子。"""
    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)
## 此钩子针对所有工具调用触发，不仅限于我们的
    ctx.register_hook("post_tool_call", _on_post_tool_call)

def handle_scan(ctx, raw_args: str):
    """通过注册表调用 terminal 工具来实现 /scan。"""
    result = ctx.dispatch_tool("terminal", {"command": f"find . -name '{raw_args}'"})
    return result  # 返回给调用者的聊天界面

def register(ctx):
## 处理器接收一个原始参数字符串；通过 lambda 闭包捕获 ctx。
    ctx.register_command(
        "scan",
        lambda raw: handle_scan(ctx, raw),
        description="查找匹配 glob 模式的文件",
    )

hermes

2 的 16 次方是多少？
将 100 华氏度转换为摄氏度
2 乘以 pi 的平方根是多少？
1.5 TB 是多少 GB？

插件 (1):
  ✓ calculator v1.0.0 (2 个工具, 1 个钩子)

HERMES_PLUGINS_DEBUG=1 hermes plugins list

hermes logs --level WARNING | grep -i plugin

~/.hermes/plugins/calculator/
├── plugin.yaml      # "我是 calculator，提供工具和钩子"
├── __init__.py      # 连接：模式 → 处理器，注册钩子
├── schemas.py       # LLM 读取的内容（描述 + 参数规范）
└── tools.py         # 运行的内容（calculate、unit_convert 函数）

## 在 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)

~/.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)

skill_view("my-plugin:my-workflow")   # → 插件的版本
skill_view("my-workflow")              # → 内置版本（不变）

## plugin.yaml — 简单格式（向后兼容）
requires_env:
  - WEATHER_API_KEY

## plugin.yaml — 丰富格式
requires_env:
  - name: WEATHER_API_KEY
    description: "OpenWeather 的 API 密钥"
    url: "https://openweathermap.org/api"
    secret: true

## tools.py
from tools.lazy_deps import ensure, FeatureUnavailable

def my_tool_handler(args, **kwargs):
    try:
        ensure("my-plugin.my-backend")   # 键必须位于 LAZY_DEPS 中
    except FeatureUnavailable as exc:
        return {"error": str(exc)}

    import my_backend_sdk   # 现在安全了
    ...

_client = None

def get_client():
    global _client
    if _client is not None:
        return _client
    _client = ExpensiveClient(...)   # ← TOCTOU 竞态
    return _client

from plugins.plugin_utils import lazy_singleton, SingletonSlot
## 零参数访问器 → 装饰它：
@lazy_singleton
def get_client():
    return ExpensiveClient(load_config())   # 恰好运行一次

client = get_client()    # 跨线程安全
get_client.reset()       # 丢弃实例（测试/拆卸）
## 接受构建参数的访问器 → 使用槽：
_slot: SingletonSlot = SingletonSlot()

def get_client(config=None):
    return _slot.get(lambda: ExpensiveClient(resolve(config)))

def reset_client():
    _slot.reset()

ctx.register_tool(
    name="my_tool",
    schema={...},
    handler=my_handler,
    check_fn=lambda: _has_optional_lib(),  # False = 对模型隐藏工具
)

def register(ctx):
    ctx.register_tool(
        name="browser_navigate",             # 与内置工具同名
        toolset="plugin_my_browser",         # 你自己的工具集命名空间
        schema={...},
        handler=my_custom_navigate,
        override=True,                       # 显式选择加入
    )

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)

## 带有 context 键的字典
return {"context": "回忆的记忆:\n- 用户偏好深色模式\n- 上一个项目: hermes-agent"}
## 纯字符串（等同于上面的字典形式）
return "回忆的记忆:\n- 用户偏好深色模式"
## 返回 None 或不返回 → 不注入（仅观察者）
return None

"""内存插件 — 从向量存储中回忆相关上下文。"""

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  # 静默失败，不要破坏代理

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)

"""分析插件 — 跟踪轮次元数据而不注入上下文。"""

import logging
logger = logging.getLogger(__name__)

def log_turn(session_id, user_message, model, is_first_turn, **kwargs):
    """在每次 LLM 调用之前触发。返回 None — 不注入上下文。"""
    logger.info("轮次: 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)

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,
    )

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="显示插件状态",
    )

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="运行异步检查")

def register(ctx):
    def _handle_deliver(raw_args: str):
        result = ctx.dispatch_tool(
            "delegate_task",
            {
                "goal": raw_args,
                "toolsets": ["terminal", "file", "web"],
            },
        )
        return result

    ctx.register_command(
        "deliver",
        handler=_handle_deliver,
        description="将目标委托给子代理",
    )

def register(ctx):
    async def _on_approve(ack, body, action):
## 在 3 秒内 ack — slack_bolt 要求。
        await ack()
## body["channel"]["id"], body["user"]["id"], body["message"]["ts"]
## action["action_id"], action["value"]
        sweep_id = (action.get("value") or "").split("|", 1)[-1]
## ...执行确定性工作，然后发布后续消息。

    ctx.register_slack_action_handler("inbox_sweep_approve", _on_approve)

## plugins/model-providers/acme/__init__.py
from providers import register_provider
from providers.base import ProviderProfile

register_provider(ProviderProfile(
    name="acme",
    aliases=("acme-inference",),
    display_name="Acme 推理",
    env_vars=("ACME_API_KEY", "ACME_BASE_URL"),
    base_url="https://api.acme.example.com/v1",
    auth_type="api_key",
    default_aux_model="acme-small-fast",
    fallback_models=("acme-large-v3", "acme-medium-v3"),
))

## plugins/model-providers/acme/plugin.yaml
name: acme-provider
kind: model-provider
version: 1.0.0
description: Acme 推理 — 兼容 OpenAI 的直接 API

## plugins/platforms/myplatform/adapter.py
from gateway.platforms.base import BasePlatformAdapter

class MyPlatformAdapter(BasePlatformAdapter):
    async def connect(self): ...
    async def send(self, chat_id, text): ...
    async def disconnect(self): ...

def check_requirements():
    import os
    return bool(os.environ.get("MYPLATFORM_TOKEN"))

def _env_enablement():
    import os
    tok = os.getenv("MYPLATFORM_TOKEN", "").strip()
    if not tok:
        return None
    return {"token": tok}

def register(ctx):
    ctx.register_platform(
        name="myplatform",
        label="MyPlatform",
        adapter_factory=lambda cfg: MyPlatformAdapter(cfg),
        check_fn=check_requirements,
        required_env=["MYPLATFORM_TOKEN"],
## 自动从环境填充 PlatformConfig.extra，以便仅环境设置
## 无需 SDK 实例化即可显示在 `hermes gateway status` 中。
        env_enablement_fn=_env_enablement,
## 选择 cron 投递：`deliver=myplatform` 路由到此变量。
        cron_deliver_env_var="MYPLATFORM_HOME_CHANNEL",
        emoji="💬",
        platform_hint="你正在通过 MyPlatform 聊天。保持回复简洁。",
    )

## plugins/platforms/myplatform/plugin.yaml
name: myplatform-platform
label: MyPlatform
kind: platform
version: 1.0.0
description: MyPlatform 网关适配器
requires_env:
  - name: MYPLATFORM_TOKEN
    description: "来自 MyPlatform 控制台的机器人令牌"
    password: true
optional_env:
  - name: MYPLATFORM_HOME_CHANNEL
    description: "cron 投递的默认频道"
    password: false

## plugins/memory/my-memory/__init__.py
from agent.memory_provider import MemoryProvider

class MyMemoryProvider(MemoryProvider):
    @property
    def name(self) -> str:
        return "my-memory"

    def is_available(self) -> bool:
        import os
        return bool(os.environ.get("MY_MEMORY_API_KEY"))

    def initialize(self, session_id: str, **kwargs) -> None:
        self._session_id = session_id

    def sync_turn(self, user_content, assistant_content, *,
                  session_id="", messages=None) -> None:
        ...

    def prefetch(self, query, *, session_id="") -> str:
        ...

    def get_tool_schemas(self) -> list[dict]:
        return []   # 必需的 @abstractmethod — 参见完整指南

def register(ctx):
    ctx.register_memory_provider(MyMemoryProvider())

## plugins/context_engine/my-engine/__init__.py
from agent.context_engine import ContextEngine

class MyContextEngine(ContextEngine):
    @property
    def name(self) -> str:
        return "my-engine"

    def update_from_response(self, usage) -> None: ...
    def should_compress(self, prompt_tokens: int = None) -> bool: ...
    def compress(self, messages, current_tokens=None, focus_topic=None) -> list: ...

def register(ctx):
    ctx.register_context_engine(MyContextEngine())

## plugins/image_gen/my-imggen/__init__.py
from agent.image_gen_provider import ImageGenProvider

class MyImageGenProvider(ImageGenProvider):
    @property
    def name(self) -> str:
        return "my-imggen"

    def is_available(self) -> bool: ...
    def generate(self, prompt: str, aspect_ratio="landscape", **kwargs) -> dict:
## 返回 success_response(...) / error_response(...)
        ...

def register(ctx):
    ctx.register_image_gen_provider(MyImageGenProvider())

## plugins/image_gen/my-imggen/plugin.yaml
name: my-imggen
kind: backend
version: 1.0.0
description: 自定义图像生成后端

mcp_servers:
  filesystem:
    command: "npx"
    args: ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/projects"]
    timeout: 120

  linear:
    url: "https://mcp.linear.app/sse"
    auth:
      type: "oauth"

## ~/.hermes/hooks/long-task-alert/HOOK.yaml
name: long-task-alert
description: 长时间任务完成时发送推送通知
events:
  - agent:end

## ~/.hermes/hooks/long-task-alert/handler.py
async def handle(event_type: str, context: dict) -> None:
    if context.get("duration_seconds", 0) > 120:
## 发送通知 …
        pass

hooks:
  - event: post_tool_call
    command: "notify-send '工具运行: {tool_name}'"
    when:
      tools: [terminal, patch, write_file]

hermes skills tap add myorg/skills-repo
hermes skills search my-workflow --source myorg/skills-repo
hermes skills install myorg/skills-repo/my-workflow

tts:
  provider: voxcpm
  providers:
    voxcpm:
      type: command
      command: "voxcpm --ref ~/voice.wav --text-file {input_path} --out {output_path}"
      output_format: mp3
      voice_compatible: true

## pyproject.toml
[project.entry-points."hermes_agent.plugins"]
my-plugin = "my_plugin_package"

pip install hermes-plugin-calculator
## 下次启动 Hermes 时自动发现插件

## 用户的 configuration.nix
services.hermes-agent.extraPythonPackages = [
  (pkgs.python312Packages.buildPythonPackage {
    pname = "my-plugin";
    version = "1.0.0";
    src = pkgs.fetchFromGitHub {
      owner = "you";
      repo = "hermes-my-plugin";
      rev = "v1.0.0";
      hash = "sha256-...";  # nix-prefetch-url --unpack
    };
    format = "pyproject";
    build-system = [ pkgs.python312Packages.setuptools ];
  })
];

services.hermes-agent.extraPlugins = [
  (pkgs.fetchFromGitHub {
    owner = "you";
    repo = "hermes-my-plugin";
    rev = "v1.0.0";
    hash = "sha256-...";
  })
];

## 错误 — 返回字典
def handler(args, **kwargs):
    return {"result": 42}
## 正确 — 返回 JSON 字符串
def handler(args, **kwargs):
    return json.dumps({"result": 42})

## 错误 — 如果 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)})

## 不好 — 模型不知道何时使用
"description": "做事情"
## 好 — 模型确切知道何时以及如何
"description": "计算数学表达式。用于算术、三角、对数。支持: +, -, *, /, **, sqrt, sin, cos, log, pi, e。"