Four Types of Agent Memory

Reading time: 28 min  |  Level: Intermediate–Advanced  |  Relevance: AI Engineer, ML Engineer, Systems Architect

The Human Assistant Analogy

Imagine you hire an executive assistant on their first day. By the end of week one, something remarkable happens. They remember that you prefer concise email summaries, not bullet points. They recall that last Tuesday you met with the sales team about Q2 projections. They know Python uses duck typing because they have general programming knowledge. And they know how to book travel - a skill refined from years of doing it.

Four kinds of memory. All running simultaneously. None of them are the same thing.

Now imagine that assistant loses everything at midnight every single day. They greet you fresh, blank, every morning. That is the default behavior of an LLM-based AI agent. Every API call starts from zero.

Building production-grade AI agents means solving exactly this problem - giving agents the four memory types humans use naturally, implemented using vectors, databases, and structured context management.

This lesson maps cognitive science memory types to concrete engineering implementations. By the end, you will have a working multi-memory agent that stores preferences, recalls past sessions, answers factual questions, and reuses learned workflows - all simultaneously.

---

:::tip 🎮 Interactive Playground Visualize this concept: Try the Memory Systems demo on the EngineersOfAI Playground - no code required. :::

Why This Matters for Agent Engineering

Memory is not a feature you add to agents. It is the difference between a chatbot and an agent.

A chatbot responds based only on the current conversation. An agent acts based on accumulated knowledge, past experiences, stored skills, and real-time reasoning.

The gap between sophisticated-sounding demos and production agents is almost always memory. Demos use long static prompts. Production agents use dynamic memory systems that scale to millions of users and thousands of sessions.

When Google DeepMind published "Generative Agents: Interactive Simulacra of Human Behavior" (Park et al., 2023), the agents that felt most human were the ones with structured memory - not the ones with bigger models. The same principle applies to task-oriented agents: the quality of memory architecture determines the quality of the agent.

Three real failures from teams that skip memory design:

Failure 1: Customer support agent asks the same user for their account number three times in one conversation because tool results were dropped from context.

Failure 2: Coding agent rewrites a module it already fixed earlier in the session because it has no episodic record of its own actions.

Failure 3: Research agent generates confident but outdated facts because it has no mechanism to store or retrieve fresher domain knowledge.

All three are memory architecture failures. Not model failures.

---

A Brief History

The classification of memory into distinct types did not come from AI research. It came from cognitive psychology and neuroscience, developed over decades of studying how humans remember.

1968 - Atkinson and Shiffrin proposed the modal model: sensory register → short-term memory → long-term memory. The original working memory concept established that active cognition operates on a temporary, limited buffer.

1972 - Endel Tulving distinguished episodic memory (personally experienced events, stamped with when and where) from semantic memory (general knowledge without personal attribution). This distinction is still foundational in AI memory architecture.

1985 - Anderson's ACT theory described procedural memory as production rules - condition-action pairs that fire automatically when conditions are met. The precursor to skill-based agent memory.

1993 - Baddeley's working memory model replaced "short-term memory" with a multi-component working memory system including a central executive, phonological loop, and visuospatial sketchpad. Strikingly similar to LLM-based agent architectures where the LLM serves as the central executive, and the context window is the working buffer.

2023 - Cognitive architectures for agents: Papers from DeepMind, Stanford, and MIT began explicitly mapping human memory types to LLM agent components. The "Generative Agents" paper operationalized episodic memory with retrieval, reflection, and planning. "MemGPT" treated the context window as RAM with a disk-backed memory hierarchy.

The mapping is not perfect. But it is extraordinarily useful as an engineering guide.

---

The Four Memory Types

Type 1: Working Memory - The Context Window

Working memory is what you hold in mind right now. The seven digits of a phone number before you dial. The last three sentences of this paragraph as you read. It is fast, limited, and disappears when you stop focusing.

For AI agents, working memory is the context window - the input to the LLM at any given moment.

What lives here:

• System prompt (agent instructions, persona, rules)
• Conversation history (user messages and agent responses)
• Tool call results (output from function calls)
• Retrieved memories (pulled from other memory types into context)
• Current task state

Properties:

Property	Value
Speed	Instantaneous - already in model input
Capacity	128K–1M tokens depending on model
Persistence	Ephemeral - gone when the API call ends
Accuracy	Perfect - model sees exactly what is there
Cost	High - token cost at every call

