How Claude, GPT, and Gemini Actually Remember (2026) – AI Agent Memory Systems

┌────────────────────────────────────────────────────────┐
│  Layer 1: In-Context (Working Memory)                  │
│  What:     Current context window — messages, tool     │
│            results, system prompt, retrieved docs      │
│  Duration: Current API call only                       │
│  Access:   Instant — already in the model's input      │
│  Limit:    Context window size (8K → 2M+ tokens)       │
└────────────────────┬───────────────────────────────────┘
                     │ retrieved/summarized into context
┌────────────────────▼───────────────────────────────────┐
│  Layer 2: External Short-Term (Session Memory)         │
│  What:     Current session / task history              │
│  Duration: Hours to days (TTL-based cleanup)           │
│  Access:   Key-value lookup by thread ID               │
│  Storage:  Redis, SQLite, PostgreSQL                   │
└────────────────────┬───────────────────────────────────┘
                     │ semantically retrieved when relevant
┌────────────────────▼───────────────────────────────────┐
│  Layer 3: External Long-Term (Episodic + Semantic)     │
│  What:     Past interactions, user preferences,        │
│            knowledge base, learned facts               │
│  Duration: Months to indefinite                        │
│  Access:   Semantic search over vector index           │
│  Storage:  Pinecone, pgvector, Weaviate, Qdrant        │
└────────────────────┬───────────────────────────────────┘
                     │ updated only via training / fine-tuning
┌────────────────────▼───────────────────────────────────┐
│  Layer 4: Parametric Memory (Model Weights)            │
│  What:     World knowledge from pretraining            │
│  Duration: Fixed until retrained                       │
│  Access:   Model inference — no external call          │
│  Limit:    Training cutoff date; cannot be selectively │
│            updated or queried precisely                │
└────────────────────────────────────────────────────────┘

import json
import sqlite3
from datetime import datetime
from typing import Optional

import numpy as np


class MemoryManager:
    """
    Manages all three external memory layers for an AI agent:
    - Session layer: SQLite-backed conversation history
    - Long-term layer: vector-indexed episodic and preference memories
    - Consolidation: merges redundant memories periodically
    """

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

    def _init_schema(self):
        self.conn.executescript("""
            CREATE TABLE IF NOT EXISTS sessions (
                thread_id TEXT PRIMARY KEY,
                history   TEXT NOT NULL,
                updated   TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            );

            CREATE TABLE IF NOT EXISTS memories (
                id          TEXT PRIMARY KEY,
                user_id     TEXT NOT NULL,
                type        TEXT NOT NULL,   -- 'preference', 'episode', 'fact'
                content     TEXT NOT NULL,
                embedding   BLOB,            -- stored as float32 numpy bytes
                created     TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                last_accessed TIMESTAMP,
                access_count  INTEGER DEFAULT 0
            );

            CREATE INDEX IF NOT EXISTS idx_memories_user ON memories(user_id);
        """)

    # ── Session layer ──────────────────────────────────────────────────────────

    def load_session(self, thread_id: str) -> list:
        row = self.conn.execute(
            "SELECT history FROM sessions WHERE thread_id = ?", (thread_id,)
        ).fetchone()
        return json.loads(row[0]) if row else []

    def save_session(self, thread_id: str, history: list) -> None:
        self.conn.execute(
            "INSERT OR REPLACE INTO sessions (thread_id, history) VALUES (?, ?)",
            (thread_id, json.dumps(history)),
        )
        self.conn.commit()

    # ── Long-term memory layer ─────────────────────────────────────────────────

    def store_memory(
        self,
        user_id: str,
        content: str,
        memory_type: str,
        embedding: np.ndarray,
    ) -> str:
        mem_id = f"mem_{datetime.now().timestamp()}"
        self.conn.execute(
            """INSERT INTO memories (id, user_id, type, content, embedding)
               VALUES (?, ?, ?, ?, ?)""",
            (mem_id, user_id, memory_type, content,
             embedding.astype(np.float32).tobytes()),
        )
        self.conn.commit()
        return mem_id

    def retrieve_relevant(
        self,
        user_id: str,
        query_embedding: np.ndarray,
        top_k: int = 5,
        threshold: float = 0.70,
    ) -> list[dict]:
        """
        Brute-force cosine similarity retrieval.
        In production with millions of memories, swap for pgvector or Pinecone.
        The interface stays identical; only the underlying call changes.
        """
        rows = self.conn.execute(
            "SELECT id, type, content, embedding FROM memories WHERE user_id = ?",
            (user_id,),
        ).fetchall()

        if not rows:
            return []

        scored = []
        query_norm = query_embedding / (np.linalg.norm(query_embedding) + 1e-8)

        for row_id, mem_type, content, emb_bytes in rows:
            stored_vec = np.frombuffer(emb_bytes, dtype=np.float32)
            stored_norm = stored_vec / (np.linalg.norm(stored_vec) + 1e-8)
            score = float(np.dot(query_norm, stored_norm))
            if score >= threshold:
                scored.append({
                    "id": row_id,
                    "type": mem_type,
                    "content": content,
                    "score": score,
                })

        # Sort by score descending, return top-k
        scored.sort(key=lambda x: x["score"], reverse=True)

        # Update access metadata for the retrieved memories
        retrieved = scored[:top_k]
        if retrieved:
            ids = [m["id"] for m in retrieved]
            placeholders = ",".join("?" * len(ids))
            self.conn.execute(
                f"""UPDATE memories
                   SET last_accessed = CURRENT_TIMESTAMP,
                       access_count  = access_count + 1
                   WHERE id IN ({placeholders})""",
                ids,
            )
            self.conn.commit()

        return retrieved

