AI Agent Memory Deep Dive: Architecture, Implementation & Forgetting Strategies

Why Agent Memory Matters

Without memory, every interaction with an LLM starts from zero. The article opens with an analogy: a brilliant freelancer who forgets everything overnight. For agents that execute multi-step tasks and improve over time, this amnesia is fatal. Memory provides three capabilities: continuity (identity and preferences), context (current task state), and learning (improving decisions from past outcomes).

Four Types of Agent Memory

1. In-Context Memory (Context Window)

The model's working memory: system prompt, conversation history, tool results, retrieved memories, and scratchpad reasoning. Limited by token window; each token costs money and latency. The article illustrates the sliding window problem and three mitigation strategies:

Summarization : periodically compress old messages.

Selective retention : keep key facts, decisions, tool outputs; discard chatter.

Offload to external storage : move important information to a vector store for on-demand retrieval.

2. External Memory (Persistent Storage)

Data outside the model session, surviving across restarts. Two sub-types:

Structured storage (PostgreSQL, Redis, SQLite): exact lookups by key/ID/SQL; ideal for user profiles and preferences.

Vector storage (Pinecone, Chroma, pgvector): semantic search for unstructured notes and episodic recall.

The article emphasizes that retrieval is the bottleneck — a good memory architecture spends 20% on storage, 80% on retrieval.

3. Episodic Memory (Event Logs)

Stores the outcomes of past actions , not just facts. Each episode is a structured log (task, approach, outcome, duration, token cost, quality score, notes, embedding). The agent retrieves semantically similar past episodes to choose strategies — essentially few-shot learning from its own history . A concrete JSON example shows an episode for summarizing a 50-page PDF with sequential chunking (quality_score 0.91, note: hierarchical chunking would be faster). The reflection loop uses these episodes to improve future decisions.

4. Semantic / Parametric Memory (Model Weights)

The model's baked-in knowledge from training. Always available but has hard limits: knowledge cutoff, no real-time updates, opacity, and hallucination risk. The article frames parametric memory as the agent's "general education" while external, episodic, and in-context memory constitute "work experience."

Memory Flow in the Agent Loop

A diagram shows memory operations bracketing the LLM call: retrieve first, then write back . The model itself is stateless; the memory system creates the illusion of a stateful, perceptive agent.

Building a Memory Layer (Python Implementation)

The article provides a complete, runnable implementation using chromadb, openai, and anthropic.

Dependencies

pip install chromadb openai anthropic python-dotenv

MemoryStore Class

Handles persistent vector memory with cosine similarity search. Key methods: remember(content, memory_type, metadata) — embeds text with text-embedding-3-small, stores in ChromaDB collection per agent. recall(query, k, memory_type, min_relevance) — semantic search with relevance threshold (default 0.6). forget(memory_id) — deletion for GDPR or stale data.

import chromadb
from openai import OpenAI
from datetime import datetime
import json, uuid

class MemoryStore:
    """Persistent vector memory for an AI agent."""
    def __init__(self, agent_id: str, persist_dir: str = "./memory_db"):
        self.agent_id = agent_id
        self.openai = OpenAI()
        self.client = chromadb.PersistentClient(path=persist_dir)
        self.collection = self.client.get_or_create_collection(
            name=f"agent_{agent_id}_memories",
            metadata={"hnsw:space": "cosine"}
        )
    def _embed(self, text: str) -> list[float]:
        response = self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding
    def remember(self, content: str, memory_type: str = "general", metadata: dict = None) -> str:
        memory_id = str(uuid.uuid4())
        embedding = self._embed(content)
        meta = {
            "type": memory_type,
            "timestamp": datetime.utcnow().isoformat(),
            "agent_id": self.agent_id,
            **(metadata or {})
        }
        self.collection.add(
            ids=[memory_id],
            embeddings=[embedding],
            documents=[content],
            metadatas=[meta]
        )
        return memory_id
    def recall(self, query: str, k: int = 5, memory_type: str = None, min_relevance: float = 0.6) -> list[dict]:
        query_embedding = self._embed(query)
        where = {"type": memory_type} if memory_type else None
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=k,
            where=where,
            include=["documents", "metadatas", "distances"]
        )
        memories = []
        for doc, meta, dist in zip(results["documents"][0], results["metadatas"][0], results["distances"][0]):
            relevance = 1 - dist
            if relevance >= min_relevance:
                memories.append({
                    "content": doc,
                    "metadata": meta,
                    "relevance": round(relevance, 3)
                })
        return sorted(memories, key=lambda x: x["relevance"], reverse=True)
    def forget(self, memory_id: str):
        self.collection.delete(ids=[memory_id])

