Memory Systems

Long-term, short-term, episodic, semantic, and procedural memory for agents.

Memory Taxonomy

Agent Memory
├── Working/Scratchpad     ← Single episode, ~context window
├── Short-term/Conversation ← Session, ~100k tokens
├── Long-term/Semantic     ← Indefinite, facts, knowledge
├── Episodic               ← Indefinite, experiences, "what happened when"
├── Procedural             ← Indefinite, skills, "how to"
└── Meta-memory            ← Indefinite, memory about memory

Working Memory (Scratchpad)

Characteristics

• Duration: Single reasoning episode
• Capacity: Context window (4k-1M tokens)
• Access: Full, sequential
• Volatile: Lost after episode

Implementation

class WorkingMemory:
    def __init__(self, max_tokens=100000):
        self.buffer = []
        self.token_count = 0
        self.max_tokens = max_tokens

    def add(self, item: MemoryItem):
        self.buffer.append(item)
        self.token_count += item.tokens
        self._evict_if_needed()

    def get_context(self, query: str = None, k: int = None) -> list:
        if query:
            return self.semantic_search(query, k)
        return self.buffer[-k:] if k else self.buffer

    def _evict_if_needed(self):
        while self.token_count > self.max_tokens:
            removed = self.buffer.pop(0)
            self.token_count -= removed.tokens

Use Cases

• Reasoning traces (CoT)
• Intermediate tool results
• Temporary calculations
• Current task state

Short-term Memory (Conversation)

Characteristics

• Duration: Session (hours to days)
• Capacity: ~100k tokens
• Access: Sequential + semantic
• Persistence: Session-level

Implementation

class ConversationMemory:
    def __init__(self, session_id: str, token_limit=100000):
        self.session_id = session_id
        self.messages = []
        self.token_limit = token_limit

    def add_message(self, role: str, content: str, metadata: dict = None):
        msg = Message(role=role, content=content, metadata=metadata)
        self.messages.append(msg)
        self._trim()

    def get_context(self, k: int = None) -> list:
        return self.messages[-k:] if k else self.messages

    def _trim(self):
        # Sliding window + importance scoring
        while self.token_count > self.token_limit:
            # Remove least important (not first/last)
            idx = self.find_least_important()
            self.remove(idx)

Compression Strategies

class CompressedConversationMemory:
    def __init__(self):
        self.recent = []  # Full messages
        self.summaries = []  # Compressed blocks

    def add(self, msg):
        self.recent.append(msg)
        if len(self.recent) > 20:
            # Compress oldest 10
            block = self.recent[:10]
            summary = self.summarize(block)
            self.summaries.append(SummaryBlock(summary, block))
            self.recent = self.recent[10:]

    def get_context(self, query):
        # Semantic search over summaries + full recent
        relevant_summaries = self.search_summaries(query, k=3)
        return relevant_summaries + self.recent

Long-term Memory (Semantic)

Vector-Based

class SemanticMemory:
    def __init__(self, vector_store, embed_model):
        self.store = vector_store
        self.embed = embed_model

    async def store(self, content: str, metadata: dict):
        embedding = await self.embed.aget_text_embedding(content)
        await self.vector_store.aupsert([{
            "id": generate_id(),
            "vector": embedding,
            "metadata": {**metadata, "content": content}
        }])

    async def search(self, query: str, k: int = 10, filter: dict = None):
        query_emb = await self.embed.aget_text_embedding(query)
        results = await self.vector_store.aquery(
            vector=query_emb,
            top_k=10,
            filter=filter
        )
        return [MemoryItem(r.metadata["content"], r.metadata) for r in results]

Knowledge Graph Memory

class GraphMemory:
    def __init__(self, graph_store):
        self.graph = graph_store

    async def store_fact(self, subject: str, predicate: str, object: str, confidence: float):
        await self.graph.aupsert_triplet(
            subject=subject,
            predicate=predicate,
            object=object,
            properties={"confidence": confidence, "timestamp": now()}
        )

    async def query(self, pattern: TriplePattern) -> list[Triplet]:
        return await self.graph.aquery(pattern)

    async def get_related(self, entity: str, depth: int = 2) -> Subgraph:
        return await self.graph.atraverse(entity, max_depth=depth)

