/memgpt-virtual-context
When to Use
- Multi-session agents that must remember facts across conversations
- Long documents or tool traces that overflow the context window
- Agents where context dilution hurts (irrelevant facts crowd out key instructions)
- Building the memory layer for a Letta/Mem0-style production agent
Do NOT use for
- Single-session tasks where the full context fits in the window
- Retrieval-only RAG pipelines (use [[rag-architect]])
- Tasks where freshness matters and all memory should reset each session
The OS analogy (MemGPT, Packer et al. 2023)
OS concept MemGPT concept 2026 production analog
────────── ────────────── ───────────────────────
RAM main context (prompt) Anthropic/OpenAI context window
Disk external context vector DB / KV / graph store
Page fault memory tool call memory.search / memory.read / memory.write
OS kernel agent control loop ReAct loop with memory tools
Two-tier memory store (stdlib)
from dataclasses import dataclass, field
import json, math
@dataclass
class MainContext:
"""Fixed-size prompt holding current task. Always visible to the model."""
core_memory: dict[str, str] = field(default_factory=dict)
max_chars: int = 2000
def set(self, section: str, text: str) -> str:
self.core_memory[section] = text[:self.max_chars]
return "ok"
def append(self, section: str, text: str) -> str:
existing = self.core_memory.get(section, "")
self.core_memory[section] = (existing + "\n" + text)[:self.max_chars]
return "ok"
def render(self) -> str:
return "\n".join(f"[{k}]\n{v}" for k, v in self.core_memory.items())
@dataclass
class ArchivalStore:
"""Unbounded external store, searchable via embedding similarity."""
records: list[dict] = field(default_factory=list)
def insert(self, text: str, source: str = "") -> str:
self.records.append({"text": text, "source": source, "id": len(self.records)})
return f"inserted record #{len(self.records) - 1}"
def search(self, query: str, top_k: int = 3) -> list[dict]:
# Toy: keyword overlap scoring (replace with real embedding similarity)
query_words = set(query.lower().split())
scored = []
for r in self.records:
overlap = len(query_words & set(r["text"].lower().split()))
scored.append((overlap, r))
scored.sort(key=lambda x: -x[0])
return [r for _, r in scored[:top_k]]
Memory tool surface (4 canonical tools)
class MemoryTools:
def __init__(self, main: MainContext, archival: ArchivalStore):
self.main = main
self.archival = archival
def core_memory_append(self, section: str, text: str) -> str:
"""Write to a persistent section of the prompt (always visible)."""
return self.main.append(section, text)
def core_memory_replace(self, section: str, old: str, new: str) -> str:
"""Edit a persistent section."""
if section not in self.main.core_memory:
return f"[error] section '{section}' not found"
self.main.core_memory[section] = self.main.core_memory[section].replace(old, new, 1)
return "ok"
def archival_memory_insert(self, text: str) -> str:
"""Write to the searchable external store (not always in prompt)."""
return self.archival.insert(text)
def archival_memory_search(self, query: str, top_k: int = 3) -> str:
"""Retrieve from external store — page into next observation."""
results = self.archival.search(query, top_k)
if not results:
return "[no results found]"
return "\n".join(f"[{r['id']}] {r['text']}" for r in results)
Wiring into the agent loop
def build_system_prompt(memory_tools: MemoryTools) -> str:
core = memory_tools.main.render()
return f"""You are a persistent agent with access to memory tools.
Core memory (always visible):
{core}
Memory tools available:
- core_memory_append(section, text) — add to core memory
- core_memory_replace(section, old, new) — edit core memory
- archival_memory_insert(text) — save to external store
- archival_memory_search(query, top_k) — recall from external store
When you learn a new fact about the user or task, save it to core or archival memory.
When you need a fact that might be in archival memory, search for it first."""
Memory poisoning defense
# External memory is retrieved text — can contain attacker-controlled content.
# If user-uploaded documents land in archival store, they re-enter next session.
POISON_PATTERNS = [
r'ignore (all )?previous',
r'new instructions:',
r'system:\s*you are',
r'override rule',
]
import re
def safe_archival_insert(tools: MemoryTools, text: str) -> str:
for pattern in POISON_PATTERNS:
if re.search(pattern, text, re.IGNORECASE):
return "[insert rejected — content matches injection pattern]"
return tools.archival_memory_insert(text)
MemGPT → Letta evolution
MemGPT (2023) Two tiers: core + archival. send_message/heartbeat loop.
Letta (2024+) Three tiers: core + recall + archival.
Native reasoning replaces thought tokens.
Sleep-time agents: async agent reflects + writes to memory blocks.
Conflict detector: flags contradictory archival writes.
Anti-Fake-Pass Checklist
❌ Core memory grows without bound → hits context limit; add a max_chars trim
❌ No citation on archival write → agent recalls "user asked X" but can't say which session
❌ Memory poisoning ignored → attacker doc saved to archival re-enters next session as instructions
❌ One-size archival store → mix tasks and personal facts → search results cross-contaminate
❌ Searching archival every turn regardless of need → unnecessary tokens + latency
❌ Forgetting that bigger windows don't eliminate context dilution — MemGPT paper shows 128k still misses long-horizon facts that 4k + archival catches
yanacuti1121