構建記憶提供者插件
記憶提供者插件為 Hermes Agent 提供持久化、跨會話的知識,超越內置的 MEMORY.md 和 USER.md。本指南將介紹如何構建一個記憶提供者插件。
記憶提供者是兩種 提供者插件 類型之一。另一種是 上下文引擎插件,用於替換內置的上下文壓縮器。兩者遵循相同的模式:單選、配置驅動,並通過 hermes plugins 進行管理。
目錄結構
每個記憶提供者都位於 plugins/memory/<name>/ 目錄下:
plugins/memory/my-provider/
├── __init__.py # MemoryProvider 實現 + register() 入口點
├── plugin.yaml # 元數據(名稱、描述、掛鉤)
└── README.md # 設置說明、配置參考、tools
MemoryProvider 抽象基類
你的插件需實現 agent/memory_provider.py 中的 MemoryProvider 抽象基類:
from agent.memory_provider import MemoryProvider
class MyMemoryProvider(MemoryProvider):
@property
def name(self) -> str:
return "my-provider"
def is_available(self) -> bool:
"""Check if this provider can activate. NO network calls."""
return bool(os.environ.get("MY_API_KEY"))
def initialize(self, session_id: str, **kwargs) -> None:
"""Called once at agent startup.
kwargs always includes:
hermes_home (str): Active HERMES_HOME path. Use for storage.
"""
self._api_key = os.environ.get("MY_API_KEY", "")
self._session_id = session_id
# ...實現剩餘的方法
必需方法
核心生命週期
| 方法 | 調用時機 | 是否必須實現? |
|---|---|---|
name (屬性) | 始終調用 | 是 |
is_available() | Agent 初始化,激活前 | 是 —— 不得進行網絡調用 |
initialize(session_id, **kwargs) | Agent 啟動時 | 是 |
get_tool_schemas() | 初始化後,用於工具注入 | 是 |
handle_tool_call(name, args) | Agent 使用你的工具時 | 是(如果你有工具) |
配置
| 方法 | 用途 | 是否必須實現? |
|---|---|---|
get_config_schema() | 聲明 hermes memory setup 使用的配置字段 | 是 |
save_config(values, hermes_home) | 將非敏感配置寫入原生位置 | 是(除非僅使用環境變量) |
可選鉤子
| 方法 | 調用時機 | 使用場景 |
|---|---|---|
system_prompt_block() | 系統提示詞組裝時 | 提供靜態提供者信息 |
prefetch(query) | 每次 API 調用前 | 返回召回的上下文 |
queue_prefetch(query) | 每輪對話後 | 為下一輪預熱 |
sync_turn(user, assistant) | 每輪對話完成後 | 持久化對話內容 |
on_session_end(messages) | 會話結束時 | 最終提取/刷新 |
on_pre_compress(messages) | 上下文壓縮前 | 在丟棄前保存洞察 |
on_memory_write(action, target, content) | 內置記憶寫入時 | 將變更同步到你的後端 |
shutdown() | 進程退出時 | 清理連接 |
配置 Schema
get_config_schema() 返回一個字段描述列表,由 hermes memory setup 使用:
def get_config_schema(self):
return [
{
"key": "api_key",
"description": "My Provider API key",
"secret": True, # → 寫入`.env`
"required": True,
"env_var": "MY_API_KEY", # 顯式環境變量名稱
"url": "https://my-provider.com/keys", # 在哪裡得到它
},
{
"key": "region",
"description": "Server region",
"default": "us-east",
"choices": ["us-east", "eu-west", "ap-south"],
},
{
"key": "project",
"description": "Project identifier",
"default": "hermes",
},
]
secret: True 且帶有 env_var 的字段將寫入 .env 文件。非敏感字段將傳遞給 save_config()。
get_config_schema() 中的每個字段都會在 hermes memory setup 時被提示。具有大量選項的提供者應保持 Schema 儘可能簡潔——僅包含用戶必須配置的字段(如 API 密鑰、必需憑證)。可選設置應記錄在配置文件參考中(例如 $HERMES_HOME/myprovider.json),而不是在設置嚮導中全部提示。這能保持設置嚮導快速響應,同時仍支持高級配置。參見 Supermemory 提供者示例——它僅提示 API 密鑰;所有其他選項均位於 supermemory.json 中。
保存配置
def save_config(self, values: dict, hermes_home: str) -> None:
"""Write non-secret config to your native location."""
import json
from pathlib import Path
config_path = Path(hermes_home) / "my-provider.json"
config_path.write_text(json.dumps(values, indent=2))
對於僅使用環境變量的提供者,可保留默認的空操作(no-op)。
插件入口點
def register(ctx) -> None:
"""Called by the memory plugin discovery system."""
ctx.register_memory_provider(MyMemoryProvider())
plugin.yaml
name: my-provider
version: 1.0.0
description: "Short description of what this provider does."
hooks:
- on_session_end # 列出您實現的鉤子
線程約定
sync_turn() 必須是非阻塞的。 如果你的後端存在延遲(如 API 調用、LLM 處理),請在守護線程中運行該工作:
def sync_turn(self, user_content, assistant_content):
def _sync():
try:
self._api.ingest(user_content, assistant_content)
except Exception as e:
logger.warning("Sync failed: %s", e)
if self._sync_thread and self._sync_thread.is_alive():
self._sync_thread.join(timeout=5.0)
self._sync_thread = threading.Thread(target=_sync, daemon=True)
self._sync_thread.start()
配置文件隔離
所有存儲路徑必須使用 initialize() 中的 hermes_home 參數,不得硬編碼 ~/.hermes:
# CORRECT — profile 範圍
from hermes_constants import get_hermes_home
data_dir = get_hermes_home() / "my-provider"
# WRONG — 在所有 profiles 之間共享
data_dir = Path("~/.hermes/my-provider").expanduser()
測試
參見 tests/agent/test_memory_plugin_e2e.py,其中包含使用真實 SQLite 提供者的完整端到端測試模式。
from agent.memory_manager import MemoryManager
mgr = MemoryManager()
mgr.add_provider(my_provider)
mgr.initialize_all(session_id="test-1", platform="cli")
# 測試tool路由
result = mgr.handle_tool_call("my_tool", {"action": "add", "content": "test"})
# 測試生命週期
mgr.sync_all("user msg", "assistant msg")
mgr.on_session_end([])
mgr.shutdown_all()
添加 CLI 命令
記憶提供者插件可以註冊自己的 CLI 子命令樹(例如 hermes my-provider status、hermes my-provider config)。該功能使用基於約定的發現機制——無需修改核心文件。
工作原理
- 在插件目錄中添加一個
cli.py文件 - 定義一個
register_cli(subparser)函數,用於構建 argparse 樹 - 記憶提供者系統通過
discover_plugin_cli_commands()在啟動時發現它 - 你的命令將出現在
hermes <provider-name> <subcommand>下
激活提供者控制: 只有當你的提供者是配置中的活動 memory.provider 時,你的 CLI 命令才會顯示。如果用戶未配置你的提供者,你的命令將不會出現在 hermes --help 中。
示例
# 插件/memory/my-provider/cli.py
def my_command(args):
"""Handler dispatched by argparse."""
sub = getattr(args, "my_command", None)
if sub == "status":
print("Provider is active and connected.")
elif sub == "config":
print("Showing config...")
else:
print("Usage: hermes my-provider <status|config>")
def register_cli(subparser) -> None:
"""Build the hermes my-provider argparse tree.
Called by discover_plugin_cli_commands() at argparse setup time.
"""
subs = subparser.add_subparsers(dest="my_command")
subs.add_parser("status", help="Show provider status")
subs.add_parser("config", help="Show provider config")
subparser.set_defaults(func=my_command)
參考實現
參見 plugins/memory/honcho/cli.py,其中包含一個完整示例,包含 13 個子命令、跨配置文件管理(--target-profile)以及配置讀寫功能。
帶 CLI 的目錄結構
plugins/memory/my-provider/
├── __init__.py # MemoryProvider 實現 + register()
├── plugin.yaml # 元數據
├── cli.py # register_cli(子解析器) — CLI 命令
└── README.md # 設置說明
單提供者規則
同一時間只能激活一個外部記憶提供者。如果用戶嘗試註冊第二個,MemoryManager 會以警告拒絕。這可防止工具 Schema 膨脹和後端衝突。