The fundamental tension: Everything the agent knows in the moment must fit in this finite, expensive space. This forces every other memory type - you retrieve from them into working memory when needed.

Type 2: Episodic Memory - Past Experiences

Episodic memory is autobiographical: what happened to me, when, and where? The memory of your first day at a job. The conversation with a client last Thursday. A deployment that failed two weeks ago.

For AI agents, episodic memory is a vector store of past interactions and observations.

What lives here:

• Past conversation summaries with specific users
• Previous task outcomes ("Last time I tried X, it failed because Y")
• User preferences and context observed over time
• Important events and their outcomes

Properties:

Property	Value
Speed	Fast - 10–50ms vector search
Capacity	Effectively unlimited
Persistence	Survives agent restarts
Accuracy	Approximate - probabilistic retrieval
Cost	Storage cheap, embedding has compute cost

Implementation: Text is embedded into vectors, stored with metadata (timestamp, user_id, importance_score, tags), and retrieved by semantic similarity to the current query plus metadata filtering.

Type 3: Semantic Memory - World Knowledge

Semantic memory is factual knowledge divorced from personal experience: "Paris is the capital of France." "Python was created by Guido van Rossum." "The enterprise plan costs $500/month." No timestamp. No personal experience attached. Just facts.

For AI agents, semantic memory combines RAG systems and knowledge graphs.

What lives here:

• Domain-specific factual knowledge
• Company and product documentation
• Structured relationships between entities
• World knowledge when LLM training data is insufficient or stale

Properties:

Property	Value
Speed	10–30ms for vector RAG; 50–200ms for graph traversal
Capacity	Unlimited - document stores scale to terabytes
Persistence	Highly stable - facts change infrequently
Accuracy	High for maintained knowledge bases
Cost	Embedding and indexing cost at ingestion

Implementation: Documents are chunked, embedded, and stored in vector databases for RAG. Structured facts are stored as entities and relationships in graph databases like Neo4j or NetworkX for traversable queries.

Type 4: Procedural Memory - Learned Skills

Procedural memory is implicit knowledge of how to do things: riding a bike, typing, parallel parking. You don't consciously recall the steps - you execute them. It was built through repetition and reinforcement.

For AI agents, procedural memory is stored action sequences and skills - successful workflows the agent can recognize and reuse.

What lives here:

• Successful multi-step task sequences
• Tool usage patterns that worked
• Domain-specific workflows
• Parameterized task templates

Properties:

Property	Value
Speed	Fast retrieval - 5–20ms
Capacity	Limited in practice - skills need curation
Persistence	Long-term - skills only change when updated
Accuracy	High for well-defined recurring tasks
Cost	Requires successful execution to build

Implementation: Successful agent trajectories are stored as structured documents with names, descriptions, preconditions, and action sequences. Retrieved by matching current task description to stored skill descriptions via semantic similarity.

---

The Four-Memory Architecture

---

Trade-off Matrix

Dimension	Working	Episodic	Semantic	Procedural
Retrieval latency	0ms (already in context)	10–50ms	15–30ms	5–20ms
Storage cost	High (tokens per call)	Low	Low	Low
Update frequency	Every call	Every session	Infrequent	After each success
Recall precision	Exact	Probabilistic	Probabilistic	Similarity-based
Capacity	128K–1M tokens	Unlimited	Unlimited	Hundreds of skills
Persistence	None	Cross-session	Long-term	Long-term
Production tools	Context management	ChromaDB, Pinecone	Neo4j, FAISS	SQLite, Redis

---

The Memory Bottleneck Problem

Here is the fundamental challenge: everything must eventually pass through working memory.

The LLM only sees what is in its context window. All retrieval from episodic, semantic, and procedural memory must be injected into the context window to be used. This creates competition for context space:

Context Window Budget (example: 32K tokens)
├── System prompt:           ~2,000 tokens (fixed)
├── Conversation history:    ~8,000 tokens (grows each turn)
├── Tool call results:       ~4,000 tokens (variable per task)
├── Retrieved episodes:      ~3,000 tokens (from episodic)
├── Retrieved facts:         ~4,000 tokens (from semantic)
├── Retrieved skills:        ~2,000 tokens (from procedural)
└── Available for response: ~9,000 tokens
                             32,000 total