def assemble_context_with_memory(
    user_id: str,
    thread_id: str,
    current_query: str,
    system_prompt: str,
    memory_manager: MemoryManager,
    embed_fn,  # callable: str -> np.ndarray
) -> list[dict]:
    """
    Assembles the full context for an API call:
    1. System prompt (with memory injected at top)
    2. Retrieved long-term memories (most relevant to current query)
    3. Session history (recent turns verbatim)
    4. Current user message
    """
    query_embedding = embed_fn(current_query)
    relevant_memories = memory_manager.retrieve_relevant(
        user_id, query_embedding, top_k=5
    )

    memory_block = ""
    if relevant_memories:
        lines = [f"- [{m['type']}] {m['content']}" for m in relevant_memories]
        memory_block = (
            "\n\nRelevant context from previous sessions:\n"
            + "\n".join(lines)
        )

    # Inject memory at the top of the system prompt — closest to start of
    # context where attention is most reliable.
    full_system_prompt = system_prompt + memory_block

    session_history = memory_manager.load_session(thread_id)

    # Budget-aware history pruning: keep recent turns, summarize the rest.
    session_history = _prune_to_budget(session_history, token_limit=8000)

    messages = session_history + [{"role": "user", "content": current_query}]

    return full_system_prompt, messages

MEMORY_EXTRACTION_PROMPT = """
Review the conversation below and extract any information worth remembering
for future sessions. Focus on:
- User preferences stated explicitly ("I prefer...", "always use...", "never...")
- Important facts the user revealed about their context (job, project, constraints)
- Significant decisions or conclusions reached during this conversation

Return a JSON array. Each item must have:
  "type": "preference" | "episode" | "fact"
  "content": concise one-sentence statement (under 120 characters)

Return ONLY valid JSON, no preamble.
Example: [{"type": "preference", "content": "User prefers TypeScript over JavaScript for all new projects."}]

If nothing worth storing was discussed, return an empty array: []
"""

def extract_memories_from_session(
    conversation_history: list,
    llm_client,
) -> list[dict]:
    """
    Uses a lightweight LLM call to extract memorable facts from a session.
    Runs once at session end, not on every turn.
    """
    history_text = "\n".join(
        f"{m['role'].upper()}: {m['content']}"
        for m in conversation_history
        if isinstance(m.get("content"), str)
    )

    response = llm_client.messages.create(
        model="claude-haiku-4-5-20251001",  # Cheap, fast — appropriate for extraction
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": MEMORY_EXTRACTION_PROMPT + f"\n\nConversation:\n{history_text}",
        }],
    )

    text = response.content[0].text.strip()
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return []  # Fail silently — a missed memory is less bad than a crash