Implementing Structured Agent Memory: Building Long-Term Persistent State with Episodic, Semantic, and Working Memory Layers for Production LLM Agents

Introduction

Imagine you're building a customer success agent for a B2B SaaS company. Your agent handles support tickets, tracks product usage patterns, and advises account managers. On day one, it works beautifully — it answers questions with precision and takes appropriate actions. But by day thirty, every conversation starts cold. The agent has no idea that a particular customer has complained about the same bug three times this month, that their contract renewal is in sixty days, or that they had a breakthrough success with a specific feature last week. Every interaction is as amnesia-riddled as the last. This is the difference between a chatbot and an agent that actually accumulates expertise and context over time.

Production LLM agents fail at memory not because the models are incapable of using context, but because most implementations treat memory as an afterthought — a simple list of messages you append to a prompt until it overflows the context window. Real agent memory requires deliberate architecture: distinct layers that serve different cognitive purposes, persistence strategies that survive process restarts and scaling, and retrieval mechanisms that surface the right information at the right time rather than dumping everything into an already-stretched context window.

By the end of this lesson, you'll have built a production-grade memory system with three distinct layers — episodic, semantic, and working memory — wired together in a coherent architecture that you can actually deploy. You'll understand not just how to implement each layer, but why they need to be separate, when they should communicate with each other, and how to avoid the subtle failure modes that plague most agent memory implementations in the wild.

What you'll learn:

• The cognitive science rationale behind episodic, semantic, and working memory layers and how they translate to agent architecture
• How to implement episodic memory using vector databases with time-aware retrieval and decay functions
• How to build a semantic memory layer that extracts and consolidates structured knowledge from raw interactions
• How to design a working memory context window manager with priority-based eviction
• How to wire all three layers together with a memory orchestration layer, including consolidation pipelines that run asynchronously
• Production considerations: consistency guarantees, memory poisoning attacks, and scaling patterns

---

Prerequisites

You should be comfortable with:

• Python at an intermediate-to-advanced level (async/await, dataclasses, type hints)
• LangChain or similar agent frameworks — we'll use framework-agnostic patterns but reference LangChain constructs
• Vector databases — basic understanding of embeddings and similarity search (we'll use Qdrant in examples, but the patterns transfer to Pinecone, Weaviate, or pgvector)
• Redis or similar for fast key-value storage
• PostgreSQL for structured relational data
• Basic async Python — our consolidation pipeline uses asyncio

If you're fuzzy on vector databases specifically, spend thirty minutes with Qdrant's quickstart before continuing. The rest will make much more sense.

---

Why Three Memory Layers? The Cognitive Architecture Argument

Before writing a single line of code, let's understand why we're building this complexity. Collapsing everything into a single memory store — a common shortcut — creates problems that compound over time.

In cognitive science, human memory is divided into distinct systems that serve different purposes. Episodic memory stores specific events with temporal context ("last Tuesday, the customer called angry about an outage"). Semantic memory stores generalized knowledge extracted from many episodes ("this customer runs latency-sensitive workloads and has low tolerance for P99 spikes"). Working memory is the limited-capacity active workspace where current reasoning happens.

The reason these are separate systems in biological cognition is not accidental — it reflects fundamentally different requirements:

Property	Episodic	Semantic	Working
Granularity	Event-level	Fact-level	Task-level
Temporal scope	Historical	Timeless	Current session
Size	Large, unbounded	Medium, grows slowly	Small, strictly bounded
Retrieval pattern	Similarity + recency	Lookup + inference	Priority queue
Persistence	Long-term	Long-term	Ephemeral
Mutation rate	Append-only	Slow evolution	High churn

When you try to use a single vector store for everything, you get catastrophic interference: a highly-specific event from three months ago competes with a generalized fact about the customer when you run a similarity query. You also lose the ability to apply different eviction, consolidation, and consistency policies to each layer.

The architecture we're building looks like this conceptually:

Incoming Interaction
        │
        ▼
┌───────────────────┐
│   Working Memory  │  ← Active reasoning context
│   (Redis/In-Mem)  │
└────────┬──────────┘
         │ writes             reads (retrieval)
         ▼                    ▲
┌───────────────────┐    ┌────────────────────┐
│  Episodic Memory  │───▶│  Semantic Memory   │
│  (Vector DB)      │    │  (PG + Vector DB)  │
└───────────────────┘    └────────────────────┘
         │                    │
         └──── Consolidation ─┘
              Pipeline (async)

Let's build each layer from the ground up.

---

Layer 1: Episodic Memory — The Event Store

Episodic memory is your agent's journal. Every significant interaction, decision, and observation gets written here with enough context to reconstruct what happened and why. The key design constraints:

1. Append-only — you never rewrite history
2. Temporally indexed — recency matters as much as semantic similarity
3. Rich metadata — enough to filter, slice, and analyze episodes without re-reading them all

Defining the Episode Schema

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, Dict, Any, List
from uuid import uuid4
import json

@dataclass
class Episode:
    """
    A single recorded experience for the agent.
    
    An episode captures one complete interaction or significant event,
    including its context, the actions taken, and their outcomes.
    """
    episode_id: str = field(default_factory=lambda: str(uuid4()))
    agent_id: str = ""
    entity_id: str = ""          # e.g., customer_id, user_id, session_id
    entity_type: str = ""        # e.g., "customer", "user", "product"
    
    timestamp: datetime = field(default_factory=datetime.utcnow)
    
    # The raw interaction content
    content: str = ""            # Full text of the interaction
    summary: str = ""            # LLM-generated summary (populated async)
    
    # What happened
    interaction_type: str = ""   # e.g., "support_ticket", "analysis_run", "proactive_alert"
    actions_taken: List[str] = field(default_factory=list)
    outcome: str = ""            # e.g., "resolved", "escalated", "pending"
    
    # Emotional/importance signals
    sentiment_score: float = 0.0  # -1.0 to 1.0
    importance_score: float = 0.5  # 0.0 to 1.0, used for retrieval weighting
    
    # Structured metadata for filtering
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    # Embedding (populated by memory store, not set manually)
    embedding: Optional[List[float]] = None
    
    def to_retrieval_text(self) -> str:
        """
        Generates the text that will be embedded for similarity search.
        This is NOT the same as content — it's a semantic-dense representation.
        """
        parts = [
            f"Entity: {self.entity_type} {self.entity_id}",
            f"Type: {self.interaction_type}",
            f"Summary: {self.summary or self.content[:500]}",
            f"Outcome: {self.outcome}",
            f"Actions: {', '.join(self.actions_taken)}",
        ]
        # Include important metadata keys
        for key in ["issue_category", "product_area", "resolution_type"]:
            if key in self.metadata:
                parts.append(f"{key}: {self.metadata[key]}")
        return "\n".join(parts)
    
    def to_context_block(self) -> str:
        """
        Human-readable representation for injection into prompt context.
        """
        age = (datetime.utcnow() - self.timestamp).days
        age_str = f"{age} days ago" if age > 0 else "today"
        return (
            f"[{self.interaction_type.upper()} — {age_str}]\n"
            f"{self.summary or self.content[:300]}\n"
            f"Outcome: {self.outcome} | Importance: {self.importance_score:.1f}"
        )

Notice we're separating to_retrieval_text() from content. This is critical: the text you embed for similarity search should be semantically concentrated — stripped of pleasantries, formatted to emphasize the dimensions you actually want to search across. If you embed raw chat transcripts, you'll get similarity matches based on conversational patterns rather than actual event similarity.

Building the Episodic Memory Store

from qdrant_client import QdrantClient
from qdrant_client.models import (
    Distance, VectorParams, PointStruct, 
    Filter, FieldCondition, MatchValue, Range
)
from openai import OpenAI
import numpy as np
from typing import Tuple
import time

class EpisodicMemoryStore:
    """
    Vector-backed episodic memory with time-weighted retrieval.
    
    Uses Qdrant for similarity search with custom scoring that
    combines semantic similarity with temporal recency.
    """
    
    COLLECTION_NAME = "agent_episodes"
    EMBEDDING_MODEL = "text-embedding-3-small"
    EMBEDDING_DIM = 1536
    RECENCY_HALF_LIFE_DAYS = 30  # Episodes lose half their recency weight every 30 days
    
    def __init__(self, qdrant_url: str, openai_client: OpenAI):
        self.qdrant = QdrantClient(url=qdrant_url)
        self.openai = openai_client
        self._ensure_collection()
    
    def _ensure_collection(self):
        """Create the collection if it doesn't exist."""
        collections = [c.name for c in self.qdrant.get_collections().collections]
        if self.COLLECTION_NAME not in collections:
            self.qdrant.create_collection(
                collection_name=self.COLLECTION_NAME,
                vectors_config=VectorParams(
                    size=self.EMBEDDING_DIM,
                    distance=Distance.COSINE
                )
            )
            # Create payload indices for fast filtering
            self.qdrant.create_payload_index(
                self.COLLECTION_NAME, "entity_id", "keyword"
            )
            self.qdrant.create_payload_index(
                self.COLLECTION_NAME, "entity_type", "keyword"
            )
            self.qdrant.create_payload_index(
                self.COLLECTION_NAME, "timestamp_unix", "float"
            )
            self.qdrant.create_payload_index(
                self.COLLECTION_NAME, "importance_score", "float"
            )
    
    def _embed(self, text: str) -> List[float]:
        response = self.openai.embeddings.create(
            model=self.EMBEDDING_MODEL,
            input=text
        )
        return response.data[0].embedding
    
    def _recency_weight(self, timestamp: datetime) -> float:
        """
        Exponential decay function for recency weighting.
        Returns a value in (0, 1] — 1.0 for right now, 
        approaching 0 for very old episodes.
        """
        age_days = (datetime.utcnow() - timestamp).total_seconds() / 86400
        return np.exp(-np.log(2) * age_days / self.RECENCY_HALF_LIFE_DAYS)
    
    async def store(self, episode: Episode) -> str:
        """Store an episode and return its ID."""
        retrieval_text = episode.to_retrieval_text()
        embedding = self._embed(retrieval_text)
        
        payload = {
            "episode_id": episode.episode_id,
            "agent_id": episode.agent_id,
            "entity_id": episode.entity_id,
            "entity_type": episode.entity_type,
            "timestamp_unix": episode.timestamp.timestamp(),
            "timestamp_iso": episode.timestamp.isoformat(),
            "content": episode.content,
            "summary": episode.summary,
            "interaction_type": episode.interaction_type,
            "actions_taken": episode.actions_taken,
            "outcome": episode.outcome,
            "sentiment_score": episode.sentiment_score,
            "importance_score": episode.importance_score,
            "metadata": json.dumps(episode.metadata),
        }
        
        self.qdrant.upsert(
            collection_name=self.COLLECTION_NAME,
            points=[PointStruct(
                id=episode.episode_id,
                vector=embedding,
                payload=payload
            )]
        )
        return episode.episode_id
    
    def retrieve(
        self,
        query: str,
        entity_id: str,
        limit: int = 5,
        recency_weight: float = 0.3,
        min_importance: float = 0.0,
        interaction_types: Optional[List[str]] = None,
    ) -> List[Tuple[Episode, float]]:
        """
        Retrieve relevant episodes with combined semantic + recency scoring.
        
        recency_weight: how much to weight recency vs pure semantic similarity.
                        0.0 = pure similarity, 1.0 = pure recency.
        """
        query_embedding = self._embed(query)
        
        # Build filter conditions
        must_conditions = [
            FieldCondition(key="entity_id", match=MatchValue(value=entity_id))
        ]
        if min_importance > 0:
            must_conditions.append(
                FieldCondition(key="importance_score", range=Range(gte=min_importance))
            )
        if interaction_types:
            must_conditions.append(
                FieldCondition(
                    key="interaction_type",
                    match=MatchValue(value=interaction_types[0])  # simplified
                )
            )
        
        # Fetch more than we need so we can re-rank
        raw_results = self.qdrant.search(
            collection_name=self.COLLECTION_NAME,
            query_vector=query_embedding,
            query_filter=Filter(must=must_conditions),
            limit=limit * 3,  # Fetch 3x to allow re-ranking
            with_payload=True,
        )
        
        # Re-rank with combined score
        scored_results = []
        for hit in raw_results:
            p = hit.payload
            timestamp = datetime.fromisoformat(p["timestamp_iso"])
            recency = self._recency_weight(timestamp)
            
            # Combined score: weighted combination of semantic similarity and recency
            # hit.score is cosine similarity in [0, 1]
            combined_score = (
                (1 - recency_weight) * hit.score +
                recency_weight * recency * p["importance_score"]
            )
            
            episode = self._payload_to_episode(p)
            scored_results.append((episode, combined_score))
        
        # Sort by combined score and return top-k
        scored_results.sort(key=lambda x: x[1], reverse=True)
        return scored_results[:limit]
    
    def _payload_to_episode(self, payload: dict) -> Episode:
        return Episode(
            episode_id=payload["episode_id"],
            agent_id=payload["agent_id"],
            entity_id=payload["entity_id"],
            entity_type=payload["entity_type"],
            timestamp=datetime.fromisoformat(payload["timestamp_iso"]),
            content=payload["content"],
            summary=payload["summary"],
            interaction_type=payload["interaction_type"],
            actions_taken=payload["actions_taken"],
            outcome=payload["outcome"],
            sentiment_score=payload["sentiment_score"],
            importance_score=payload["importance_score"],
            metadata=json.loads(payload["metadata"]),
        )

Key design decision: We fetch limit * 3 results from Qdrant and then re-rank them ourselves. This lets us apply the recency weighting after semantic filtering, which produces much better results than trying to bake temporal weights into the embedding itself (a common mistake that pollutes your embedding space).

---

Layer 2: Semantic Memory — The Knowledge Graph

While episodic memory is the journal, semantic memory is the extracted wisdom from that journal. It stores facts about the world and about entities the agent interacts with — facts that have been validated across multiple episodes and generalized into stable, queryable knowledge.

The key insight here: semantic memory should be hard-won. A single episode shouldn't immediately update semantic memory. Knowledge should be consolidated from multiple corroborating episodes, which is why we run the consolidation pipeline asynchronously.

Defining the Semantic Memory Schema

We'll use PostgreSQL for structured facts (because you want ACID guarantees on your agent's beliefs about the world) plus a vector index for semantic search over fact content.

