建立記憶體 Provider 外掛
記憶體 provider 外掛為 Hermes Agent 提供超出內建 MEMORY.md 和 USER.md 範圍的持久化跨 session 知識。本指南涵蓋如何建立一個。
提示
記憶體 provider 是兩種 provider 外掛類型之一。另一種是 Context Engine Plugins,用於取代內建的上下文壓縮器。兩者遵循相同的模式:單選、設定驅動、透過
hermes plugins管理。
目錄結構
每個記憶體 provider 位於 plugins/memory/<name>/:
plugins/memory/my-provider/
├── __init__.py # MemoryProvider 實作 + register() 入口點
├── plugin.yaml # 中繼資料(名稱、描述、鉤子)
└── README.md # 設定說明、設定參考、工具
MemoryProvider ABC
你的外掛實作來自 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:
"""檢查此 provider 是否可以啟動。不能有網路呼叫。"""
return bool(os.environ.get("MY_API_KEY"))
def initialize(self, session_id: str, **kwargs) -> None:
"""在代理程式啟動時呼叫一次。
kwargs 始終包含:
hermes_home (str): 使用中的 HERMES_HOME 路徑。用於儲存。
"""
self._api_key = os.environ.get("MY_API_KEY", "")
self._session_id = session_id
# ... 實作其餘方法
必要方法
核心生命週期
| 方法 | 呼叫時機 | 必須實作? |
|---|---|---|
name(屬性) | 始終 | 是 |
is_available() | 代理程式初始化,啟動前 | 是 — 不能有網路呼叫 |
initialize(session_id, **kwargs) | 代理程式啟動時 | 是 |
get_tool_schemas() | 初始化後,用於工具注入 | 是 |
handle_tool_call(tool_name, args, **kwargs) | 當代理程式使用你的工具時 | 是(如果你有工具) |
設定
| 方法 | 用途 | 必須實作? |
|---|---|---|
get_config_schema() | 為 hermes memory setup 宣告設定欄位 | 是 |
save_config(values, hermes_home) | 將非機密設定寫入原生位置 | 是(除非僅使用環境變數) |
選用鉤子
| 方法 | 呼叫時機 | 使用場景 |
|---|---|---|
system_prompt_block() | 系統提示詞組裝 | 靜態 provider 資訊 |
prefetch(query, *, session_id="") | 每次 API 呼叫前 | 回傳召回的上下文 |
queue_prefetch(query) | 每個回合後 | 為下一個回合預熱 |
sync_turn(user, assistant, *, session_id="") | 每個完成的回合後 | 持久化對話 |
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()。
提示 — 最小化 vs 完整 Schema
get_config_schema()中的每個欄位都會在hermes memory setup時進行提示。有很多選項的 provider 應該保持 schema 最小化 — 只包含使用者必須設定的欄位(API key、必要認證)。選用設定應記錄在設定檔參考中(例如$HERMES_HOME/myprovider.json),而不是在設定過程中全部提示。這使設定精靈保持快速同時仍然支援進階設定。參見 Supermemory provider 作為範例 — 它只提示 API key;所有其他選項都在supermemory.json中。
儲存設定
def save_config(self, values: dict, hermes_home: str) -> None:
"""將非機密設定寫入你的原生位置。"""
import json
from pathlib import Path
config_path = Path(hermes_home) / "my-provider.json"
config_path.write_text(json.dumps(values, indent=2))
對於僅使用環境變數的 provider,保持預設的無操作即可。
外掛入口點
def register(ctx) -> None:
"""由記憶體外掛發現系統呼叫。"""
ctx.register_memory_provider(MyMemoryProvider())
plugin.yaml
name: my-provider
version: 1.0.0
description: "此 provider 功能的簡短描述。"
hooks:
- on_session_end # 列出你實作的鉤子
執行緒契約
sync_turn() 必須是非阻塞的。 如果你的後端有延遲(API 呼叫、LLM 處理),請在守護執行緒中執行工作:
def sync_turn(self, user_content, assistant_content, *, session_id="", messages=None):
def _sync():
try:
self._api.ingest(user_content, assistant_content, session_id=session_id, messages=messages)
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()
messages 是可選的 OpenAI 風格對話上下文,來自已完成的回合。當提供時,它包含使用者/助理訊息、助理工具呼叫和工具結果訊息。不需要原始回合上下文的 provider 可以省略 messages 參數;Hermes 會繼續使用舊簽名呼叫它們。
雲端 provider 應該記錄 messages 的哪些部分會被發送到裝置外。工具呼叫和工具結果可能包含檔案路徑、命令輸出或其他工作區資料。
設定檔隔離
所有儲存路徑必須使用 initialize() 中的 hermes_home 參數,而不是硬編碼的 ~/.hermes:
# 正確 — 按設定檔範圍
from hermes_constants import get_hermes_home
data_dir = get_hermes_home() / "my-provider"
# 錯誤 — 所有設定檔共用
data_dir = Path("~/.hermes/my-provider").expanduser()
測試
端對端模式請見 tests/agent/test_memory_provider.py 和相鄰的記憶體測試(tests/agent/test_memory_session_switch.py、tests/agent/test_memory_user_id.py、tests/run_agent/test_memory_provider_init.py)。
from agent.memory_manager import MemoryManager
mgr = MemoryManager()
mgr.add_provider(my_provider)
mgr.initialize_all(session_id="test-1", platform="cli")
# 測試工具路由
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 命令
記憶體 provider 外掛可以註冊自己的 CLI 子命令樹(例如 hermes my-provider status、hermes my-provider config)。這使用基於慣例的發現系統 — 不需要修改核心檔案。
運作原理
- 在你的外掛目錄中新增一個
cli.py檔案 - 定義一個
register_cli(subparser)函式來建構 argparse 樹 - 記憶體外掛系統在啟動時透過
discover_plugin_cli_commands()發現它 - 你的命令會出現在
hermes <provider-name> <subcommand>下
啟用 provider 門控: 你的 CLI 命令只在你的 provider 是設定中啟用的 memory.provider 時才會出現。如果使用者尚未設定你的 provider,你的命令不會出現在 hermes --help 中。
範例
# plugins/memory/my-provider/cli.py
def my_command(args):
"""由 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:
"""建構 hermes my-provider argparse 樹。
在 argparse 設定時由 discover_plugin_cli_commands() 呼叫。
"""
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(subparser) — CLI 命令
└── README.md # 設定說明
單一 Provider 規則
同一時間只能有一個外部記憶體 provider 啟動。如果使用者嘗試註冊第二個,MemoryManager 會拒絕並發出警告。這防止了工具 schema 膨脹和衝突的後端。