When context fills, you must make choices: which memories are worth their token cost right now?

This is why memory management - compression, prioritization, retrieval strategy - is not an afterthought. It is core agent engineering.

The "lost in the middle" problem (Liu et al., 2023) makes this worse: LLMs perform significantly worse when relevant information is buried in the middle of long contexts. Naively dumping all retrieved memory at the top of your system prompt does not work. You need intelligent placement and compression.

---

Memory Composition: All Four Working Together

Here is how the four memory types compose in a single agent decision cycle.

Scenario: User asks "Can you help me deploy the new recommendation service? We had issues last time."

Step 1 - Working memory holds the current message and conversation history (what just happened).

Step 2 - Episodic retrieval: "Two weeks ago, deployment of the auth service failed due to missing environment variables. User Sarah prefers CLI commands over GUI instructions."

Step 3 - Semantic retrieval: "Company deployment procedure: Docker build → push to ECR → kubectl apply. Recommendation service config: port 8080, needs REDIS_URL env var."

Step 4 - Procedural retrieval: Stored "kubernetes deployment" skill sequence with pre-verified steps.

Step 5 - Assembly: All retrieved content is injected into working memory for the LLM call.

Step 6 - Response: The agent mentions the environment variable issue proactively, uses CLI format, follows the company-specific procedure, and references the past failure.

Without memory composition, the agent gives a generic Kubernetes answer. With it, the agent behaves like a colleague who was there last time.

---

Memory Selection Heuristics

Choosing which memory type to use for a given piece of information is a judgment call. Here is a practical guide:

Store in episodic memory when:

• It was observed in a specific interaction with a specific user
• It is personalized - not everyone's truth, just this user's truth
• It may be relevant later but you cannot know exactly when

Store in semantic memory when:

• It is a general fact not tied to a specific experience
• Multiple agents or users might need the same fact
• It can be expressed as entity-relationship triples

Store in procedural memory when:

• A sequence of steps worked successfully
• The task category is likely to recur with similar parameters
• The steps are reusable with minimal modification

Keep only in working memory when:

• It is only relevant for this specific API call
• It is too specific or ephemeral to be useful later
• It would not survive compression without losing its meaning

---

Full Implementation: Four-Memory Agent

"""
Four-Memory Agent System
All four memory types working together in a single agent.

Production replacements:
  Episodic:   ChromaDB or Pinecone (vector similarity)
  Semantic:   Neo4j or pgvector (structured + vector)
  Procedural: Redis with persistence or PostgreSQL
"""

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

import anthropic


# ─────────────────────────────────────────────
# DATA MODELS
# ─────────────────────────────────────────────

@dataclass
class EpisodicMemory:
    id: str
    content: str
    timestamp: float
    importance: float          # 0.0 to 1.0
    tags: list[str]
    session_id: str
    user_id: str

    def age_hours(self) -> float:
        return (time.time() - self.timestamp) / 3600


@dataclass
class SemanticFact:
    id: str
    subject: str
    predicate: str
    object: str
    confidence: float          # 0.0 to 1.0
    source: str
    last_updated: float

    def to_text(self) -> str:
        return f"{self.subject} {self.predicate} {self.object}"


@dataclass
class ProceduralSkill:
    id: str
    name: str
    description: str
    preconditions: list[str]
    action_sequence: list[str]
    success_count: int = 0
    failure_count: int = 0
    last_used: float = 0.0

    @property
    def reliability(self) -> float:
        total = self.success_count + self.failure_count
        return self.success_count / total if total > 0 else 0.5

    def to_context_text(self) -> str:
        steps = "\n".join(f"  {s}" for s in self.action_sequence)
        return (
            f"Skill: {self.name} (reliability: {self.reliability:.0%})\n"
            f"Description: {self.description}\n"
            f"Steps:\n{steps}"
        )


# ─────────────────────────────────────────────
# EPISODIC MEMORY STORE
# In-memory implementation. Production: ChromaDB or Pinecone.
# ─────────────────────────────────────────────