-- Run this migration to set up the semantic memory schema

CREATE TABLE agent_entities (
    entity_id       VARCHAR(255) PRIMARY KEY,
    entity_type     VARCHAR(50) NOT NULL,
    display_name    VARCHAR(255),
    created_at      TIMESTAMPTZ DEFAULT NOW(),
    updated_at      TIMESTAMPTZ DEFAULT NOW(),
    metadata        JSONB DEFAULT '{}'
);

CREATE TABLE semantic_facts (
    fact_id         UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    entity_id       VARCHAR(255) REFERENCES agent_entities(entity_id),
    
    -- The fact itself
    subject         VARCHAR(255) NOT NULL,   -- e.g., "customer:acme_corp"
    predicate       VARCHAR(255) NOT NULL,   -- e.g., "prefers_channel"
    object          TEXT NOT NULL,           -- e.g., "email"
    
    -- Provenance and confidence
    confidence      FLOAT DEFAULT 0.5,       -- 0.0 to 1.0
    source_episodes UUID[] DEFAULT '{}',     -- IDs of supporting episodes
    episode_count   INT DEFAULT 1,
    
    -- Validity
    valid_from      TIMESTAMPTZ DEFAULT NOW(),
    valid_until     TIMESTAMPTZ,             -- NULL means still valid
    is_active       BOOLEAN DEFAULT TRUE,
    
    -- Full-text for semantic search
    fact_narrative  TEXT,                    -- Human-readable statement
    embedding       VECTOR(1536),            -- pgvector
    
    created_at      TIMESTAMPTZ DEFAULT NOW(),
    updated_at      TIMESTAMPTZ DEFAULT NOW(),
    
    UNIQUE(entity_id, predicate, object)     -- One belief per predicate-object pair
);

CREATE INDEX ON semantic_facts USING ivfflat (embedding vector_cosine_ops);
CREATE INDEX ON semantic_facts (entity_id, is_active);
CREATE INDEX ON semantic_facts (predicate);
CREATE INDEX ON semantic_facts (confidence DESC) WHERE is_active = TRUE;

The Semantic Memory Store

import asyncpg
from pgvector.asyncpg import register_vector
from dataclasses import dataclass
from typing import Optional

@dataclass
class SemanticFact:
    fact_id: Optional[str]
    entity_id: str
    subject: str
    predicate: str
    object: str
    confidence: float
    fact_narrative: str
    source_episodes: List[str] = field(default_factory=list)
    episode_count: int = 1
    is_active: bool = True