Consolidation

class MemoryConsolidator:
    async def consolidate(self, working: WorkingMemory, semantic: SemanticMemory):
        # 1. Extract patterns from working memory
        patterns = await self.extract_patterns(working.buffer)

        # 2. Check if already in semantic memory
        for pattern in patterns:
            existing = await semantic.search(pattern.content, k=1)
            if existing and existing[0].similarity > 0.9:
                # Update existing
                await semantic.update(existing[0].id, pattern)
            else:
                # Store new
                await semantic.store(pattern.content, pattern.metadata)

        # 3. Clear working memory
        working.clear()

Episodic Memory

Structure

@dataclass
class Episode:
    id: str
    timestamp: datetime
    context: dict  # Goal, environment, participants
    actions: list[Action]
    outcomes: list[Outcome]
    reflections: list[str]  # What worked, what didn't
    importance: float  # 0-1
    tags: list[str]

class EpisodicMemory:
    def __init__(self, vector_store, graph_store):
        self.vector = vector_store
        self.graph = graph_store

    async def store(self, episode: Episode):
        # Vector for semantic search
        embedding = embed(episode.summary())
        await self.vector.upsert(episode.id, embedding, episode.metadata)

        # Graph for temporal/topic links
        await self.graph.add_episode(episode)

    async def recall(self, query: str, k: int = 5) -> list[Episode]:
        # Semantic search
        vector_results = await self.vector.search(query, k=k*2)

        # Graph expansion (related episodes)
        episode_ids = [r.id for r in vector_results]
        graph_results = await self.graph.get_related(episode_ids, depth=1)

        # Merge and rank
        return self.rank_episodes(vector_results + graph_results, query)

Use Cases

• "What did I do last time this error occurred?"
• "Show me similar debugging sessions"
• "What was the outcome when I tried approach X?"
• Learning from past successes/failures

Procedural Memory (Skills)

Skill Representation

@dataclass
class Skill:
    id: str
    name: str
    description: str
    preconditions: list[Condition]
    steps: list[SkillStep]
    postconditions: list[Condition]
    success_rate: float
    usage_count: int
    version: int

class SkillStep:
    action: str  # Tool name or sub-skill
    parameters: dict
    expected_outcome: str
    fallback: str | None

class ProceduralMemory:
    def __init__(self, vector_store, skill_registry):
        self.vector = vector_store
        self.registry = skill_registry

    async def store_skill(self, skill: Skill):
        # Embed description for retrieval
        embedding = embed(skill.description)
        await self.vector.upsert(skill.id, embedding, skill.metadata)
        self.registry.register(skill)

    async def retrieve_skill(self, goal: str) -> Skill | None:
        results = await self.vector.search(goal, k=5)
        for r in results:
            skill = self.registry.get(r.id)
            if skill and skill.applicable(goal):
                return skill
        return None

    async def learn_skill(self, name: str, demonstration: list[Action]):
        # Induce skill from demonstration
        skill = await self.induce_skill(name, demonstration)
        await self.store_skill(skill)

Skill Induction (Learning)

class SkillLearner:
    async def induce(self, goal: str, trajectory: list[Action]) -> Skill:
        # 1. Abstract trajectory
        abstract = await self.abstract_trajectory(trajectory)

        # 2. Identify parameters
        params = self.identify_parameters(abstract)

        # 3. Generate pre/post conditions
        pre = await self.infer_preconditions(trajectory)
        post = await self.infer_postconditions(trajectory)

        return Skill(
            name=goal,
            steps=abstract,
            parameters=params,
            preconditions=pre,
            postconditions=post
        )

Meta-Memory (Memory About Memory)

Monitoring