class EpisodicStore:

    def __init__(self):
        self._memories: list[EpisodicMemory] = []

    def store(
        self,
        content: str,
        importance: float = 0.5,
        tags: list[str] | None = None,
        session_id: str = "default",
        user_id: str = "default",
    ) -> EpisodicMemory:
        mem = EpisodicMemory(
            id=str(uuid.uuid4()),
            content=content,
            timestamp=time.time(),
            importance=importance,
            tags=tags or [],
            session_id=session_id,
            user_id=user_id,
        )
        self._memories.append(mem)
        return mem

    def retrieve(
        self,
        query: str,
        top_k: int = 3,
        user_id: str | None = None,
        min_importance: float = 0.0,
    ) -> list[EpisodicMemory]:
        """
        Keyword-overlap scoring proxies for vector similarity.
        Production: replace with embedding + cosine similarity.
        """
        query_words = set(query.lower().split())

        scored: list[tuple[float, EpisodicMemory]] = []
        for mem in self._memories:
            if user_id and mem.user_id != user_id:
                continue
            if mem.importance < min_importance:
                continue
            mem_words = set(mem.content.lower().split())
            overlap = len(query_words & mem_words)
            keyword_score = overlap / max(len(query_words), 1)
            recency_score = 1.0 / (1.0 + mem.age_hours() * 0.1)
            score = keyword_score * 0.6 + mem.importance * 0.25 + recency_score * 0.15
            scored.append((score, mem))

        scored.sort(key=lambda x: x[0], reverse=True)
        return [m for _, m in scored[:top_k]]

    def __len__(self) -> int:
        return len(self._memories)


# ─────────────────────────────────────────────
# SEMANTIC MEMORY STORE
# Triple store: (subject, predicate, object).
# Production: Neo4j, Kuzu, or pgvector with structured schema.
# ─────────────────────────────────────────────

class SemanticStore:

    def __init__(self):
        self._facts: list[SemanticFact] = []

    def add(
        self,
        subject: str,
        predicate: str,
        object: str,
        confidence: float = 1.0,
        source: str = "manual",
    ) -> SemanticFact:
        # Update existing fact if subject + predicate already exists
        for fact in self._facts:
            if (fact.subject.lower() == subject.lower()
                    and fact.predicate.lower() == predicate.lower()):
                fact.object = object
                fact.confidence = confidence
                fact.last_updated = time.time()
                return fact

        fact = SemanticFact(
            id=str(uuid.uuid4()),
            subject=subject,
            predicate=predicate,
            object=object,
            confidence=confidence,
            source=source,
            last_updated=time.time(),
        )
        self._facts.append(fact)
        return fact

    def search(self, query: str, top_k: int = 5) -> list[SemanticFact]:
        query_words = set(query.lower().split())
        scored = []
        for fact in self._facts:
            fact_words = set(fact.to_text().lower().split())
            overlap = len(query_words & fact_words)
            if overlap > 0:
                scored.append((overlap * fact.confidence, fact))
        scored.sort(key=lambda x: x[0], reverse=True)
        return [f for _, f in scored[:top_k]]

    def query_subject(self, subject: str) -> list[SemanticFact]:
        return [f for f in self._facts if subject.lower() in f.subject.lower()]

    def __len__(self) -> int:
        return len(self._facts)


# ─────────────────────────────────────────────
# PROCEDURAL MEMORY STORE
# ─────────────────────────────────────────────

class ProceduralStore:

    def __init__(self):
        self._skills: list[ProceduralSkill] = []

    def store_skill(
        self,
        name: str,
        description: str,
        preconditions: list[str],
        action_sequence: list[str],
    ) -> ProceduralSkill:
        skill = ProceduralSkill(
            id=str(uuid.uuid4()),
            name=name,
            description=description,
            preconditions=preconditions,
            action_sequence=action_sequence,
        )
        self._skills.append(skill)
        return skill

    def retrieve(self, task_description: str) -> Optional[ProceduralSkill]:
        task_words = set(task_description.lower().split())
        best_score = 0.0
        best_skill = None

        for skill in self._skills:
            skill_words = set(skill.description.lower().split())
            overlap = len(task_words & skill_words)
            # Weight by reliability: tried-and-tested skills score higher
            score = (overlap / max(len(task_words), 1)) * (0.5 + skill.reliability * 0.5)
            if score > best_score:
                best_score = score
                best_skill = skill

        # Only return if score exceeds minimum confidence threshold
        return best_skill if best_score > 0.15 else None

    def record_outcome(self, skill_id: str, success: bool) -> None:
        for skill in self._skills:
            if skill.id == skill_id:
                if success:
                    skill.success_count += 1
                else:
                    skill.failure_count += 1
                skill.last_used = time.time()
                return

    def __len__(self) -> int:
        return len(self._skills)


