Memory Stratification: Short-Term Scratchpads to Long-Term Vector Stores

A stateless LLM remembers nothing between requests. Every interaction begins tabula rasa. While conversational chatbots can get by with appending recent messages into a prompt buffer, autonomous agents operating over hours or days require a sophisticated, tiered memory hierarchy.

If an agent blindly accumulates every intermediate tool output, observation, and scratchpad thought, its context window rapidly overflows, token costs surge exponentially, and the model suffers from attention distraction (the "needle-in-a-haystack" phenomenon where performance degrades over long contexts).

In this post, we engineer a multi-tiered memory architecture that allows agents to maintain coherent long-term state within finite token budgets.

1. The Multi-Tier Memory Taxonomy

Humans do not store every sensory input in permanent memory; we operate across sensory memory, working memory, episodic memory, and semantic archives. A production AI agent requires an identical stratification:

+-------------------------------------------------------------------------+
|                       AGENT MEMORY ARCHITECTURE                         |
|                                                                         |
|  TIER 1: WORKING MEMORY (Context Window)                                |
|  ┌───────────────────────────────────────────────────────────────────┐  |
|  │  Active Goal • Immediate Scratchpad • Current Tool Call Trajectory│  |
|  │  Capacity: 8K - 128K Tokens | Latency: 0ms | Ephemeral            │  |
|  └─────────────────────────────────┬─────────────────────────────────┘  |
|                                    ▼                                    |
|  TIER 2: EPISODIC MEMORY (Session History & Checkpoints)               |
|  ┌───────────────────────────────────────────────────────────────────┐  |
|  │  Summarized Run Traces • Past Interactions • State Checkpoints    │  |
|  │  Storage: Redis / PostgreSQL / SQLite | Latency: 2-15ms           │  |
|  └─────────────────────────────────┬─────────────────────────────────┘  |
|                                    ▼                                    |
|  TIER 3: SEMANTIC MEMORY (Factual Knowledge & Archives)                |
|  ┌───────────────────────────────────────────────────────────────────┐  |
|  │  Domain Facts • User Profiles • Tool Execution Solutions          │  |
|  │  Storage: Qdrant / Pinecone / pgvector | Latency: 20-80ms         │  |
|  └───────────────────────────────────────────────────────────────────┘  |
+-------------------------------------------------------------------------+

Comparative Characteristics:

2. Working Memory Management: Sliding Windows & Summarization

To prevent context window saturation during long-running tasks, we implement a Sliding Buffer with Dynamic LLM Summarization:

class WorkingMemoryBuffer:
    """Manages active scratchpad memory with automated token condensation."""
    def __init__(self, token_limit: int = 4000, keep_recent_turns: int = 4):
        self.token_limit = token_limit
        self.keep_recent_turns = keep_recent_turns
        self.system_prompt = ""
        self.summary_accumulator = ""
        self.active_history = []

    def add_message(self, role: str, content: str):
        self.active_history.append({"role": role, "content": content})
        self._enforce_budget()

    def _estimate_tokens(self, text: str) -> int:
        # Rough heuristic: ~4 characters per token
        return len(text) // 4

    def _enforce_budget(self):
        total_tokens = sum(self._estimate_tokens(m["content"]) for m in self.active_history)

        if total_tokens > self.token_limit and len(self.active_history) > self.keep_recent_turns:
            # Extract older turns for condensation
            overflow_turns = self.active_history[:-self.keep_recent_turns]
            self.active_history = self.active_history[-self.keep_recent_turns:]

            # Condense older turns into summary accumulator
            text_to_compress = "\n".join([f"{m['role']}: {m['content']}" for m in overflow_turns])
            self._compress_and_merge(text_to_compress)

    def _compress_and_merge(self, text_to_compress: str):
        # In production, call a lightweight fast model (e.g. gpt-4o-mini / Haiku)
        print("⚡ [CONDENSING OVERFLOW MEMORY] Compacting older execution steps...")
        # Appends structured key takeaways to self.summary_accumulator
        self.summary_accumulator += f"\n- Executed earlier steps: {text_to_compress[:120]}..."

    def compile_context(self) -> list:
        """Assembles the final context payload for the LLM."""
        messages = [{"role": "system", "content": self.system_prompt}]
        if self.summary_accumulator:
            messages.append({
                "role": "system",
                "content": f"<prior_session_summary>\n{self.summary_accumulator}\n</prior_session_summary>"
            })
        messages.extend(self.active_history)
        return messages

3. Episodic Memory: Checkpointing State with SQLite

When an agent is interrupted—due to an API timeout, rate limit, or awaiting human approval—it must be able to restore its exact state.

Here is a lightweight SQLite checkpointer for session persistence:

import sqlite3
import json
from datetime import datetime

class SQLiteCheckpointer:
    def __init__(self, db_path: str = "agent_state.db"):
        self.conn = sqlite3.connect(db_path)
        self._init_schema()

    def _init_schema(self):
        with self.conn:
            self.conn.execute("""
                CREATE TABLE IF NOT EXISTS checkpoints (
                    session_id TEXT,
                    step_index INTEGER,
                    state_json TEXT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    PRIMARY KEY (session_id, step_index)
                )
            """)

    def save_checkpoint(self, session_id: str, step_index: int, state: dict):
        with self.conn:
            self.conn.execute(
                "INSERT OR REPLACE INTO checkpoints (session_id, step_index, state_json) VALUES (?, ?, ?)",
                (session_id, step_index, json.dumps(state))
            )

    def load_latest(self, session_id: str) -> dict:
        cursor = self.conn.cursor()
        cursor.execute(
            "SELECT state_json FROM checkpoints WHERE session_id = ? ORDER BY step_index DESC LIMIT 1",
            (session_id,)
        )
        row = cursor.fetchone()
        return json.loads(row[0]) if row else None

4. Semantic Memory: Vectorized Experience Retrieval

Beyond conversational context, mature agents write their own solution reflections into vector stores. When presented with a complex SQL error or API migration, the agent first checks its semantic memory:

"Have I solved a problem similar to this in prior sessions?"

By storing past trajectory tuples (User Goal, Tools Used, Solution Trace) as vector embeddings, the agent develops accumulated expertise across tasks.

Summary & Next Steps

Memory stratification prevents agent amnesia while safeguarding your context window. By maintaining a lean working scratchpad, checkpointing episodic transitions to durable storage, and retrieving historical solutions semantically, your agents achieve genuine persistence.

In [Post 04: Planning & Reasoning Strategies], we explore how agents break down open-ended objectives: comparing Chain-of-Thought (CoT), Plan-and-Solve, and self-correcting Reflexion architectures.