class SemanticMemoryStore:
    """
    Structured fact storage with confidence tracking and provenance.
    
    Facts are extracted from episodes by the consolidation pipeline,
    never written directly from raw interactions.
    """
    
    def __init__(self, dsn: str, openai_client: OpenAI):
        self.dsn = dsn
        self.openai = openai_client
        self._pool = None
    
    async def connect(self):
        self._pool = await asyncpg.create_pool(self.dsn)
        async with self._pool.acquire() as conn:
            await register_vector(conn)
    
    def _embed(self, text: str) -> List[float]:
        response = self.openai.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding
    
    async def upsert_fact(self, fact: SemanticFact) -> str:
        """
        Insert or update a semantic fact.
        
        If a fact with the same entity_id + predicate + object exists,
        we increase its confidence and add the supporting episode.
        This is the core of the Bayesian belief update mechanism.
        """
        embedding = self._embed(fact.fact_narrative)
        
        async with self._pool.acquire() as conn:
            # Try to update existing fact first
            existing = await conn.fetchrow(
                """
                SELECT fact_id, confidence, episode_count, source_episodes
                FROM semantic_facts
                WHERE entity_id = $1 AND predicate = $2 AND object = $3
                  AND is_active = TRUE
                """,
                fact.entity_id, fact.predicate, fact.object
            )
            
            if existing:
                # Update confidence using a simple accumulation model:
                # confidence grows toward 1.0 with each corroborating episode
                new_count = existing["episode_count"] + 1
                new_confidence = 1.0 - (1.0 - existing["confidence"]) * 0.7
                new_episodes = list(set(
                    existing["source_episodes"] + fact.source_episodes
                ))
                
                await conn.execute(
                    """
                    UPDATE semantic_facts
                    SET confidence = $1,
                        episode_count = $2,
                        source_episodes = $3,
                        fact_narrative = $4,
                        embedding = $5,
                        updated_at = NOW()
                    WHERE fact_id = $6
                    """,
                    new_confidence, new_count, new_episodes,
                    fact.fact_narrative, embedding, existing["fact_id"]
                )
                return str(existing["fact_id"])
            else:
                row = await conn.fetchrow(
                    """
                    INSERT INTO semantic_facts 
                        (entity_id, subject, predicate, object, confidence,
                         source_episodes, episode_count, fact_narrative, embedding)
                    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
                    RETURNING fact_id
                    """,
                    fact.entity_id, fact.subject, fact.predicate, fact.object,
                    fact.confidence, fact.source_episodes, fact.episode_count,
                    fact.fact_narrative, embedding
                )
                return str(row["fact_id"])
    
    async def retrieve_facts(
        self,
        query: str,
        entity_id: str,
        limit: int = 10,
        min_confidence: float = 0.4,
    ) -> List[SemanticFact]:
        """
        Retrieve relevant facts for an entity using semantic search.
        Only returns active facts above the confidence threshold.
        """
        embedding = self._embed(query)
        
        async with self._pool.acquire() as conn:
            rows = await conn.fetch(
                """
                SELECT *, 
                       1 - (embedding <=> $1) AS similarity
                FROM semantic_facts
                WHERE entity_id = $2
                  AND is_active = TRUE
                  AND confidence >= $3
                ORDER BY similarity DESC
                LIMIT $4
                """,
                embedding, entity_id, min_confidence, limit
            )
        
        return [self._row_to_fact(r) for r in rows]
    
    async def get_entity_profile(self, entity_id: str) -> Dict[str, Any]:
        """
        Returns a structured profile of all high-confidence facts about an entity.
        Used for building the initial context block in working memory.
        """
        async with self._pool.acquire() as conn:
            rows = await conn.fetch(
                """
                SELECT predicate, object, confidence, fact_narrative
                FROM semantic_facts
                WHERE entity_id = $1 
                  AND is_active = TRUE
                  AND confidence >= 0.6
                ORDER BY confidence DESC
                """,
                entity_id
            )
        
        profile = {}
        for row in rows:
            profile[row["predicate"]] = {
                "value": row["object"],
                "confidence": row["confidence"],
                "narrative": row["fact_narrative"]
            }
        return profile
    
    async def invalidate_fact(self, entity_id: str, predicate: str, reason: str):
        """
        Mark a fact as no longer valid. Never delete — maintain full provenance.
        """
        async with self._pool.acquire() as conn:
            await conn.execute(
                """
                UPDATE semantic_facts
                SET is_active = FALSE,
                    valid_until = NOW(),
                    metadata = metadata || $1
                WHERE entity_id = $2 AND predicate = $3 AND is_active = TRUE
                """,
                json.dumps({"invalidation_reason": reason}),
                entity_id, predicate
            )
    
    def _row_to_fact(self, row) -> SemanticFact:
        return SemanticFact(
            fact_id=str(row["fact_id"]),
            entity_id=row["entity_id"],
            subject=row["subject"],
            predicate=row["predicate"],
            object=row["object"],
            confidence=row["confidence"],
            fact_narrative=row["fact_narrative"],
            source_episodes=list(row["source_episodes"]),
            episode_count=row["episode_count"],
            is_active=row["is_active"],
        )

Warning — The Confidence Inflation Trap: Notice our confidence update formula: new_confidence = 1.0 - (1.0 - old_confidence) * 0.7. This lets confidence grow toward 1.0 asymptotically — each new corroborating episode reduces the remaining uncertainty by 30%. Don't use simple averaging or confidence will plateau too quickly. Equally important: consider implementing a confidence decay for facts that haven't been seen in a long time. A preference that was true six months ago may not be true today.

---

Layer 3: Working Memory — The Active Context Manager

Working memory is the scratchpad for active reasoning. It's what gets directly injected into the LLM's context window for each invocation. The core challenge is that your context window is finite, your working memory items are not, and the cost of getting the wrong things in context is high.

The naive approach — prepend everything and let the model sort it out — fails in three ways: it hits context limits, it buries the most relevant signals in noise, and it dramatically increases latency and cost.

from dataclasses import dataclass, field
from enum import Enum
from typing import Callable
import heapq

class MemoryItemType(Enum):
    ENTITY_PROFILE = "entity_profile"      # High-confidence semantic facts
    RECENT_EPISODE = "recent_episode"      # Relevant past interactions
    CURRENT_TASK = "current_task"          # What we're currently doing
    TOOL_RESULT = "tool_result"            # Results from tool calls
    AGENT_THOUGHT = "agent_thought"        # Intermediate reasoning
    CONSTRAINT = "constraint"              # Hard rules/instructions

@dataclass
class WorkingMemoryItem:
    item_id: str
    item_type: MemoryItemType
    content: str
    priority: float           # 0.0 to 1.0, higher = more important
    token_estimate: int       # Rough token count
    created_at: datetime = field(default_factory=datetime.utcnow)
    expires_at: Optional[datetime] = None
    
    def is_expired(self) -> bool:
        if self.expires_at is None:
            return False
        return datetime.utcnow() > self.expires_at
    
    def __lt__(self, other):
        # For heap operations: lower priority = should be evicted first
        return self.priority < other.priority

