| name | hermes-memory-manager |
| description | Single-integration-point memory orchestrator — delegates to multiple providers, one external allowed, background async sync so slow providers never block a turn. Lifecycle hooks for compression/session-switch. Source: NousResearch/hermes-agent (MIT). |
Implementation (real, runnable — added 2026-06-19)
- Module:
core/lib/hermes_adapted/memory_manager.py (registration, prefetch/sync)
- Lifecycle hooks (on_turn_start/on_session_end/on_session_switch/
on_pre_compress/on_memory_write/on_delegation) + tool-call dispatch
(get_all_tool_schemas/handle_tool_call/flush_pending) mixed in from
core/lib/hermes_adapted/memory_manager_lifecycle.py
- Tests:
tests/test_hermes_memory_manager.py (14 passing)
MemoryManager is condensed from the original (hermes' MemoryProvider ABC
and reserved-tool-name set replaced with a plain duck-typed Protocol and
an injectable reserved_tool_names set), but the real invariants carry
over: exactly one external provider at a time, prefetch/sync run off the
calling thread on a single-worker executor (ordering preserved), bounded-
timeout shutdown so a wedged provider can't block teardown, and a failing
provider never breaks the hook for every other provider.
/hermes-memory-manager
When to Use
- Building an agent that connects to multiple memory backends (built-in L1 + external vector store)
- Need memory sync to not block the turn response (background write)
- Switching between memory providers without rewriting all dependent code
- Any agent memory layer that needs provider isolation and lifecycle hooks
Do NOT use for
- Single-provider memory with no switching needs — direct integration is simpler
- Synchronous memory that must complete before next turn (use direct await)
- See also: [[hermes-streaming-scrubber]] for keeping memory blocks out of UI
Design Principles
1. One integration point — run_agent.py calls MemoryManager, not each provider directly
2. One external provider — prevents tool schema bloat; multiple built-ins ok
3. Background sync — sync_all() runs in a separate worker thread; turn never waits
4. Ordered prompt assembly — built-in providers first, then external
5. Reserved tool names — external providers cannot shadow core tool names
Provider Interface
from abc import ABC, abstractmethod
from concurrent.futures import ThreadPoolExecutor
class MemoryProvider(ABC):
"""Implement this to add a memory backend."""
@property
@abstractmethod
def name(self) -> str: ...
@property
def tool_names(self) -> list[str]:
"""Tool names this provider registers. Must not collide with core."""
return []
def build_system_prompt_block(self) -> str:
"""Return text to inject into system prompt (stable tier)."""
return ""
def prefetch(self, messages: list[dict]) -> None:
"""Called before each turn — load context synchronously."""
pass
def sync(self, messages: list[dict]) -> None:
"""Called after each turn — persist in background worker."""
pass
def on_turn_start() -> :
() -> :
() -> :
() -> :
() -> :
() -> :
MemoryManager
CORE_RESERVED_TOOLS = {
"read_file", "write_file", "terminal", "bash", "web_search",
"web_fetch", "grep", "glob", "memory_recall",
}
class MemoryManager:
def __init__(self):
self._providers: list[MemoryProvider] = []
self._tool_to_provider: dict[str, MemoryProvider] = {}
self._external_count: int = 0
self._sync_executor = ThreadPoolExecutor(max_workers=1)
def add_provider(self, provider: MemoryProvider, is_external: bool = False) -> None:
if is_external and self._external_count >= 1:
raise ValueError(
f"Only 1 external memory provider allowed. "
f"Already registered: {[p.name for p in self._providers if p != provider]}"
)
for tool in provider.tool_names:
if tool in CORE_RESERVED_TOOLS:
raise ValueError(f"Provider '' shadows reserved tool ''")
tool ._tool_to_provider:
ValueError()
._tool_to_provider[tool] = provider
._providers.append(provider)
is_external:
._external_count +=
() -> :
blocks = [p.build_system_prompt_block() p ._providers]
.join(b b blocks b)
() -> :
provider ._providers:
:
provider.prefetch(messages)
Exception e:
()
() -> :
provider ._providers:
._sync_executor.submit(._safe_sync, provider, messages)
() -> :
:
provider.sync(messages)
Exception e:
()
() -> :
provider = ._tool_to_provider.get(tool_name)
provider:
KeyError()
provider.handle_tool(tool_name, args)
() -> :
provider ._providers:
provider.on_session_end()
._sync_executor.shutdown(wait=, timeout=drain_timeout)
Example: Built-in + external vector store
from mem0 import MemoryClient
class Mem0Provider(MemoryProvider):
name = "mem0"
tool_names = ["mem0_recall", "mem0_save"]
def __init__(self, client: MemoryClient, user_id: str):
self._client = client
self._user_id = user_id
self._context = ""
def prefetch(self, messages):
last_user = next((m["content"] for m in reversed(messages)
if m["role"] == "user"), "")
results = self._client.search(last_user, user_id=self._user_id, limit=5)
self._context = "\n".join(r["memory"] for r in results)
def build_system_prompt_block(self) -> str:
if not self._context:
return ""
return (
"<memory-context>\n"
)
manager = MemoryManager()
manager.add_provider(BuiltinL1Provider())
manager.add_provider(Mem0Provider(client, user_id=), is_external=)
Anti-Fake-Pass Checklist
❌ sync_all() called synchronously (await) in turn path — defeats background isolation
❌ Two external providers registered — tool schema doubles; second registration raises
❌ Provider shadowing a core tool name — memory_recall hijacked by external backend
❌ No drain timeout on shutdown — process hangs if background worker stalls
❌ prefetch errors swallowed silently without logging — memory silently empty