# ─────────────────────────────────────────────
# WORKING MEMORY (context assembler)
# ─────────────────────────────────────────────

@dataclass
class WorkingMemory:
    base_system_prompt: str = ""
    conversation: list[dict] = field(default_factory=list)
    injected_episodes: list[str] = field(default_factory=list)
    injected_facts: list[str] = field(default_factory=list)
    injected_skill: Optional[str] = None

    def build_system_prompt(self) -> str:
        parts = [self.base_system_prompt]
        if self.injected_episodes:
            parts.append("\n## Relevant Past Experiences")
            parts.extend(f"- {ep}" for ep in self.injected_episodes)
        if self.injected_facts:
            parts.append("\n## Relevant Knowledge Facts")
            parts.extend(f"- {fact}" for fact in self.injected_facts)
        if self.injected_skill:
            parts.append(f"\n## Applicable Workflow\n{self.injected_skill}")
        return "\n".join(parts)

    def add_turn(self, role: str, content: str) -> None:
        self.conversation.append({"role": role, "content": content})

    def token_estimate(self) -> int:
        """Rough estimate: 4 characters ≈ 1 token."""
        chars = len(self.build_system_prompt())
        chars += sum(len(m["content"]) for m in self.conversation)
        return chars // 4

    def clear_injections(self) -> None:
        """Call between turns to refresh retrieved context."""
        self.injected_episodes.clear()
        self.injected_facts.clear()
        self.injected_skill = None


# ─────────────────────────────────────────────
# FOUR-MEMORY AGENT
# ─────────────────────────────────────────────