class WorkingMemory:
    """
    Priority-managed context window for active agent reasoning.
    
    Maintains a fixed token budget and uses priority-based eviction
    to keep the most relevant information in the active context.
    """
    
    # Reserved slots by type (priorities)
    TYPE_PRIORITIES = {
        MemoryItemType.CONSTRAINT: 1.0,       # Never evict hard constraints
        MemoryItemType.CURRENT_TASK: 0.95,    # Current task is almost as sacred
        MemoryItemType.ENTITY_PROFILE: 0.8,   # Entity profile is highly stable
        MemoryItemType.TOOL_RESULT: 0.7,      # Recent tool results are important
        MemoryItemType.RECENT_EPISODE: 0.5,   # Episodes are evictable under pressure
        MemoryItemType.AGENT_THOUGHT: 0.3,    # Old thoughts are most evictable
    }
    
    def __init__(self, token_budget: int = 6000):
        self.token_budget = token_budget
        self._items: Dict[str, WorkingMemoryItem] = {}
        self._current_tokens: int = 0
    
    def add(self, item: WorkingMemoryItem) -> bool:
        """
        Add an item, evicting lower-priority items if necessary.
        Returns True if item was successfully added.
        """
        # Apply type-based priority floor
        item.priority = max(item.priority, self.TYPE_PRIORITIES.get(item.item_type, 0.0))
        
        # Remove expired items first
        self._evict_expired()
        
        # If item already exists, update it
        if item.item_id in self._items:
            old = self._items[item.item_id]
            self._current_tokens -= old.token_estimate
            self._items[item.item_id] = item
            self._current_tokens += item.token_estimate
            return True
        
        # Check if we have budget
        if self._current_tokens + item.token_estimate <= self.token_budget:
            self._items[item.item_id] = item
            self._current_tokens += item.token_estimate
            return True
        
        # Need to evict — find lowest priority items
        return self._evict_and_add(item)
    
    def _evict_and_add(self, new_item: WorkingMemoryItem) -> bool:
        """Evict lowest-priority items until we have budget for new_item."""
        # Sort current items by priority (ascending — lowest first for eviction)
        evictable = sorted(
            [item for item in self._items.values() 
             if item.priority < new_item.priority],
            key=lambda x: x.priority
        )
        
        freed_tokens = 0
        evicted_ids = []
        
        for candidate in evictable:
            if self._current_tokens - freed_tokens + new_item.token_estimate <= self.token_budget:
                break
            freed_tokens += candidate.token_estimate
            evicted_ids.append(candidate.item_id)
        
        # Check if we freed enough space
        if self._current_tokens - freed_tokens + new_item.token_estimate > self.token_budget:
            # Even after evicting everything possible, not enough space
            # Only reject if new item is lower priority than what remains
            return False
        
        for item_id in evicted_ids:
            del self._items[item_id]
        self._current_tokens -= freed_tokens
        
        self._items[new_item.item_id] = new_item
        self._current_tokens += new_item.token_estimate
        return True
    
    def _evict_expired(self):
        expired = [k for k, v in self._items.items() if v.is_expired()]
        for k in expired:
            self._current_tokens -= self._items[k].token_estimate
            del self._items[k]
    
    def render_context(self) -> str:
        """
        Render working memory as an ordered context block for LLM injection.
        Higher-priority items appear first.
        """
        ordered = sorted(
            self._items.values(),
            key=lambda x: (x.priority, x.created_at),
            reverse=True
        )
        
        sections = []
        type_order = [
            MemoryItemType.CONSTRAINT,
            MemoryItemType.CURRENT_TASK,
            MemoryItemType.ENTITY_PROFILE,
            MemoryItemType.RECENT_EPISODE,
            MemoryItemType.TOOL_RESULT,
            MemoryItemType.AGENT_THOUGHT,
        ]
        
        for item_type in type_order:
            items_of_type = [i for i in ordered if i.item_type == item_type]
            if not items_of_type:
                continue
            
            header = f"## {item_type.value.replace('_', ' ').title()}"
            content_blocks = [f"- {item.content}" for item in items_of_type]
            sections.append(header + "\n" + "\n".join(content_blocks))
        
        return "\n\n".join(sections)
    
    def get_token_utilization(self) -> Dict[str, int]:
        """Useful for monitoring and debugging."""
        by_type = {}
        for item in self._items.values():
            key = item.item_type.value
            by_type[key] = by_type.get(key, 0) + item.token_estimate
        return {"total": self._current_tokens, "budget": self.token_budget, **by_type}

---

The Consolidation Pipeline — Making Memory Coherent

The three layers only work together through a consolidation pipeline that runs asynchronously. This is the piece most implementations skip, and it's what separates a truly intelligent memory system from a slightly smarter context stuffer.

The consolidation pipeline does two things:

1. Episode Summarization: Generates semantic summaries for new episodes
2. Fact Extraction: Uses an LLM to extract semantic facts from new episodes and upserts them into semantic memory

import asyncio
from openai import AsyncOpenAI