EpisodicLogger Class

Builds on MemoryStore to log structured episodes and retrieve similar past tasks.

from .store import MemoryStore
from dataclasses import dataclass, asdict
from typing import Optional
import time

@dataclass
class Episode:
    task: str
    approach: str
    outcome: str  # "success" | "partial" | "failure"
    duration_ms: int
    token_cost: int
    quality_score: float  # 0.0 - 1.0
    notes: str = ""
    error: Optional[str] = None

class EpisodicLogger:
    def __init__(self, memory_store: MemoryStore):
        self.store = memory_store
    def log(self, episode: Episode):
        doc = (f"Task: {episode.task}
"
               f"Approach: {episode.approach}
"
               f"Outcome: {episode.outcome}
"
               f"Notes: {episode.notes}")
        self.store.remember(
            content=doc,
            memory_type="episode",
            metadata={
                "outcome": episode.outcome,
                "quality_score": episode.quality_score,
                "duration_ms": episode.duration_ms,
                "token_cost": episode.token_cost,
            }
        )
    def recall_similar(self, task: str, k: int = 3) -> list[dict]:
        return self.store.recall(
            query=task,
            k=k,
            memory_type="episode",
            min_relevance=0.65
        )

MemoryAugmentedAgent Class

Orchestrates retrieval, injection, model call, and post-interaction logging.

import anthropic
from memory.store import MemoryStore
from memory.episodic import EpisodicLogger, Episode
import time