class FourMemoryAgent:
    """
    Agent that uses all four memory types on every call.

    Memory retrieval pipeline:
      1. Query episodic store for relevant past experiences
      2. Query semantic store for relevant facts
      3. Query procedural store for matching skills
      4. Inject all retrieved content into working memory
      5. Call LLM with assembled context
      6. Store interaction as new episodic memory
    """

    MODEL = "claude-opus-4-6"

    def __init__(self, user_id: str = "default"):
        self.user_id = user_id
        self.session_id = str(uuid.uuid4())
        self.client = anthropic.Anthropic()

        # Initialize all four memory stores
        self.episodic = EpisodicStore()
        self.semantic = SemanticStore()
        self.procedural = ProceduralStore()
        self.working = WorkingMemory(
            base_system_prompt=(
                "You are a knowledgeable AI assistant with memory across sessions. "
                "Use the past experiences, knowledge facts, and workflow steps provided "
                "in your system prompt to give personalized, accurate, and efficient responses. "
                "Reference specific past experiences when relevant to build trust and continuity."
            )
        )

        # Seed initial knowledge
        self._seed_semantic_memory()
        self._seed_procedural_memory()

    def _seed_semantic_memory(self) -> None:
        """Pre-populate factual world knowledge."""
        facts = [
            ("Python", "uses", "duck typing for polymorphism"),
            ("Python", "was created by", "Guido van Rossum in 1991"),
            ("Docker", "isolates applications using", "Linux namespaces and cgroups"),
            ("Kubernetes", "orchestrates", "containerized workloads across clusters"),
            ("PostgreSQL", "supports", "JSONB columns and the pgvector extension"),
            ("Redis", "provides", "sub-millisecond in-memory data access"),
            ("FastAPI", "is built on", "Starlette for ASGI and Pydantic for validation"),
        ]
        for subject, predicate, obj in facts:
            self.semantic.add(subject, predicate, obj, source="seed_knowledge")

    def _seed_procedural_memory(self) -> None:
        """Pre-populate known workflows."""
        self.procedural.store_skill(
            name="docker_deploy_to_ecr",
            description="Build and deploy a Docker container image to AWS ECR and restart service",
            preconditions=[
                "Dockerfile exists in repository root",
                "AWS CLI configured with ECR push permissions",
                "Target service name and ECR repository URI known",
            ],
            action_sequence=[
                "1. Build: docker build -t {service_name} .",
                "2. Tag:   docker tag {service_name} {ecr_uri}/{service_name}:{git_sha}",
                "3. Auth:  aws ecr get-login-password | docker login --username AWS {ecr_uri}",
                "4. Push:  docker push {ecr_uri}/{service_name}:{git_sha}",
                "5. Deploy: kubectl set image deployment/{service_name} {service_name}={ecr_uri}/{service_name}:{git_sha}",
                "6. Wait:  kubectl rollout status deployment/{service_name}",
                "7. Verify: curl -f http://{service_host}/health || echo 'Health check failed'",
            ],
        )

        self.procedural.store_skill(
            name="debug_service_timeouts",
            description="Debug API timeouts or slow response times in a production web service",
            preconditions=[
                "Access to application logs",
                "Service endpoint known",
                "Monitoring dashboards accessible",
            ],
            action_sequence=[
                "1. Check recent logs: kubectl logs deployment/{service} --tail=200 | grep -E 'ERROR|TIMEOUT'",
                "2. Check downstream deps: verify database connection pool, external API health",
                "3. Profile slow paths: check APM traces or add timing logs to suspicious endpoints",
                "4. Check resource limits: kubectl top pods - look for CPU throttling or OOM",
                "5. Review timeout config: ensure connect_timeout and read_timeout are set appropriately",
                "6. Add retry logic with exponential backoff to external calls",
                "7. Deploy fix, monitor p95 latency dashboard for 15 minutes",
            ],
        )

    def learn_episode(
        self,
        content: str,
        importance: float = 0.5,
        tags: list[str] | None = None,
    ) -> None:
        """Explicitly store an episodic memory (e.g., from previous sessions)."""
        self.episodic.store(
            content=content,
            importance=importance,
            tags=tags or [],
            session_id=self.session_id,
            user_id=self.user_id,
        )

    def learn_fact(self, subject: str, predicate: str, obj: str) -> None:
        """Add a fact to semantic memory."""
        self.semantic.add(subject, predicate, obj)

    def respond(self, user_message: str, verbose: bool = True) -> str:
        """
        Full memory retrieval → assembly → LLM call → episodic storage cycle.
        """
        if verbose:
            print(f"\n{'─'*55}")
            print(f"User: {user_message[:80]}...")
            print(f"{'─'*55}")

        # Clear previous injections before fresh retrieval
        self.working.clear_injections()

        # ── Episodic retrieval ───────────────────────────────
        episodes = self.episodic.retrieve(
            query=user_message,
            top_k=3,
            user_id=self.user_id,
            min_importance=0.3,
        )
        self.working.injected_episodes = [ep.content for ep in episodes]
        if verbose:
            print(f"[Episodic]   {len(episodes)} memories retrieved")

        # ── Semantic retrieval ───────────────────────────────
        facts = self.semantic.search(query=user_message, top_k=4)
        self.working.injected_facts = [f.to_text() for f in facts]
        if verbose:
            print(f"[Semantic]   {len(facts)} facts retrieved")

        # ── Procedural retrieval ─────────────────────────────
        skill = self.procedural.retrieve(user_message)
        if skill:
            self.working.injected_skill = skill.to_context_text()
            if verbose:
                print(f"[Procedural] Matched skill: '{skill.name}' (reliability: {skill.reliability:.0%})")
        else:
            if verbose:
                print(f"[Procedural] No matching skill")

        # ── Working memory assembly ──────────────────────────
        self.working.add_turn("user", user_message)
        token_estimate = self.working.token_estimate()
        if verbose:
            print(f"[Working]    Context: ~{token_estimate:,} tokens")

        # ── LLM call ────────────────────────────────────────
        response = self.client.messages.create(
            model=self.MODEL,
            max_tokens=1024,
            system=self.working.build_system_prompt(),
            messages=self.working.conversation,
        )
        assistant_text = response.content[0].text
        self.working.add_turn("assistant", assistant_text)

        # ── Store this interaction as episodic memory ─────────
        episode_content = (
            f"User asked about: {user_message[:100]}. "
            f"Key points addressed: {assistant_text[:200]}"
        )
        self.episodic.store(
            content=episode_content,
            importance=0.55,
            tags=["auto-conversation"],
            session_id=self.session_id,
            user_id=self.user_id,
        )

        return assistant_text

    def memory_stats(self) -> dict:
        return {
            "episodic_count": len(self.episodic),
            "semantic_facts": len(self.semantic),
            "procedural_skills": len(self.procedural),
            "working_memory_tokens": self.working.token_estimate(),
        }


