AIMemoryManager

Purpose

The top-level orchestrator tying together every other subsystem in this package: ShortTermMemory, WorkingMemory, LongTermMemory, SemanticMemory, MemorySelector, MemoryCompressor, ForgettingMechanism, UserProfileManager, ContextBuilder, and the privacy/security layer. This is the class most applications should use directly rather than composing the lower-level pieces themselves — see Quickstart.

_TenantBundle (private, not exported)

class _TenantBundle:
    def __init__(self, tenant_id: str, config: MemoryConfig) -> None:
        self.stm = ShortTermMemory(window_size=config.window_size, tenant_id=tenant_id)
        self.working = WorkingMemory(capacity=config.max_working, token_budget=config.max_tokens, tenant_id=tenant_id)
        self.ltm = LongTermMemory(db_path=f"{config.persistence_path}/{tenant_id}_ltm.db", tenant_id=tenant_id)
        self.semantic = SemanticMemory(model_name=config.embedding_model, cache_size=config.embedding_cache_size, tenant_id=tenant_id)

One bundle is lazily created per unique tenant_id, the first time that tenant is touched by _bundle(). This is where a tenant's SQLite file is actually created on disk — not at AIMemoryManager.__init__ time.

Constructor

def __init__(
    self,
    config: Optional[MemoryConfig] = None,
    llm: Optional[LLMProtocol] = None,
    *,
    encryption_key: Optional[str] = None,
    enable_privacy_masking: bool = True,
    max_workers: int = 4,
) -> None

Also constructs, once, shared across all tenants: MemorySelector (with top_k=config.retrieval_limit), MemoryCompressor, ForgettingMechanism (with base_decay_rate/min_importance/deletion_threshold sourced from config), UserProfileManager (persisted to {config.persistence_path}/profiles only if config.enable_persistence), ContextBuilder (with total_token_budget=config.max_tokens), and AccessController.

Access control helpers

grant(user_id, tenant_id, role="writer") -> None / revoke(user_id, tenant_id, role) -> None

Thin wrappers over AccessController.assign_role/revoke_role.

_check(user_id, tenant_id, perm) -> None (internal)

def _check(self, user_id, tenant_id, perm):
    try:
        self._access_ctrl.require(user_id, tenant_id, perm)
    except PermissionError:
        perms = self._access_ctrl.list_permissions(user_id, tenant_id)
        if not perms:
            pass  # Open by default
        else:
            raise

[!DANGER] Fail-open access control for ungranted users Called at the top of both save_interaction and build_context. If a user has never been granted any role for a tenant, they are allowed through regardless of the requested permission — only users who have at least one role, but a role lacking the specific permission, are actually denied. See the full explanation and implications on the AccessController page. If your application needs real per-user isolation, you must call grant() explicitly for every user you intend to restrict.

save_interaction()

def save_interaction(
    self, user_id: str, tenant_id: str, user_input: str, assistant_output: str, *,
    importance: float = 0.5, tags=None, metadata=None, topics=None,
) -> MemoryEntry