class MemoryConsolidationPipeline:
    """
    Asynchronous pipeline that converts raw episodic memories
    into structured semantic knowledge.
    
    Run this as a background task or scheduled job, not in the
    hot path of your agent's request handling.
    """
    
    FACT_EXTRACTION_PROMPT = """
You are a knowledge extraction system. Given an interaction episode, extract factual 
statements about the entity involved. Focus on stable, generalizable facts — not 
one-time events.

EPISODE:
{episode_content}

ENTITY: {entity_type} {entity_id}

Extract 3-7 facts in the following JSON format:
{{
  "facts": [
    {{
      "predicate": "short_snake_case_predicate",
      "object": "the value or description",
      "confidence": 0.0-1.0,
      "narrative": "Full sentence: [Entity] [predicate] [object]."
    }}
  ]
}}

Only extract facts with confidence >= 0.4. Focus on preferences, patterns, constraints, 
and characteristics that would be useful to know in future interactions.

Predicates should be consistent and reusable across entities:
Good predicates: "prefers_contact_channel", "has_recurring_issue_type", "risk_tolerance_level"
Bad predicates: "mentioned_email_once", "seemed_frustrated_tuesday"
"""
    
    def __init__(
        self,
        episodic_store: EpisodicMemoryStore,
        semantic_store: SemanticMemoryStore,
        openai_client: AsyncOpenAI,
        batch_size: int = 10,
    ):
        self.episodic = episodic_store
        self.semantic = semantic_store
        self.openai = openai_client
        self.batch_size = batch_size
    
    async def consolidate_episode(self, episode: Episode) -> List[SemanticFact]:
        """
        Process a single episode: summarize it and extract semantic facts.
        Returns the list of facts that were extracted and stored.
        """
        # Step 1: Generate a concise summary if not already present
        if not episode.summary:
            episode.summary = await self._generate_summary(episode)
            await self.episodic.update_summary(episode.episode_id, episode.summary)
        
        # Step 2: Extract semantic facts
        facts = await self._extract_facts(episode)
        
        # Step 3: Store facts in semantic memory
        for fact in facts:
            await self.semantic.upsert_fact(fact)
        
        return facts
    
    async def _generate_summary(self, episode: Episode) -> str:
        """Generate a concise, retrieval-optimized summary of the episode."""
        response = await self.openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "Summarize this interaction in 2-3 sentences. "
                        "Focus on: what was the issue/task, what action was taken, "
                        "and what was the outcome. Be specific and factual."
                    )
                },
                {"role": "user", "content": episode.content[:4000]}
            ],
            max_tokens=150
        )
        return response.choices[0].message.content
    
    async def _extract_facts(self, episode: Episode) -> List[SemanticFact]:
        """Use LLM to extract structured facts from episode content."""
        prompt = self.FACT_EXTRACTION_PROMPT.format(
            episode_content=episode.content[:3000],
            entity_type=episode.entity_type,
            entity_id=episode.entity_id,
        )
        
        response = await self.openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            max_tokens=800
        )
        
        try:
            data = json.loads(response.choices[0].message.content)
            facts = []
            for f in data.get("facts", []):
                if f.get("confidence", 0) < 0.4:
                    continue
                facts.append(SemanticFact(
                    fact_id=None,
                    entity_id=episode.entity_id,
                    subject=f"{episode.entity_type}:{episode.entity_id}",
                    predicate=f["predicate"],
                    object=f["object"],
                    confidence=f["confidence"],
                    fact_narrative=f["narrative"],
                    source_episodes=[episode.episode_id],
                ))
            return facts
        except (json.JSONDecodeError, KeyError) as e:
            # Consolidation failures should not crash the system
            print(f"Fact extraction failed for episode {episode.episode_id}: {e}")
            return []
    
    async def run_batch(self, episodes: List[Episode]) -> Dict[str, int]:
        """Process a batch of episodes concurrently with rate limiting."""
        semaphore = asyncio.Semaphore(5)  # Max 5 concurrent LLM calls
        
        async def process_with_semaphore(ep):
            async with semaphore:
                return await self.consolidate_episode(ep)
        
        results = await asyncio.gather(
            *[process_with_semaphore(ep) for ep in episodes],
            return_exceptions=True
        )
        
        successes = sum(1 for r in results if not isinstance(r, Exception))
        failures = sum(1 for r in results if isinstance(r, Exception))
        total_facts = sum(len(r) for r in results if not isinstance(r, Exception))
        
        return {"processed": successes, "failed": failures, "facts_extracted": total_facts}

---

The Memory Orchestrator — Tying It All Together

Now we wire all three layers into a single interface that an agent actually calls:

class AgentMemoryOrchestrator:
    """
    The single interface an agent uses for all memory operations.
    
    Handles the read/write routing across episodic, semantic, and 
    working memory layers, and triggers consolidation as a side effect.
    """
    
    def __init__(
        self,
        episodic: EpisodicMemoryStore,
        semantic: SemanticMemoryStore,
        working: WorkingMemory,
        consolidation: MemoryConsolidationPipeline,
        openai_client: OpenAI,
    ):
        self.episodic = episodic
        self.semantic = semantic
        self.working = working
        self.consolidation = consolidation
        self.openai = openai_client
    
    async def prime_context(self, entity_id: str, entity_type: str, current_query: str):
        """
        Called at the start of each agent turn.
        Loads relevant memory into working memory based on the current query.
        """
        # 1. Load entity profile from semantic memory (high priority, stable)
        profile = await self.semantic.get_entity_profile(entity_id)
        if profile:
            profile_text = self._format_profile(profile, entity_type, entity_id)
            self.working.add(WorkingMemoryItem(
                item_id=f"profile_{entity_id}",
                item_type=MemoryItemType.ENTITY_PROFILE,
                content=profile_text,
                priority=0.85,
                token_estimate=self._estimate_tokens(profile_text),
            ))
        
        # 2. Retrieve relevant episodes (lower priority, contextual)
        relevant_episodes = self.episodic.retrieve(
            query=current_query,
            entity_id=entity_id,
            limit=5,
            recency_weight=0.25,
        )
        
        for episode, score in relevant_episodes:
            if score < 0.3:  # Skip low-relevance episodes
                continue
            episode_text = episode.to_context_block()
            self.working.add(WorkingMemoryItem(
                item_id=f"ep_{episode.episode_id}",
                item_type=MemoryItemType.RECENT_EPISODE,
                content=episode_text,
                priority=0.4 + score * 0.3,  # Scale priority by relevance
                token_estimate=self._estimate_tokens(episode_text),
                expires_at=datetime.utcnow().replace(hour=23, minute=59),
            ))
        
        # 3. Retrieve relevant semantic facts for the specific query
        relevant_facts = await self.semantic.retrieve_facts(
            query=current_query,
            entity_id=entity_id,
            limit=5,
            min_confidence=0.5,
        )
        
        for fact in relevant_facts:
            self.working.add(WorkingMemoryItem(
                item_id=f"fact_{fact.fact_id}",
                item_type=MemoryItemType.ENTITY_PROFILE,
                content=fact.fact_narrative,
                priority=0.6 + fact.confidence * 0.2,
                token_estimate=self._estimate_tokens(fact.fact_narrative),
            ))
    
    async def record_interaction(
        self,
        entity_id: str,
        entity_type: str,
        content: str,
        interaction_type: str,
        actions_taken: List[str],
        outcome: str,
        metadata: Dict[str, Any] = None,
        trigger_consolidation: bool = True,
    ) -> Episode:
        """
        Record a completed interaction and optionally trigger consolidation.
        This is called at the end of each agent turn.
        """
        episode = Episode(
            agent_id="customer_success_agent",
            entity_id=entity_id,
            entity_type=entity_type,
            content=content,
            interaction_type=interaction_type,
            actions_taken=actions_taken,
            outcome=outcome,
            importance_score=self._calculate_importance(outcome, metadata or {}),
            metadata=metadata or {},
        )
        
        await self.episodic.store(episode)
        
        # Trigger consolidation asynchronously — don't block the agent
        if trigger_consolidation:
            asyncio.create_task(self.consolidation.consolidate_episode(episode))
        
        return episode
    
    def get_context_for_llm(self) -> str:
        """Returns the current working memory as a formatted context block."""
        return self.working.render_context()
    
    def _calculate_importance(self, outcome: str, metadata: dict) -> float:
        """Heuristic importance scoring based on outcome and metadata signals."""
        base = 0.5
        
        if outcome in ("escalated", "unresolved", "critical_bug"):
            base += 0.3
        elif outcome == "resolved":
            base += 0.1
        
        if metadata.get("customer_tier") == "enterprise":
            base += 0.1
        if metadata.get("churn_risk", False):
            base += 0.2
        
        return min(base, 1.0)
    
    def _format_profile(self, profile: dict, entity_type: str, entity_id: str) -> str:
        lines = [f"Entity Profile: {entity_type} {entity_id}"]
        for predicate, data in profile.items():
            conf_indicator = "✓" if data["confidence"] > 0.8 else "~"
            lines.append(f"  {conf_indicator} {predicate}: {data['value']}")
        return "\n".join(lines)
    
    def _estimate_tokens(self, text: str) -> int:
        """Rough token estimate: ~4 chars per token for English text."""
        return max(1, len(text) // 4)

---

Hands-On Exercise

Build a memory-augmented support agent for a fictional SaaS product called DataFlow. Your agent should handle support tickets and demonstrate genuine memory across sessions.

Exercise Setup

import asyncio
from datetime import datetime, timedelta

async def run_memory_exercise():
    """
    Simulate a series of interactions with customer 'acme_corp'
    and verify that memory accumulates correctly across sessions.
    """
    
    # Initialize components (use your actual connection strings)
    openai_client = OpenAI(api_key="your-key")
    async_openai = AsyncOpenAI(api_key="your-key")
    
    episodic = EpisodicMemoryStore(
        qdrant_url="http://localhost:6333",
        openai_client=openai_client
    )
    semantic = SemanticMemoryStore(
        dsn="postgresql://localhost/agent_memory",
        openai_client=openai_client
    )
    await semantic.connect()
    
    working = WorkingMemory(token_budget=4000)
    
    pipeline = MemoryConsolidationPipeline(
        episodic_store=episodic,
        semantic_store=semantic,
        openai_client=async_openai,
    )
    
    orchestrator = AgentMemoryOrchestrator(
        episodic=episodic,
        semantic=semantic,
        working=working,
        consolidation=pipeline,
        openai_client=openai_client,
    )
    
    # === WEEK 1: First interaction ===
    print("=== Week 1: Initial Support Ticket ===")
    await orchestrator.record_interaction(
        entity_id="acme_corp",
        entity_type="customer",
        content="""
        Customer reported that their ETL pipeline using DataFlow's batch connector 
        fails consistently after processing exactly 10,000 rows. Error: MemoryError 
        in partition handler. Customer runs pipelines at 2 AM daily. Prefer email 
        for updates, not Slack. Their infrastructure team lead is Marcus Chen.
        Resolution: Identified memory leak in v2.3.1 batch connector. 
        Temporary workaround provided: set max_batch_size=5000.
        """,
        interaction_type="support_ticket",
        actions_taken=["provided_workaround", "filed_bug_report", "sent_email_update"],
        outcome="partially_resolved",
        metadata={
            "customer_tier": "enterprise", 
            "issue_category": "performance",
            "product_area": "batch_connector",
            "ticket_id": "TK-4892"
        }
    )
    
    # Allow consolidation to run
    await asyncio.sleep(3)
    
    # === WEEK 2: New session — does the agent remember? ===
    print("\n=== Week 2: Follow-up Session ===")
    # Clear working memory (simulate new session)
    orchestrator.working = WorkingMemory(token_budget=4000)
    
    new_query = "Acme Corp is reporting pipeline failures again. What do we know?"
    await orchestrator.prime_context("acme_corp", "customer", new_query)
    
    context = orchestrator.get_context_for_llm()
    print("Context loaded into working memory:")
    print(context)
    print(f"\nToken utilization: {orchestrator.working.get_token_utilization()}")

asyncio.run(run_memory_exercise())

What to verify:

1. After the first interaction, check that a semantic fact with predicate prefers_contact_channel = email was created
2. In week 2, verify the entity profile appears in the working memory context
3. Check that the episode from week 1 is retrieved as relevant to "pipeline failures"
4. Add a second interaction where you discover Acme Corp has migrated to a new platform — verify invalidate_fact correctly marks their old infrastructure facts as stale

---

Production Considerations

Memory Poisoning and Security

The most underappreciated risk in agent memory systems is injection attacks through memory. If user-provided content is stored verbatim as episodic memory and later retrieved into context, a malicious user can craft inputs that pollute future agent behavior.

Mitigations:

• Never store raw user input — always store the agent's interpretation of the interaction
• Apply content filtering before storage: check for prompt injection patterns
• Use confidence thresholds aggressively: a single poisoned episode shouldn't overwhelm accumulated legitimate knowledge
• Tag memory provenance: source: "user_input" vs source: "agent_observation" — and weight them differently

Consistency Guarantees

Your episodic store (Qdrant) and semantic store (PostgreSQL) have different consistency models. The consolidation pipeline creates a window where an episode exists in episodic memory but its derived facts haven't yet been written to semantic memory.

Design implications:

• Always retrieve from both layers when priming context — don't assume semantic memory is always up to date
• For critical facts (e.g., "customer account is suspended"), bypass consolidation and write directly to semantic memory
• Consider using a message queue (Redis Streams or Kafka) for the consolidation pipeline rather than asyncio.create_task — this survives process restarts

Scaling Patterns

For single-agent, single-entity scenarios, this architecture is already solid. For multi-agent, multi-tenant production systems:

• Shard by entity_id: Both Qdrant and PostgreSQL should be sharded on entity_id to keep entity data co-located
• Working memory is always process-local: Never share working memory across agent instances — each invocation builds its own from the persistent layers
• Use Redis for working memory persistence: If your agent needs to pause mid-task (multi-turn async), serialize working memory to Redis so it can be restored on the next invocation
• Rate-limit the consolidation pipeline per entity: If one entity generates a burst of episodes, you don't want consolidation to starve other entities of LLM quota

Token Budget Management

A common failure mode: the agent's memory system is well-designed, but token estimates are wildly wrong, causing either context overflow or severe underutilization. Use tiktoken for accurate estimation:

import tiktoken

_ENCODER = tiktoken.encoding_for_model("gpt-4o")

def accurate_token_count(text: str) -> int:
    return len(_ENCODER.encode(text))

Replace the character-count heuristic in WorkingMemory with this for production systems, especially if you're mixing languages or content types.

---

Common Mistakes & Troubleshooting

Mistake 1: Embedding Raw Conversation Transcripts

Symptom: Episodic retrieval returns episodes that are tonally similar but semantically irrelevant. Two conversations that were both "polite and technical" rank higher than one that actually discussed the same product issue.

Fix: Always embed the to_retrieval_text() representation, never raw content. Your retrieval text should be a compressed, keyword-rich semantic summary of what mattered.

Mistake 2: Running Consolidation Synchronously

Symptom: Agent response time degrades significantly after the first few interactions because every turn triggers an LLM call for fact extraction.

Fix: Consolidation should never block agent response. Use asyncio.create_task(), a background worker, or a message queue. The agent returns its response immediately; facts get extracted in parallel.

Mistake 3: Forgetting to Invalidate Stale Facts

Symptom: The agent confidently makes decisions based on a fact that was true six months ago but has since changed (customer's preferred contact, their team lead, their infrastructure details).

Fix: Whenever a new episode contradicts an existing semantic fact, log it explicitly and trigger invalidate_fact. The easiest way: include contradiction detection in your fact extraction prompt by providing current active facts for the entity as context.

Mistake 4: No Entity Profile "Cold Start" Strategy

Symptom: New entities have no semantic memory, and the agent's first interaction is pure zero-shot without any context — even when contextual information is available from external systems.

Fix: Pre-populate semantic memory from external CRM data for known entities before their first agent interaction. Write SemanticFact records from structured data (CRM fields, subscription data, etc.) with a source tag and initial confidence of 0.6. This gives the agent a useful starting point.

Mistake 5: Retrieval Without Diversity Enforcement

Symptom: Episodic retrieval returns five episodes from the same week, all about the same incident, because they're semantically very similar to each other. You get high recall on one topic and zero coverage of everything else.

Fix: Implement Maximal Marginal Relevance (MMR) for episode retrieval. After retrieving the initial candidate set, greedily select episodes that are both relevant to the query AND maximally different from already-selected episodes. This ensures breadth of coverage.

def maximal_marginal_relevance(
    query_embedding: List[float],
    candidates: List[Tuple[Episode, List[float], float]],  # episode, embedding, score
    k: int,
    lambda_param: float = 0.6,
) -> List[Episode]:
    """
    lambda_param: trade-off between relevance (1.0) and diversity (0.0)
    """
    import numpy as np
    
    selected = []
    remaining = list(candidates)
    
    while len(selected) < k and remaining:
        if not selected:
            # First selection: pick highest relevance
            best = max(remaining, key=lambda x: x[2])
        else:
            # Subsequent: balance relevance vs dissimilarity to selected
            selected_embeddings = np.array([s[1] for s in selected])
            
            def mmr_score(candidate):
                rel = candidate[2]
                emb = np.array(candidate[1])
                sim_to_selected = max(
                    np.dot(emb, sel_emb) / (np.linalg.norm(emb) * np.linalg.norm(sel_emb) + 1e-8)
                    for sel_emb in selected_embeddings
                )
                return lambda_param * rel - (1 - lambda_param) * sim_to_selected
            
            best = max(remaining, key=mmr_score)
        
        selected.append(best)
        remaining.remove(best)
    
    return [ep for ep, _, _ in selected]

---

Summary & Next Steps

You've built a production-grade, three-layer memory system for LLM agents. Let's recap the essential architecture decisions and why they matter:

Episodic memory gives your agent a retrievable journal of events. The key innovations are: embedding retrieval-optimized text rather than raw content, combining semantic similarity with temporal recency in a single scoring function, and storing rich metadata for filtered retrieval.

Semantic memory stores extracted, validated knowledge in a structured form with confidence tracking and provenance. The critical insight: facts must be earned through consolidation from multiple corroborating episodes. Single observations shouldn't directly update beliefs.

Working memory is a strict token-budget context manager with priority-based eviction. The key discipline: always have a principled answer to "what gets dropped when the context fills up?" Never let the answer be "whatever ran out of space first."

The consolidation pipeline is the glue that makes the system intelligent over time — transforming raw events into generalized knowledge asynchronously, without blocking agent response time.

Where to Go Next

1. Add a forgetting mechanism: Implement scheduled jobs that decay importance scores on old episodic memories and reduce confidence on facts not recently corroborated. Memory systems without forgetting degrade in quality over time.

2. Build memory-aware evaluation: Your agent's memory system is only useful if it improves outcomes. Set up A/B tests comparing memory-augmented vs. baseline agent interactions, and instrument retrieval quality (did the retrieved episodes actually matter for the outcome?).

3. Explore graph-based semantic memory: As your fact store grows, consider migrating from a relational+vector hybrid to a true knowledge graph (Neo4j + vector indexes). Graph traversal lets you reason over relationships between entities, not just facts about individual entities.

4. Implement multi-agent memory sharing: In multi-agent systems, different agents should share the same episodic and semantic layers but maintain separate working memories. Study the coordination patterns for concurrent memory writes.

5. Memory for tool use: Extend the consolidation pipeline to learn from tool call results — tracking which tools work reliably for which classes of problems, and storing that as semantic memory that informs future tool selection.

The memory system you've built here is the foundation. The agents that will actually be useful in production — the ones that get better over time rather than starting from scratch every interaction — are built exactly this way.