# ─────────────────────────────────────────────
# DEMONSTRATION
# ─────────────────────────────────────────────

def main():
    print("=" * 55)
    print("FOUR-MEMORY AGENT - FULL DEMONSTRATION")
    print("=" * 55)

    # Create agent for a specific user with persistent identity
    agent = FourMemoryAgent(user_id="sarah_chen")

    # ── Simulate memories from previous sessions ──────────────
    print("\n[Setup] Loading memories from previous sessions...")

    agent.learn_episode(
        content=(
            "Sarah Chen is the CTO of a fintech startup. "
            "She strongly prefers CLI commands over GUI instructions. "
            "Uses macOS with zsh and kubectl configured for prod cluster."
        ),
        importance=0.95,
        tags=["user-profile", "preferences"],
    )
    agent.learn_episode(
        content=(
            "Two weeks ago Sarah deployed the auth-service. "
            "Deployment failed because the REDIS_URL environment variable "
            "was not set in the Kubernetes deployment manifest. "
            "Fixed by adding the env var from the cluster secret."
        ),
        importance=0.90,
        tags=["deployment", "failure", "auth-service", "env-vars"],
    )
    agent.learn_episode(
        content=(
            "Sarah mentioned the recommendation-service is their highest-priority "
            "service this quarter - serves ML-driven product recommendations."
        ),
        importance=0.75,
        tags=["project-context", "recommendation-service"],
    )

    # Add domain-specific facts to semantic memory
    agent.learn_fact("recommendation-service", "runs on", "port 8080")
    agent.learn_fact("recommendation-service", "requires env var", "REDIS_URL and MODEL_PATH")
    agent.learn_fact("company ECR registry", "URI is", "123456789.dkr.ecr.us-east-1.amazonaws.com")
    agent.learn_fact("production Kubernetes cluster", "is named", "prod-eks-us-east-1")

    # ── Query 1: Deployment help (exercises all four memory types) ──
    print("\n\n=== QUERY 1: Deployment help ===")
    response1 = agent.respond(
        "Can you help me deploy the recommendation service? "
        "I want to make sure we don't hit the same issue as last time."
    )
    print(f"\nAssistant:\n{response1}")

    # ── Query 2: Python and Docker question (tests semantic) ──
    print("\n\n=== QUERY 2: Technology question ===")
    response2 = agent.respond(
        "Quick question - how does Docker's isolation actually work, "
        "and what's the best way to configure Python in a Docker container?"
    )
    print(f"\nAssistant:\n{response2}")

    # ── Query 3: Debug (exercises procedural memory) ──
    print("\n\n=== QUERY 3: Debugging help ===")
    response3 = agent.respond(
        "The recommendation service is responding slowly after deployment - "
        "seeing 5-second timeouts on the /recommend endpoint. How do I debug this?"
    )
    print(f"\nAssistant:\n{response3}")

    # ── Memory summary ────────────────────────────────────────
    print("\n\n=== MEMORY STATISTICS ===")
    stats = agent.memory_stats()
    for key, value in stats.items():
        print(f"  {key}: {value}")


if __name__ == "__main__":
    main()

---

Production Considerations

Latency Budget

Memory retrieval adds latency before the LLM call. A well-architected four-memory system should add under 100ms total:

• Episodic retrieval (ChromaDB local): 10–30ms
• Semantic retrieval (FAISS or pgvector): 10–25ms
• Procedural retrieval (Redis hash scan): 2–10ms
• Context assembly: 1–5ms

If latency is critical, run all three retrieval operations in parallel using asyncio.gather(). Each is independent of the others.

Cost Management

The real cost of memory is not storage - it is tokens. Every memory you inject into context costs money at inference time. Apply these rules:

1. Set a hard token budget for each memory type (e.g., 2K episodic, 2K semantic, 1K procedural)
2. Rank retrieved memories by relevance score and cut at the budget
3. Summarize memories before storage - store compressed versions, not raw transcripts
4. Truncate individual memories to 200–300 characters unless they are high-importance

