# 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
```python
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
```python
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
| Strategy | Description | Trade-off |
|---|---|---|
| **Sliding Window** | Keep last N messages | Simple, loses history |
| **Summarization** | Compress old messages | Retains gist, loses detail |
| **RAG over History** | Embed + retrieve relevant | Scalable, retrieval errors |
| **Hierarchical** | Summary tree + details on demand | Best balance, complex |

```python
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
```python
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
```python
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
```python
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
```python
@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
```python
@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)
```python
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
```python
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)
```python
# 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)
```python
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)
```python
# 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

```python
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

- [[ContextEngineering/RAG-Fundamentals.md]] — RAG as semantic memory
- [[ContextEngineering/Advanced-RAG.md]] — Agentic RAG as memory
- [[ContextEngineering/Context-Window-Management.md]] — Working memory = context
- [[ContextEngineering/Knowledge-Graphs.md]] — Graph memory
- [[Foundations/Core-Abstractions.md]] — Memory as core abstraction
- [[Architectures/Single-Agent-Patterns.md]] — Reflexion uses episodic
- [[Frameworks/LangGraph.md]] — Checkpointing = state memory
- [[Frameworks/MemGPT.md]] — Virtual context management
- [[Frontiers/Lifelong-Learning.md]] — 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.*