class MetaMemory:
    def __init__(self):
        self.stats = {
            "working": {"size": 0, "hit_rate": 0.0},
            "short_term": {"size": 0, "hit_rate": 0.0},
            "semantic": {"size": 0, "hit_rate": 0.0},
            "episodic": {"size": 0, "hit_rate": 0.0},
            "procedural": {"size": 0, "hit_rate": 0.0}
        }

    def record_access(self, memory_type: str, hit: bool):
        self.stats[memory_type]["size"] += 1
        # Exponential moving average
        self.stats[memory_type]["hit_rate"] = (
            0.9 * self.stats[memory_type]["hit_rate"] + 0.1 * (1.0 if hit else 0.0)
        )

    def get_recommendations(self) -> list[str]:
        recs = []
        if self.stats["semantic"]["hit_rate"] < 0.5:
            recs.append("Semantic memory retrieval poor - check embeddings/chunking")
        if self.stats["episodic"]["size"] > 10000:
            recs.append("Episodic memory large - consider consolidation")
        return recs

Memory Architecture Patterns

1. MemGPT (Virtual Context)

# OS-style memory management
class MemGPTMemory:
    def __init__(self):
        self.main_context = []  # "RAM" - in context window
        self.external = VectorStore()  # "Disk" - unlimited
        self.page_size = 2000  # Tokens per page

    def page_in(self, query: str):
        # Retrieve relevant from external
        results = self.external.search(query, k=5)
        self.main_context = self._make_room(results)

    def page_out(self):
        # Move oldest to external
        oldest = self.main_context.pop(0)
        self.external.store(oldest)

2. Generative Agents (Reflection)

class GenerativeAgentMemory:
    def __init__(self):
        self.observations = []  # Raw events
        self.reflections = []   # Synthesized insights

    def add_observation(self, obs):
        self.observations.append(obs)
        if len(self.observations) % 10 == 0:
            self.reflect()

    def reflect(self):
        # Synthesize higher-level insights
        recent = self.observations[-100:]
        reflection = llm.reflect(recent)
        self.reflections.append(reflection)

    def retrieve(self, query):
        # Search observations + reflections
        return search(self.observations + self.reflections, query)

3. Zep (Temporal + Entity)

# Temporal knowledge graph
class ZepMemory:
    def __init__(self, zep_client):
        self.client = zep_client

    async def add_memory(self, session_id: str, messages: list[Message]):
        await self.client.add_memory(session_id, messages)

    async def search(self, session_id: str, query: str) -> list:
        return await self.client.search(session_id, query)

    async def get_entities(self, session_id: str) -> list[Entity]:
        return await self.client.get_entities(session_id)

Memory Consolidation Pipeline

class MemoryPipeline:
    def __init__(self):
        self.working = WorkingMemory()
        self.short_term = ConversationMemory()
        self.semantic = SemanticMemory()
        self.episodic = EpisodicMemory()
        self.procedural = ProceduralMemory()

    async def step(self, event: Event):
        # 1. Working memory
        self.working.add(event)

        # 2. Short-term
        self.short_term.add(event)

        # 3. Periodic consolidation
        if self.should_consolidate():
            await self.consolidate()

    async def consolidate(self):
        # Working → Short-term (already there)
        # Short-term → Semantic (extract facts)
        facts = await self.extract_facts(self.short_term)
        await self.semantic.store_batch(facts)

        # Short-term → Episodic (form episodes)
        episodes = await self.form_episodes(self.short_term)
        await self.episodic.store_batch(episodes)

        # Identify skills → Procedural
        skills = await self.identify_skills(self.short_term)
        await self.procedural.store_batch(skills)

        # Clear old short-term
        self.short_term.trim(keep_recent=50)

Related Notes

• RAG Fundamentals — RAG as semantic memory
• Advanced RAG — Agentic RAG as memory
• Context Window Management — Working memory = context
• Knowledge Graphs — Graph memory
• Core Abstractions — Memory as core abstraction
• Single Agent Patterns — Reflexion uses episodic
• LangGraph — Checkpointing = state memory
• MemGPT — Virtual context management
• Lifelong Learning — Continual memory consolidation

Sources: "Generative Agents" (Park et al., 2023), "MemGPT" (Packer et al., 2023), "Reflexion" (Shinn et al., 2023), "Zep" (Zep AI), "Mem0" (Mem0 AI), "Memory in LLMs" surveys, cognitive architecture literature (Soar, ACT-R, LIDA), memory consolidation neuroscience.