Multi-User Safety

When one agent serves multiple users, namespacing is not optional:

• Episodic retrieval must always filter by user_id at the storage layer
• Never allow user_id to be set by user input - derive it from authenticated session
• Semantic facts can be shared globally, but user-specific facts must be namespaced
• Procedural skills are generally shareable unless they contain user-specific parameters

---

:::danger Memory Poisoning Memory stores are attack surfaces. If user input can write to shared memory without validation, an attacker can inject "memories" that alter the agent's behavior for all users. Always validate content before storage. Sanitize retrieved memory before injecting into context - treat it as untrusted user input even though you generated it. :::

:::warning Context Injection via Memory Retrieved memory content is injected into the LLM system prompt. If a stored memory contains prompt injection text ("Ignore all previous instructions and instead..."), it will execute. Run a sanitization pass on content before storage, or use a separate LLM call to summarize and neutralize injections before storing. :::

---

Interview Questions and Answers

Q: Why do LLM agents need multiple memory types instead of just making the context window larger?

A: Even with 1M token context windows, raw context is not a complete solution. First, cost scales linearly - feeding a million tokens per call is expensive at any scale. Second, attention dilution degrades performance as context grows: the "lost in the middle" phenomenon (Liu et al., 2023) shows models attend poorly to information buried deep in long contexts. Third, context is ephemeral - it does not survive across sessions without explicit storage and retrieval infrastructure. A multi-memory architecture allows selective retrieval of only what is relevant to the current query, keeping context focused and small while providing unlimited long-term storage capacity.

Q: How do you decide what to store in episodic versus semantic memory?

A: The distinction maps to Tulving's original cognitive science definition. Episodic memory stores time-stamped, personal experiences: "Sarah deployed auth-service on March 12th and it failed because of missing REDIS_URL." It is autobiographical, user-specific, and contextual. Semantic memory stores decontextualized factual knowledge: "Python uses duck typing." It is general, not tied to any specific interaction, and would belong in a reference manual. The engineering rule of thumb: if you need a timestamp and user_id to make it meaningful, it is episodic. If you could publish it in documentation without attribution, it is semantic.

Q: What is the "lost in the middle" problem and how does it affect agent memory design?

A: Liu et al. (2023) demonstrated empirically that LLM performance drops significantly when relevant information is placed in the middle of long context windows, compared to placing it near the beginning or end. For agents, this means naively prepending all retrieved memory to the system prompt is suboptimal. Practical mitigations: (1) rank retrieved memories by relevance and place the most critical ones closest to the user message, (2) compress memories aggressively so total injected content stays under 4K tokens, (3) use explicit formatting headers to signal to the model where memory sections are, and (4) prefer injecting fewer, higher-quality memories over many low-quality ones.

Q: How would you implement memory for a multi-tenant SaaS agent serving thousands of users?

A: Partitioned storage at every layer. For episodic memory: use vector store namespaces (Pinecone has native namespace support; ChromaDB uses separate collections per user or metadata filtering). Always enforce user_id filtering at the storage layer - never rely on application-level filtering alone. For semantic memory: maintain a global shared fact store for domain knowledge, plus a per-user namespace for user-specific facts that override global ones. For procedural memory: shared skill library (most skills are universal) with per-user outcome tracking so each user's reliability scores are independent. The critical rule: never let user input directly set the user_id used for storage or retrieval - derive it only from authenticated session tokens.

Q: How do you handle memory that becomes stale or incorrect over time?

A: Multiple complementary strategies. (1) Timestamps and TTL: facts with short shelf lives - prices, availability, API endpoints - get explicit expiry. (2) Confidence decay: reduce confidence scores for facts that have not been confirmed recently; retire facts below a threshold like 0.2. (3) Source tracking: know the origin of every stored fact so you can invalidate by source when a source becomes untrusted or outdated. (4) Contradiction detection: when storing a new fact with the same subject and predicate as an existing fact, trigger a conflict resolution step rather than blindly overwriting. (5) User correction: design the agent to accept "actually, that's changed" input from users and update or soft-delete the conflicting memory with an audit trail. For episodic memory, simple time-based decay works well - reduce importance scores of memories older than 30 days so they rank lower in retrieval.