memory-management

Memory Management for Long-Running Agents

Overview

Long-running agents face a fundamental problem: they can't remember everything, but forgetting the wrong thing breaks their usefulness. This skill covers memory architectures that balance context retention, token budget, and retrieval accuracy for agents that run for hours, days, or continuously.

---

Core Concepts

The Memory Problem

Issue	Symptom	Cost
Context Overflow	Agent forgets early instructions	Task failure, incoherent responses
Token Bloat	Every message keeps growing	10x+ cost increase per task
Memory Pollution	Irrelevant memories distract agent	Hallucination, off-target responses
Stale Memories	Outdated information used as fact	Incorrect decisions
Memory Leaks	Unused data accumulates unbounded	Crash from OOM, endless context

Memory Tiers

Tier	Storage	Capacity	Access Speed	Cost	Best For
L1 — Working	In-context (LLM window)	8K-200K tokens	Instant	$$$	Current task, immediate context
L2 — Recent	Sliding window buffer	~2K turns	< 10ms	$$	Recent conversation history
L3 — Episodic	Event log / timeseries	Millions of events	< 50ms	$	Past actions, outcomes, decisions
L4 — Semantic	Vector database	Unlimited	< 100ms	$	Knowledge, facts, relationships
L5 — Archival	Object storage	Unlimited	> 1s	$	Backups, compliance, audit

---

Step-by-Step Implementation

Step 1: Build a Tiered Memory System

from dataclasses import dataclass, field
from typing import Optional
import json
import time

@dataclass
class MemoryEntry:
    content: str
    timestamp: float = None
    importance: float = 0.5  # 0.0 (trivial) to 1.0 (critical)
    tags: list[str] = field(default_factory=list)
    token_count: int = 0
    
    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = time.time()

class TieredMemory:
    """Multi-tier memory with automatic promotion and demotion."""
    
    def __init__(self, llm, vector_store, max_context_tokens: int = 8000):
        self.llm = llm
        self.vector_store = vector_store
        self.max_context_tokens = max_context_tokens
        
        # L1: Working context (in-memory)
        self.working_memory: list[MemoryEntry] = []
        self.current_tokens = 0
        
        # L2: Recent history buffer
        self.recent_buffer: list[MemoryEntry] = []
        self.buffer_size = 50
        
        # L3: Episodic memory
        self.episodes: list[MemoryEntry] = []
        
        # L4: Semantic memory (vector DB)
        # Initialized externally
    
    async def remember(self, content: str, importance: float = 0.5, 
                       tags: list[str] = None):
        """Store a new memory across tiers."""
        entry = MemoryEntry(
            content=content,
            importance=importance,
            tags=tags or [],
            token_count=self._count_tokens(content)
        )
        
        # Always add to working memory
        self.working_memory.append(entry)
        self.current_tokens += entry.token_count
        
        # If important, store in episodic + semantic
        if importance > 0.7:
            self.episodes.append(entry)
            await self.vector_store.store(entry)
        
        # Trim if needed
        await self._trim_working_memory()