class MemoryAugmentedAgent:
    def __init__(self, agent_id: str):
        self.client = anthropic.Anthropic()
        self.memory = MemoryStore(agent_id)
        self.episodes = EpisodicLogger(self.memory)
    def _build_memory_context(self, user_message: str) -> str:
        memories = self.memory.recall(user_message, k=4)
        episodes = self.episodes.recall_similar(user_message, k=2)
        context_parts = []
        if memories:
            context_parts.append("## Relevant memories
" +
                "
".join([f"- [{m['metadata']['type']}] {m['content']} (relevance: {m['relevance']})" for m in memories]))
        if episodes:
            context_parts.append("## Past similar tasks
" +
                "
".join([f"- {e['content'][:200]}..." for e in episodes]))
        return "

".join(context_parts) if context_parts else ""
    def run(self, user_message: str) -> str:
        start = time.time()
        memory_context = self._build_memory_context(user_message)
        system = """You are a helpful agent with memory.
You have access to relevant context from past interactions.
Use this context to give better, more personalized responses."""
        if memory_context:
            system += f"

{memory_context}"
        response = self.client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            system=system,
            messages=[{"role": "user", "content": user_message}]
        )
        answer = response.content[0].text
        duration = int((time.time() - start) * 1000)
        self.memory.remember(f"User asked: {user_message[:200]}", memory_type="interaction")
        self.episodes.log(Episode(
            task=user_message[:200],
            approach="single-turn with memory retrieval",
            outcome="success",
            duration_ms=duration,
            token_cost=response.usage.input_tokens + response.usage.output_tokens,
            quality_score=1.0
        ))
        return answer

Vector Databases and Similarity Search

Vector DBs enable semantic search by finding nearest neighbors in high-dimensional space. Each memory becomes a 1536-dim vector (OpenAI embeddings). Cosine similarity measures conceptual closeness: 1.0 = identical meaning, 0.0 = unrelated, -1.0 = opposite. The article provides a NumPy implementation:

import numpy as np
def cosine_similarity(a: list[float], b: list[float]) -> float:
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# Example
embedding_a = embed("The user prefers dark mode")
embedding_b = embed("They like their interface theme to be dark")
score = cosine_similarity(embedding_a, embedding_b)  # ~0.91

Recommendations: start local with ChromaDB ; if already on Postgres, use pgvector ; for massive scale, consider Pinecone or Qdrant .

Memory Management: Forgetting Strategies

Unbounded growth degrades retrieval (noise, latency, contradictions). Three complementary approaches:

1. Time-Based Decay

Score memories by combining relevance, importance, and recency. Inspired by the Generative Agents paper (Park et al., 2023):

import math
from datetime import datetime
def memory_score(relevance: float, importance: float, created_at: datetime,
                 recency_weight: float = 0.3, decay_factor: float = 0.995) -> float:
    hours_old = (datetime.utcnow() - created_at).total_seconds() / 3600
    recency = math.pow(decay_factor, hours_old)
    return (relevance * 0.4 + importance * 0.3 + recency * recency_weight)

2. Importance Scoring at Write Time

Ask the LLM to rate each piece of information (0.0 trivial, 0.5 moderate, 1.0 critical) before storing; only persist high-scoring entries. Example prompt and parsing logic included.

import re
async def score_importance(client, content: str) -> float:
    prompt = f"""Rate the importance of saving this for future interactions.
0.0 = trivial (greeting)
0.5 = moderately useful
1.0 = critical (preferences, errors, decisions)
Information: {content}
Reply with ONLY the number."""
    try:
        response = await client.messages.create(
            model="claude-3-haiku-20240307",
            max_tokens=10,
            messages=[{"role": "user", "content": prompt}]
        )
        text = response.content[0].text.strip()
        match = re.search(r"[-+]?\d*\.\d+|\d+", text)
        if match:
            score = float(match.group())
            return max(0.0, min(1.0, score))
    except Exception:
        pass
    return 0.5

3. Periodic Consolidation

A nightly job merges near-duplicate memories (cosine similarity >= 0.92) into unified summaries, mimicking human sleep consolidation. The algorithm uses the vector store's own search for efficiency, then atomically replaces the collection.

async def consolidate_memories(store: MemoryStore, similarity_threshold: float = 0.92):
    all_mems = store.collection.get(include=["documents", "embeddings", "ids"])
    if not all_mems["ids"]:
        return
    visited = set()
    consolidated_docs = []
    for i, (mem_id, doc, emb) in enumerate(zip(all_mems["ids"], all_mems["documents"], all_mems["embeddings"])):
        if mem_id in visited:
            continue
        results = store.collection.query(
            query_embeddings=[emb],
            n_results=10,
            include=["documents", "distances"]
        )
        group = [doc]
        visited.add(mem_id)
        for res_id, res_doc, dist in zip(results["ids"][0], results["documents"][0], results["distances"][0]):
            sim = 1.0 - dist
            if res_id != mem_id and res_id not in visited and sim >= similarity_threshold:
                group.append(res_doc)
                visited.add(res_id)
        if len(group) > 1:
            summary = await summarize_group(group)
            consolidated_docs.append(summary)
        else:
            consolidated_docs.append(doc)
    store.collection.delete(where={})
    for doc in consolidated_docs:
        await store.remember(doc)

Conclusion

Memory transforms an AI from a stateless tool into a partner that understands, adapts, and evolves. The real leverage lies not in the model itself but in how you design the memory mechanism — what to remember, what to forget, and how to use it. Get the memory layer right, and everything else becomes smarter.