1. _check(..., Permission.WRITE).
2. Masks PII in user_input/assistant_output if masking enabled (discarding the mask report — only the masked text is kept; you don't get to see what was redacted from this call).
3. Encrypts the {"input", "output"} dict if encryption_key was configured, wrapping it as {"encrypted": <ciphertext-string>}.
4. Boosts importance by the user's profile.memory_retention_boost (clamped to 1.0).
5. Stores into bundle.stm (synchronously, on the calling thread).
6. Submits bundle.semantic.add(entry) to the background thread pool (fire-and-forget — no way to await completion or detect failure from this call).
7. If importance >= config.importance_threshold, also submits bundle.ltm.store(entry) to the background thread pool.
8. Updates the user's profile via update_from_interaction.
9. Returns the MemoryEntry that was stored in STM (the in-process representation — if encryption is active, its .content is the {"encrypted": ...} wrapper, not the original plaintext).

[!WARNING] Encryption interacts with encrypted-content filtering downstream build_context()'s history assembly explicitly filters out entries where metadata.get("encrypted") is truthy — but save_interaction never actually sets metadata["encrypted"], it wraps the content as {"encrypted": ...}. This means the filter in build_context (e.metadata.get("encrypted")) checks the wrong location and will not actually exclude encrypted entries from the history list passed to ContextBuilder — an encrypted entry's content, {"encrypted": "<ciphertext>"}, will still be included in history as-is (an opaque ciphertext string embedded directly into the prompt, rather than being filtered out or decrypted). See Known Issues.

[!NOTE] Semantic indexing runs on plaintext-derived embeddings even when encrypted bundle.semantic.add(entry) is submitted with the entry as constructed in step 5/6 above — if encryption is active, entry.content is the {"encrypted": ...} wrapper by the time it reaches SemanticMemory.add(), so _entry_to_text falls through to str(content) (stringifying the ciphertext dict) rather than embedding the original meaningful text. In practice, encryption effectively defeats semantic search for a tenant with encryption_key configured, since the embedded "text" is just a stringified ciphertext blob with no semantic content.

build_context()

def build_context(
    self, user_id: str, tenant_id: str, query: str, *,
    documents=None, top_k_semantic: int = 8, top_k_ltm: int = 10, include_profile: bool = True,
) -> BuiltContext

1. _check(..., Permission.READ).
2. Semantic search (min_score=0.1, hardcoded — not configurable via this method's parameters).
3. Top-top_k_ltm entries from LTM by importance (min_importance=config.min_importance).
4. Merges semantic + LTM + full STM contents, deduplicated by entry.id (first occurrence wins — semantic results take priority over LTM/STM in case of overlap, since they're concatenated first).
5. Ranks via self._selector.select_entries(query, candidates, sem_scores).
6. Loads the selection into bundle.working (replacing its prior contents).
7. Builds profile_text if include_profile=True.
8. Builds history from the last 5 STM entries (bundle.stm.get_recent(5)), filtering to dict-content, non-"encrypted"-metadata entries — see the encryption caveat above regarding why this filter doesn't actually catch encrypted entries.
9. Delegates to self._context_builder.build(...).

build_prompt()

def build_prompt(
    self, user_id: str, tenant_id: str, query: str, *,
    documents=None, system_instruction: str = "You are a helpful, knowledgeable assistant.",
) -> str

ctx = self.build_context(user_id, tenant_id, query, documents=documents)
return self._context_builder.build_prompt(
    query, memories=None, system_instruction=system_instruction, user_profile_text=None,
).replace(
    "=== Current Query ===\n" + query,
    ctx.full_text + "\n\n=== Current Query ===\n" + query,
)

[!DANGER] Brittle string-splice implementation This method calls build_context() once to get the real assembled context (ctx), then calls self._context_builder.build_prompt() a second time with memories=None and user_profile_text=None — producing a mostly-empty template — purely to get its system-instruction/query wrapper text, and then uses a literal string .replace() to splice ctx.full_text in before the "=== Current Query ===" marker. This works only as long as:

1. The exact substring "=== Current Query ===\n" + query appears in the generated template exactly once (it will, under ContextBuilder's current fixed template — see ContextBuilder).
2. query itself doesn't happen to contain that literal marker string, which would make .replace() match at an unintended location or produce unexpected duplication.
3. ContextBuilder's template format never changes in a way that alters the "=== Current Query ===" header text — this is duplicated (implicitly, as a string literal) between context_builder.py and memory_manager.py, so changing one without the other would silently break build_prompt().

A more robust implementation would build the full prompt directly from ctx.full_text plus system_instruction plus the query, without calling ContextBuilder.build_prompt() a second redundant time. See Known Issues.

run_maintenance(tenant_id) -> Dict[str, Any]

Runs, in order: STM decay + eviction, LTM global decay + deletion, forgetting on working memory, and conditional LTM compression (only if ltm_count > config.max_long_term * 0.8).

Also note: decay is applied to STM/LTM/working memory unconditionally, regardless of config.enable_decay's value — that config flag is not actually checked anywhere in this method (see Configuration).

Async wrappers

asave_interaction, abuild_context, arun_maintenance — each does loop.run_in_executor(self._executor, lambda: self.<sync_method>(...)). All inherit the exact same defects as their synchronous counterparts (e.g. arun_maintenance will surface the same ValueError from an empty WorkingMemory, just wrapped in whatever asyncio does with an exception raised inside run_in_executor — it propagates out of the awaited call normally).

stats(tenant_id) -> Dict[str, Any]

Aggregates stm.stats(), working.stats(), ltm.stats(), semantic.stats(), profile_manager.aggregate_stats() (note: this is global across all users, not scoped to tenant_id — UserProfileManager doesn't partition by tenant at all), plus encryption_active and privacy_masking_active booleans.

clear_tenant(tenant_id) -> None

Wipes STM, working memory, LTM, and semantic memory for the tenant. Does not clear that tenant's user profiles (profiles are keyed by user_id, not tenant_id, and UserProfileManager has no tenant-scoped clear operation) — a "cleared" tenant's users retain their accumulated profile data.

Raises

Thread safety

See Architecture → Concurrency Model and Multi-tenancy for the full picture. In short: LongTermMemory and UserProfileManager are internally locked; SemanticMemory, ShortTermMemory, and WorkingMemory are not, and are mutated both synchronously and via background thread-pool submissions — concurrent save_interaction/build_context calls for the same tenant carry a genuine race-condition risk on those unlocked components.

Complexity

Dominated by whichever subsystem operation is slowest per call — typically the SemanticMemory.search() brute-force scan (numpy backend) or the LLM call itself (if summarization/entity-extraction is involved upstream). run_maintenance() is O(n) in LTM size for the decay pass (bounded at 100,000 entries per call, per LongTermMemory.apply_global_decay).

Examples

from zerda.memory import AIMemoryManager, MemoryConfig

class MyLLM:
    def generate(self, prompt: str, max_tokens: int = 256) -> str:
        return "..."

mgr = AIMemoryManager(config=MemoryConfig(), llm=MyLLM())

# 1. Basic
mgr.save_interaction("alice", "acme-corp", "Hi there", "Hello! How can I help?")

# 2. Intermediate — build a prompt-ready context
ctx = mgr.build_context("alice", "acme-corp", "What did we just talk about?")
print(ctx.full_text)

# 3. Advanced — access control
mgr.grant("bob", "acme-corp", role="reader")
try:
    mgr.save_interaction("bob", "acme-corp", "hi", "hello")  # bob only has READ
except PermissionError as e:
    print(e)

# 4. Error handling — DO NOT call run_maintenance on a tenant with no built context yet
# mgr.run_maintenance("brand-new-tenant")  # raises ValueError (empty WorkingMemory)
# Workaround: call build_context() at least once first, or guard the tenant's state:
bundle_working_empty = True  # (illustrative — inspect via mgr.stats(tenant_id)["working"]["entries"] == 0)
if not bundle_working_empty:
    mgr.run_maintenance("brand-new-tenant")

# 5. Production — async usage in a request handler
import asyncio
async def handle_request():
    entry = await mgr.asave_interaction("alice", "acme-corp", "async hi", "async hello")
    ctx = await mgr.abuild_context("alice", "acme-corp", "follow-up question")
    return ctx.full_text

asyncio.run(handle_request())

Best practices

• Call mgr.grant(user_id, tenant_id, role=...) for every user you want subject to real access control — don't rely on the absence of a grant to mean "denied."
• Guard against the empty-WorkingMemory run_maintenance() crash — either ensure build_context() has run at least once for a tenant before scheduling maintenance, or wrap the call in a try/except ValueError until this is patched upstream.
• Don't combine encryption_key with an expectation that semantic search will still work meaningfully — as documented above, encryption currently defeats semantic search for that tenant.
• Use build_context() directly (and assemble your own prompt string) rather than build_prompt() if you want to avoid the brittle string-splice implementation described above.

Common mistakes

• Calling run_maintenance() on a schedule for all tenants without first confirming each has non-empty working memory.
• Assuming encryption_key is a drop-in security upgrade with no other effects — it silently breaks semantic retrieval and the "exclude encrypted content from history" filter doesn't actually work as intended.
• Assuming stats(tenant_id)["profile_manager"] is scoped to that tenant — it's global across all users the process has seen.

Related APIs

• Every class listed in the API Overview — this class orchestrates all of them.
• Architecture — full sequence diagrams for save_interaction and build_context.
• Known Issues & Bugs — the complete, consolidated list of defects referenced throughout this page.