AI SystemsSeptember 14, 2026 · 22 min read

Agent Memory Architectures: Short-Term to Procedural Memory

Q: What's the difference between agent memory and RAG?

RAG is semantic memory only - retrieving relevant documents from a knowledge base. Agent memory includes RAG plus episodic memory (past experiences), procedural memory (learned skills), and working memory (reasoning traces). RAG answers "what do I know?", while full memory systems answer "what have I experienced?" and "what have I learned works?"

Q: How much memory should an agent have?

For production AI agents: Short-term: 10-20 messages (4,000-8,000 tokens) Semantic: Unlimited via vector database Episodic: 500-1,000 most relevant episodes per user Procedural: Top 100 learned patterns Working: Minimal, task-specific only Balance memory size against retrieval latency and cost.

Q: Should I store all conversations in memory?

Store summaries, not raw conversations. Raw conversation storage grows unbounded and slows retrieval. Instead: 1. Maintain raw conversation in short-term memory 2. Extract key facts, decisions, and learnings 3. Store extracted insights in semantic/episodic memory 4. Archive raw conversations separately for compliance

Q: How do I prevent outdated memories?

Implement memory decay: Time-based: Weight recent memories higher in retrieval Relevance-based: Prune low-relevance memories periodically Explicit updates: Allow updating/deleting specific memories Version control: Track when knowledge was added/updated Mark memories with timestamps and last-accessed dates for pruning.

Q: What's the best vector database for agent memory?

Choose based on scale: Prototypes: Chroma, FAISS (local, simple) Production <1M vectors: Pinecone, Weaviate (managed) Production 1M vectors: Milvus, Qdrant (self-hosted, scalable) Enterprise: Pgvector (if already using PostgreSQL) All work with LangChain. See LangGraph framework comparison for integration patterns.

Q: How do I implement memory privacy?

Implement per-user memory isolation: Never mix user memories without explicit consent. Implement GDPR-compliant deletion.

Q: Should episodic memory store successes or failures?

Store both. Successes show what works; failures prevent repeated mistakes. Weight successes higher in retrieval but include relevant failures as warnings. Our pattern: Success episodes: Recommend similar approaches Failure episodes: Warn against repeating mistakes

Q: How do I test agent memory systems?

Test at multiple levels: 1. Unit tests: Individual memory operations (add, retrieve, update) 2. Integration tests: Memory systems working together 3. Retrieval tests: Relevant memories retrieved for queries 4. Performance tests: Retrieval latency at scale 5. Consistency tests: Memories persist across sessions See our guide on testing AI agents.

Q: What's the latency impact of memory retrieval?

Typical latency: Short-term: <1ms (in-memory) Semantic: 50-200ms (vector search) Episodic: 100-300ms (vector + DB query) Procedural: 20-50ms (DB query) Optimize with caching, parallel retrieval, and result limiting. Budget 200-400ms total for comprehensive memory retrieval.

Q: How do I implement memory in multi-agent systems?

Implement shared semantic memory (common knowledge) plus agent-specific episodic/procedural memory (individual experiences): See multi-agent orchestration patterns.

Agent memory architectures explained: short-term buffers, semantic, episodic, procedural and working memory, with LangGraph code and a decision table.

Muhammad Abdul Sami

September 14, 2026 · 22 min read

AI Agents
Agent Memory
LangGraph
Vector Databases
RAG

AI agents without memory are stateless tools, not intelligent systems. Production agent memory architectures combine short-term working memory, long-term semantic memory, episodic memory for experiences, and procedural memory for learned behaviors. This guide walks through each memory type, shows how to implement it with LangGraph and a vector store, and gives you a decision table for picking the right combination for your agent.

Key Takeaways:

Short-term memory is a token budget problem: summarize old turns before you hit the context limit, and keep the last 3-5 messages verbatim.

Semantic memory is RAG. If you already have a retrieval pipeline, reuse it instead of building a second vector index for "memory".

Episodic memory only pays off when you record outcomes (success/failure), not just transcripts. Without outcome labels the agent cannot learn what worked.

Procedural memory should be gated by a minimum sample size (3+ successes) before the agent is steered toward a tool sequence; otherwise one lucky run becomes a bad habit.

Consolidation is an offline job. Run it after the session ends, not inside the request path, and budget 200-400 ms for retrieval across all memory types at request time.

Start with short-term plus semantic memory; add episodic and procedural memory only when you can measure the improvement.

Table of Contents:

Why Agents Need Memory
Agent Memory Types Compared
Short-Term Memory
Long-Term Semantic Memory
Episodic Memory
Procedural Memory
Working Memory
Memory Retrieval Strategies
Memory Consolidation
Memory Storage Architecture
Memory in LangGraph
Performance Optimization
Choosing an Agent Memory Architecture
Agent Memory Failure Modes
Frequently Asked Questions

Why Agents Need Memory

Memory transforms agents from reactive tools to intelligent systems that learn, adapt, and maintain context across interactions.

Without memory:

❌ Agent forgets previous conversations
❌ Repeats failed approaches
❌ Ignores learned user preferences
❌ Cannot reference past decisions
❌ Lacks situational awareness

With memory:

✅ Maintains conversation context
✅ Learns from mistakes
✅ Personalizes to user preferences
✅ References past interactions
✅ Builds cumulative knowledge

Our production AI agent systems use all five memory types, but rarely all at once in a single agent. The trick is knowing which type solves which problem.

The vocabulary here comes from cognitive science and was popularized for LLM agents by the Generative Agents paper (memory stream, reflection, retrieval scored by recency, importance and relevance) and by MemGPT, which treats the context window as RAM and external storage as disk, with the LLM paging memories in and out. Both ideas show up in the implementations below.

Agent Memory Types Compared

The five memory types differ in where they live, how long they persist, and how they are retrieved. That last column matters most for engineering: retrieval strategy dictates latency and cost.

Memory type	What it stores	Lifetime	Backing store	Retrieval	Typical latency
Short-term	Current messages, tool calls, intermediate results	One session	In-process list / checkpointer	Sequential (whole buffer)	<1 ms
Semantic	Facts, docs, domain knowledge	Persistent	Vector DB (pgvector, Qdrant, Pinecone)	Embedding similarity + metadata filters	50-200 ms
Episodic	Past interactions with outcomes	Persistent, pruned	Document DB + embeddings	Similarity on context, filtered by user	100-300 ms
Procedural	Successful tool sequences, user preferences	Persistent, updated incrementally	Relational / document DB	Keyed lookup by context type	20-50 ms
Working	Plan, current step, scratch values	One task	Graph state object	Direct key access	<1 ms

Latencies are illustrative for a well-indexed store in the same region; they are the numbers to budget against, not guarantees.

Modeled after human cognitive architecture:

1. Short-Term Memory (Conversation Buffer)

Holds current interaction context - messages, tool calls, intermediate results.
Duration: Current session only
Capacity: ~4,000-8,000 tokens (LLM context window)
Use case: Maintaining conversation flow

2. Long-Term Semantic Memory (Knowledge Base)

Stores facts, concepts, and relationships - product docs, company knowledge, domain expertise.
Duration: Persistent across all sessions
Capacity: Unlimited (vector database)
Use case: RAG, question answering, knowledge retrieval

3. Episodic Memory (Experience Log)

Records specific events and interactions - past conversations, decisions made, outcomes observed.
Duration: Persistent, with decay/pruning
Capacity: Filtered by relevance
Use case: Learning from experience, avoiding repeated mistakes

4. Procedural Memory (Learned Behaviors)

Encodes skills and learned procedures - successful tool sequences, problem-solving patterns, user preferences.
Duration: Persistent, updated incrementally
Capacity: Structured storage (database)
Use case: Optimization, personalization

5. Working Memory (Scratch Space)

Temporary computation space - intermediate calculations, partial results, reasoning traces.
Duration: Current task only
Capacity: Minimal, task-specific
Use case: Multi-step reasoning, planning

Let's implement each type with production patterns.

Short-Term Memory

Short-term memory maintains conversation context within LLM context window:

python

from typing import List, Dict
from collections import deque

class ShortTermMemory:
    """Conversation buffer with token management."""
    
    def __init__(
        self,
        max_messages: int = 10,
        max_tokens: int = 4000,
        summary_threshold: int = 3500
    ):
        self.max_messages = max_messages
        self.max_tokens = max_tokens
        self.summary_threshold = summary_threshold
        self.messages = deque(maxlen=max_messages)
        self.token_count = 0
    
    def add_message(self, message: Dict):
        """Add message with token counting."""
        tokens = estimate_tokens(message["content"])
        if self.token_count + tokens > self.summary_threshold:
            self.summarize_and_compress()
        
        self.messages.append(message)
        self.token_count += tokens
    
    def summarize_and_compress(self):
        """Compress old messages into summary."""
        if len(self.messages) < 4:
            return
        
        # Extract messages to summarize (keep recent 3)
        to_summarize = list(self.messages)[:-3]
        
        summary_prompt = f"""Summarize this conversation concisely:
        {format_messages(to_summarize)}
        
        Focus on: key decisions, user preferences, important facts."""
        
        summary = llm.invoke([
            {"role": "system", "content": "Create concise summary."},
            {"role": "user", "content": summary_prompt}
        ])
        
        # Replace old messages with summary
        self.messages.clear()
        self.messages.append({
            "role": "system",
            "content": f"Previous conversation summary: {summary.content}"
        })
        
        # Re-add recent messages
        for msg in list(to_summarize)[-3:]:
            self.messages.append(msg)
        
        # Recalculate tokens
        self.token_count = sum(
            estimate_tokens(msg["content"]) for msg in self.messages
        )
    
    def get_context(self) -> List[Dict]:
        """Get conversation history for LLM."""
        return list(self.messages)
    
    def clear(self):
        """Clear short-term memory."""
        self.messages.clear()
        self.token_count = 0

def estimate_tokens(text: str) -> int:
    """Rough token estimation (1 token ≈ 4 chars)."""
    return len(text) // 4

LangGraph Short-Term Memory

LangGraph provides built-in message memory:

python

from langgraph.graph import StateGraph, MessagesState
from typing import Annotated
import operator

class AgentState(MessagesState):
    """State with message history."""
    # messages field provided by MessagesState
    # Uses Annotated[list, operator.add] for appending
    iteration: int = 0

def agent_node(state: AgentState):
    """Agent with automatic message history."""
    # state["messages"] contains full conversation
    response = llm.invoke(state["messages"])
    
    # Automatically appended to message list
    return {"messages": [response]}

workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.set_entry_point("agent")
graph = workflow.compile()

# Conversation is maintained in state
result = graph.invoke({
    "messages": [{"role": "user", "content": "Hello"}]
})

This integrates with stateful agent architectures.

Long-Term Semantic Memory

Semantic memory stores retrievable knowledge using vector databases:

python

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Pinecone
from langchain.text_splitter import RecursiveCharacterTextSplitter
from pinecone import Pinecone as PineconeClient
import os

class SemanticMemory:
    """Long-term knowledge storage with vector search."""
    
    def __init__(self, index_name: str = "agent-knowledge"):
        # Initialize Pinecone
        pc = PineconeClient(api_key=os.getenv("PINECONE_API_KEY"))
        self.index = pc.Index(index_name)
        
        # Embeddings model
        self.embeddings = OpenAIEmbeddings()
        
        # Vector store
        self.vector_store = Pinecone(
            index=self.index,
            embedding=self.embeddings,
            text_key="text"
        )
        
        # Text splitter for chunking
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50
        )
    
    async def add_knowledge(
        self,
        text: str,
        metadata: Dict = None
    ):
        """Store new knowledge."""
        # Split into chunks
        chunks = self.splitter.split_text(text)
        
        # Add metadata to each chunk
        metadatas = [
            {
                **(metadata or {}),
                "chunk_id": i,
                "total_chunks": len(chunks),
                "added_at": datetime.now().isoformat()
            }
            for i in range(len(chunks))
        ]
        
        # Store in vector database
        await self.vector_store.aadd_texts(
            texts=chunks,
            metadatas=metadatas
        )
    
    async def retrieve(
        self,
        query: str,
        k: int = 5,
        filter: Dict = None
    ) -> List[Dict]:
        """Retrieve relevant knowledge."""
        results = await self.vector_store.asimilarity_search_with_score(
            query=query,
            k=k,
            filter=filter
        )
        
        return [
            {
                "content": doc.page_content,
                "metadata": doc.metadata,
                "score": score
            }
            for doc, score in results
        ]
    
    async def update_knowledge(
        self,
        doc_id: str,
        new_text: str,
        metadata: Dict = None
    ):
        """Update existing knowledge."""
        # Delete old version
        self.index.delete(ids=[doc_id])
        
        # Add updated version
        await self.add_knowledge(new_text, metadata)

# Usage in agent
semantic_memory = SemanticMemory()

async def retrieve_relevant_knowledge(query: str) -> str:
    """RAG pattern for knowledge retrieval."""
    results = await semantic_memory.retrieve(query, k=3)
    
    if not results:
        return "No relevant knowledge found."
    
    # Format for LLM context
    context = "\n\n".join([
        f"[Relevance: {r['score']:.2f}] {r['content']}"
        for r in results
    ])
    
    return f"Relevant knowledge:\n{context}"

Semantic Memory with Metadata Filtering

python

# Store knowledge with structured metadata
await semantic_memory.add_knowledge(
    text="Product X costs $99 and includes free shipping.",
    metadata={
        "type": "product_info",
        "product_id": "prod_x",
        "category": "pricing",
        "last_updated": "2026-09-14"
    }
)

# Retrieve with filtering
pricing_info = await semantic_memory.retrieve(
    query="How much does Product X cost?",
    k=3,
    filter={"type": "product_info", "category": "pricing"}
)

This pattern is fundamental to production AI agents.

Episodic Memory

Episodic memory records specific interactions and outcomes:

python

from dataclasses import dataclass, asdict
from typing import Optional
import json

@dataclass
class Episode:
    """Single interaction episode."""
    episode_id: str
    user_id: str
    timestamp: datetime
    context: str  # What was happening
    action: str   # What agent did
    outcome: str  # What resulted
    success: bool
    learned: Optional[str] = None  # What was learned
    
class EpisodicMemory:
    """Memory of past interactions and outcomes."""
    
    def __init__(self, db_connection):
        self.db = db_connection
        self.embeddings = OpenAIEmbeddings()
    
    async def record_episode(
        self,
        user_id: str,
        context: str,
        action: str,
        outcome: str,
        success: bool
    ) -> str:
        """Record new episode."""
        episode_id = str(uuid.uuid4())
        
        episode = Episode(
            episode_id=episode_id,
            user_id=user_id,
            timestamp=datetime.now(),
            context=context,
            action=action,
            outcome=outcome,
            success=success
        )
        
        # Store in database
        await self.db.episodes.insert_one(asdict(episode))
        
        # Also embed for semantic search
        embedding = await self.embeddings.aembed_query(
            f"{context} -> {action} -> {outcome}"
        )
        
        await self.db.episode_embeddings.insert_one({
            "episode_id": episode_id,
            "embedding": embedding,
            "user_id": user_id
        })
        
        return episode_id
    
    async def retrieve_similar_episodes(
        self,
        current_context: str,
        user_id: Optional[str] = None,
        k: int = 5
    ) -> List[Episode]:
        """Find similar past episodes."""
        # Embed current context
        query_embedding = await self.embeddings.aembed_query(current_context)
        
        # Vector similarity search
        filter = {"user_id": user_id} if user_id else {}
        
        similar = await self.db.episode_embeddings.aggregate([
            {"$match": filter},
            {
                "$addFields": {
                    "similarity": {
                        "$reduce": {
                            "input": {"$zip": {"inputs": ["$embedding", query_embedding]}},
                            "initialValue": 0,
                            "in": {"$add": ["$$value", {"$multiply": "$$this"}]}
                        }
                    }
                }
            },
            {"$sort": {"similarity": -1}},
            {"$limit": k}
        ]).to_list(length=k)
        
        # Fetch full episodes
        episode_ids = [e["episode_id"] for e in similar]
        episodes = await self.db.episodes.find(
            {"episode_id": {"$in": episode_ids}}
        ).to_list(length=k)
        
        return [Episode(**e) for e in episodes]
    
    async def learn_from_episode(
        self,
        episode_id: str,
        learning: str
    ):
        """Update episode with learning."""
        await self.db.episodes.update_one(
            {"episode_id": episode_id},
            {"$set": {"learned": learning}}
        )

# Usage in agent
episodic_memory = EpisodicMemory(db)

async def check_past_experience(context: str) -> Optional[str]:
    """Check if similar situation occurred before."""
    similar = await episodic_memory.retrieve_similar_episodes(
        current_context=context,
        k=3
    )
    
    if not similar:
        return None
    
    # Find successful past approaches
    successful = [e for e in similar if e.success]
    
    if successful:
        return f"""Similar situation encountered before:
        
        Context: {successful[0].context}
        Successful action: {successful[0].action}
        Outcome: {successful[0].outcome}
        
        Consider similar approach."""
    
    # Learn from failures
    failed = [e for e in similar if not e.success]
    if failed:
        return f"""Warning: Similar situation failed before:
        
        Context: {failed[0].context}
        Failed action: {failed[0].action}
        Outcome: {failed[0].outcome}
        
        Try different approach."""

Episodic Memory in Decision-Making

python

def agent_with_episodic_memory(state: AgentState):
    """Agent that learns from experience."""
    current_context = state["messages"][-1]["content"]
    
    # Check past experience
    experience = await check_past_experience(current_context)
    
    # Include in prompt
    messages = state["messages"].copy()
    if experience:
        messages.insert(0, {
            "role": "system",
            "content": f"Relevant past experience:\n{experience}"
        })
    
    response = llm.invoke(messages)
    return {"messages": [response]}

This enables agent reasoning patterns that improve over time.

Procedural Memory

Procedural memory encodes learned skills and patterns:

python

from collections import defaultdict
from typing import List, Tuple

class ProceduralMemory:
    """Memory of successful procedures and skills."""
    
    def __init__(self, db_connection):
        self.db = db_connection
        self.success_counts = defaultdict(int)
        self.failure_counts = defaultdict(int)
    
    async def record_tool_sequence(
        self,
        sequence: List[str],
        context: str,
        success: bool
    ):
        """Record tool call sequence and outcome."""
        sequence_key = " -> ".join(sequence)
        
        await self.db.tool_sequences.update_one(
            {
                "sequence": sequence_key,
                "context_type": classify_context(context)
            },
            {
                "$inc": {
                    "success_count" if success else "failure_count": 1
                },
                "$set": {
                    "last_used": datetime.now(),
                    "sequence_list": sequence
                }
            },
            upsert=True
        )
    
    async def get_recommended_sequence(
        self,
        context: str
    ) -> Optional[List[str]]:
        """Get most successful tool sequence for context."""
        context_type = classify_context(context)
        
        # Find sequences for this context type
        sequences = await self.db.tool_sequences.find({
            "context_type": context_type,
            "success_count": {"$gte": 3}  # At least 3 successes
        }).to_list(length=10)
        
        if not sequences:
            return None
        
        # Calculate success rates
        ranked = []
        for seq in sequences:
            total = seq["success_count"] + seq["failure_count"]
            success_rate = seq["success_count"] / total
            ranked.append((seq["sequence_list"], success_rate))
        
        # Return best sequence
        ranked.sort(key=lambda x: x[1], reverse=True)
        return ranked[0][0] if ranked else None
    
    async def record_user_preference(
        self,
        user_id: str,
        preference_type: str,
        preference_value: any
    ):
        """Store learned user preference."""
        await self.db.user_preferences.update_one(
            {"user_id": user_id},
            {
                "$set": {
                    f"preferences.{preference_type}": preference_value,
                    "updated_at": datetime.now()
                }
            },
            upsert=True
        )
    
    async def get_user_preferences(
        self,
        user_id: str
    ) -> Dict:
        """Retrieve learned preferences."""
        result = await self.db.user_preferences.find_one(
            {"user_id": user_id}
        )
        return result.get("preferences", {}) if result else {}

def classify_context(context: str) -> str:
    """Classify context type for pattern matching."""
    if "search" in context.lower():
        return "information_retrieval"
    elif "order" in context.lower():
        return "transaction"
    elif "schedule" in context.lower():
        return "calendar"
    else:
        return "general"

# Usage
procedural_memory = ProceduralMemory(db)

async def optimize_tool_sequence(state: AgentState):
    """Use learned procedures."""
    context = state["messages"][-1]["content"]
    
    # Get recommended approach
    recommended = await procedural_memory.get_recommended_sequence(context)
    
    if recommended:
        # Guide agent to use successful pattern
        return {
            "messages": [{
                "role": "system",
                "content": f"""Based on past successes, recommended approach:
                {' -> '.join(recommended)}
                
                Consider following this pattern."""
            }]
        }
    
    return state

Working Memory

Working memory provides scratch space for reasoning:

python

class WorkingMemory:
    """Temporary computation space."""
    
    def __init__(self):
        self.scratch = {}
        self.reasoning_trace = []
    
    def store_intermediate(self, key: str, value: any):
        """Store intermediate result."""
        self.scratch[key] = value
        self.reasoning_trace.append(f"Stored {key}: {value}")
    
    def retrieve_intermediate(self, key: str) -> Optional[any]:
        """Retrieve intermediate result."""
        value = self.scratch.get(key)
        if value:
            self.reasoning_trace.append(f"Retrieved {key}: {value}")
        return value
    
    def get_reasoning_trace(self) -> List[str]:
        """Get full reasoning trace."""
        return self.reasoning_trace.copy()
    
    def clear(self):
        """Clear working memory."""
        self.scratch.clear()
        self.reasoning_trace.clear()

# Usage in multi-step reasoning
class AgentState(TypedDict):
    messages: list
    working_memory: WorkingMemory
    result: Optional[dict]

def planning_node(state: AgentState):
    """Store plan in working memory."""
    plan = create_plan(state["messages"])
    
    wm = state.get("working_memory", WorkingMemory())
    wm.store_intermediate("plan", plan)
    wm.store_intermediate("current_step", 0)
    
    return {"working_memory": wm}

def execution_node(state: AgentState):
    """Execute using working memory."""
    wm = state["working_memory"]
    plan = wm.retrieve_intermediate("plan")
    current_step = wm.retrieve_intermediate("current_step")
    
    # Execute step
    result = execute_step(plan[current_step])
    wm.store_intermediate(f"step_{current_step}_result", result)
    wm.store_intermediate("current_step", current_step + 1)
    
    return {"working_memory": wm}

Memory Retrieval Strategies

Effective retrieval is critical:

Hybrid Search (Semantic + Keyword)

python

async def hybrid_retrieval(
    query: str,
    semantic_weight: float = 0.7,
    keyword_weight: float = 0.3,
    k: int = 5
) -> List[Dict]:
    """Combine semantic and keyword search."""
    
    # Semantic search (vector similarity)
    semantic_results = await semantic_memory.retrieve(query, k=k*2)
    
    # Keyword search (BM25 or similar)
    keyword_results = await keyword_search(query, k=k*2)
    
    # Combine and rerank
    combined_scores = {}
    
    for result in semantic_results:
        doc_id = result["metadata"].get("doc_id")
        combined_scores[doc_id] = semantic_weight * result["score"]
    
    for result in keyword_results:
        doc_id = result["metadata"].get("doc_id")
        combined_scores[doc_id] = combined_scores.get(doc_id, 0) + keyword_weight * result["score"]
    
    # Sort by combined score
    ranked = sorted(
        combined_scores.items(),
        key=lambda x: x[1],
        reverse=True
    )[:k]
    
    # Fetch full documents
    return await fetch_documents([doc_id for doc_id, _ in ranked])

Recency-Weighted Retrieval

python

async def recency_weighted_retrieval(
    query: str,
    k: int = 5,
    recency_weight: float = 0.3
) -> List[Dict]:
    """Boost recent memories."""
    results = await semantic_memory.retrieve(query, k=k*2)
    
    now = datetime.now()
    
    # Rerank with recency boost
    reranked = []
    for result in results:
        added_at = datetime.fromisoformat(result["metadata"]["added_at"])
        age_hours = (now - added_at).total_seconds() / 3600
        
        # Exponential decay: newer = higher boost
        recency_score = math.exp(-age_hours / 168)  # Half-life of 1 week
        
        combined_score = (
            (1 - recency_weight) * result["score"] +
            recency_weight * recency_score
        )
        
        reranked.append({
            **result,
            "combined_score": combined_score
        })
    
    reranked.sort(key=lambda x: x["combined_score"], reverse=True)
    return reranked[:k]

Memory Consolidation

Consolidate short-term memories into long-term:

python

async def consolidate_memories(session_id: str):
    """Convert session memories to long-term storage."""
    
    # Retrieve session interactions
    session = await db.sessions.find_one({"session_id": session_id})
    
    if not session:
        return
    
    # Extract key facts and decisions
    consolidation_prompt = f"""Analyze this interaction and extract:
    1. Key facts learned
    2. User preferences discovered
    3. Successful approaches
    4. Failures to avoid
    
    Session: {json.dumps(session["messages"], indent=2)}"""
    
    analysis = await llm.ainvoke([
        {"role": "system", "content": "Extract learnings from conversation."},
        {"role": "user", "content": consolidation_prompt}
    ])
    
    learnings = parse_learnings(analysis.content)
    
    # Store in appropriate memory systems
    for fact in learnings.get("facts", []):
        await semantic_memory.add_knowledge(
            text=fact,
            metadata={
                "type": "learned_fact",
                "session_id": session_id,
                "learned_at": datetime.now().isoformat()
            }
        )
    
    for preference in learnings.get("preferences", []):
        await procedural_memory.record_user_preference(
            user_id=session["user_id"],
            preference_type=preference["type"],
            preference_value=preference["value"]
        )
    
    for approach in learnings.get("successful_approaches", []):
        await episodic_memory.record_episode(
            user_id=session["user_id"],
            context=approach["context"],
            action=approach["action"],
            outcome=approach["outcome"],
            success=True
        )

Memory Storage Architecture

Production memory architecture:

python

from dataclasses import dataclass
from typing import Optional

@dataclass
class MemoryConfig:
    """Memory system configuration."""
    # Short-term
    max_conversation_messages: int = 20
    max_conversation_tokens: int = 4000
    
    # Semantic
    vector_index_name: str = "agent-knowledge"
    embedding_model: str = "text-embedding-3-small"
    chunk_size: int = 500
    chunk_overlap: int = 50
    
    # Episodic
    max_episodes_per_user: int = 1000
    episode_retention_days: int = 90
    
    # Procedural
    min_successes_for_pattern: int = 3
    pattern_confidence_threshold: float = 0.7

class UnifiedMemorySystem:
    """Integrated memory architecture."""
    
    def __init__(self, config: MemoryConfig, db_connection, vector_client):
        self.config = config
        
        # Initialize all memory types
        self.short_term = ShortTermMemory(
            max_messages=config.max_conversation_messages,
            max_tokens=config.max_conversation_tokens
        )
        
        self.semantic = SemanticMemory(
            index_name=config.vector_index_name
        )
        
        self.episodic = EpisodicMemory(db_connection)
        self.procedural = ProceduralMemory(db_connection)
        self.working = WorkingMemory()
    
    async def remember(
        self,
        query: str,
        user_id: Optional[str] = None,
        memory_types: List[str] = None
    ) -> Dict[str, any]:
        """Unified memory retrieval."""
        if memory_types is None:
            memory_types = ["semantic", "episodic", "procedural"]
        
        memories = {}
        
        if "semantic" in memory_types:
            memories["semantic"] = await self.semantic.retrieve(query)
        
        if "episodic" in memory_types and user_id:
            memories["episodic"] = await self.episodic.retrieve_similar_episodes(
                query, user_id
            )
        
        if "procedural" in memory_types:
            memories["procedural"] = await self.procedural.get_recommended_sequence(
                query
            )
        
        return memories
    
    async def consolidate(self, session_id: str):
        """Consolidate session memories."""
        await consolidate_memories(session_id)
    
    def get_context(self) -> Dict:
        """Get complete memory context for agent."""
        return {
            "conversation": self.short_term.get_context(),
            "working_memory": self.working.get_reasoning_trace()
        }

Memory in LangGraph

Integrate memory with LangGraph:

python

from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres import PostgresCheckpointer

class MemoryAgentState(TypedDict):
    messages: Annotated[list, operator.add]
    user_id: str
    memories: Dict
    iteration: int

async def memory_retrieval_node(state: MemoryAgentState):
    """Retrieve relevant memories."""
    query = state["messages"][-1]["content"]
    user_id = state["user_id"]
    
    # Retrieve from all memory systems
    memories = await unified_memory.remember(
        query=query,
        user_id=user_id
    )
    
    return {"memories": memories}

async def agent_with_memory_node(state: MemoryAgentState):
    """Agent using retrieved memories."""
    memories = state.get("memories", {})
    
    # Build context with memories
    context_parts = []
    
    if memories.get("semantic"):
        context_parts.append("Relevant knowledge:")
        for mem in memories["semantic"][:3]:
            context_parts.append(f"- {mem['content']}")
    
    if memories.get("episodic"):
        context_parts.append("\nPast similar experiences:")
        for episode in memories["episodic"][:2]:
            context_parts.append(
                f"- {episode.action} resulted in {episode.outcome}"
            )
    
    if memories.get("procedural"):
        context_parts.append("\nRecommended approach:")
        context_parts.append(f"- {' -> '.join(memories['procedural'])}")
    
    memory_context = "\n".join(context_parts)
    
    # Add to messages
    messages = state["messages"].copy()
    if memory_context:
        messages.insert(-1, {
            "role": "system",
            "content": memory_context
        })
    
    response = await llm.ainvoke(messages)
    return {"messages": [response]}

# Build graph
workflow = StateGraph(MemoryAgentState)
workflow.add_node("retrieve_memories", memory_retrieval_node)
workflow.add_node("agent", agent_with_memory_node)

workflow.add_edge("retrieve_memories", "agent")
workflow.add_edge("agent", END)
workflow.set_entry_point("retrieve_memories")

memory_graph = workflow.compile(
    checkpointer=PostgresCheckpointer(DATABASE_URL)
)

This provides stateful agents with persistent memory.

Performance Optimization

Optimize memory systems for production:

Caching

python

from functools import lru_cache
import asyncio

class CachedSemanticMemory(SemanticMemory):
    """Semantic memory with caching."""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.cache = {}
        self.cache_ttl = 300  # 5 minutes
    
    async def retrieve(
        self,
        query: str,
        k: int = 5,
        filter: Dict = None
    ) -> List[Dict]:
        """Retrieve with caching."""
        cache_key = f"{query}:{k}:{json.dumps(filter or {})}"
        
        # Check cache
        if cache_key in self.cache:
            cached, timestamp = self.cache[cache_key]
            if (datetime.now() - timestamp).seconds < self.cache_ttl:
                return cached
        
        # Cache miss - retrieve and cache
        results = await super().retrieve(query, k, filter)
        self.cache[cache_key] = (results, datetime.now())
        
        return results

Batch Operations

python

async def batch_memory_operations(operations: List[Dict]):
    """Execute memory operations in batch."""
    semantic_ops = [op for op in operations if op["type"] == "semantic"]
    episodic_ops = [op for op in operations if op["type"] == "episodic"]
    
    # Execute in parallel
    results = await asyncio.gather(
        process_semantic_batch(semantic_ops),
        process_episodic_batch(episodic_ops)
    )
    
    return results

Lazy Loading

python

class LazyMemorySystem:
    """Load memories only when accessed."""
    
    def __init__(self):
        self._semantic = None
        self._episodic = None
    
    @property
    def semantic(self):
        if self._semantic is None:
            self._semantic = SemanticMemory()
        return self._semantic
    
    @property
    def episodic(self):
        if self._episodic is None:
            self._episodic = EpisodicMemory(db)
        return self._episodic

Choosing an Agent Memory Architecture

Most teams over-build memory. A support bot that answers from a knowledge base needs short-term plus semantic memory and nothing else. Use the table below to decide what to add, and only add a layer when you can name the failure it fixes.

Agent type	Short-term	Semantic	Episodic	Procedural	Working	Why
Single-turn Q&A over docs	Minimal	Yes	No	No	No	Pure RAG; conversation history adds noise
Multi-turn support assistant	Yes (summarized)	Yes	Per-user, optional	Preferences only	No	Needs context continuity and personalization
Research / planning agent	Yes	Yes	No	No	Yes	Multi-step reasoning needs a scratchpad, not history
Autonomous ops agent (runbooks, tickets)	Yes	Yes	Yes	Yes	Yes	Learns which tool sequences resolve which incident types
Multi-agent workflow	Per agent	Shared	Per agent	Per agent	Per agent	Shared facts, private experience

Three decision criteria drive the table:

1. Does the task span turns? If every request is independent, short-term memory is a liability: stale context leaks into unrelated answers. Keep it small or drop it.

2. Can you label outcomes? Episodic and procedural memory are useless without a success signal. If you cannot tell whether an interaction succeeded (a resolved ticket, a merged PR, an explicit thumbs-up), the agent will retrieve "similar" past episodes with no idea which ones to imitate. Decide how outcomes are captured before writing the memory code.

3. Is the knowledge shared or private? Product documentation is shared and belongs in one semantic index. User preferences and past interactions are private and must be namespaced by user_id at the storage layer, not filtered in application code after retrieval. The same isolation logic applies to multi-tenant RAG deployments.

For a deeper treatment of the short-term versus long-term split specifically, see our guide on short-term vs long-term agent memory.

Agent Memory Failure Modes

Memory systems fail in ways that look like model problems. These are the ones we debug most often.

Context Pollution From Over-Retrieval

Pulling five semantic results, three episodes, and a recommended procedure into every prompt sounds thorough. In practice it buries the user's actual question in the middle of the context, which is exactly where models attend least (Liu et al., "Lost in the Middle"). Cap memory injections at a token budget (a few hundred tokens is usually enough) and put the freshest user message last. Our context window management guide covers budget allocation in detail.

Summarization Drift

Each summarize-and-compress cycle in short-term memory is lossy. After five or six cycles a long session can lose the constraint the user stated in turn two. Mitigations: pin explicit user constraints into a structured facts field that is never summarized, and re-inject it verbatim; summarize with a prompt that asks for "decisions, constraints, and open questions" rather than a narrative recap.

Stale Semantic Memory

Consolidation writes "learned facts" into the semantic store. Six months later the fact is wrong and the agent still asserts it confidently. Every consolidated fact needs learned_at, a source_session_id, and a retrieval-time recency weight (the exponential decay above). For facts that change (pricing, policy), prefer re-indexing from the source of truth over storing what the model inferred.

Procedural Lock-In

Once a tool sequence has a 90% success rate, the agent is steered toward it every time, so the alternatives never get sampled and the success rate never gets challenged. Add a small exploration rate (skip the recommendation on 5-10% of runs) and decay old counts so a sequence that stopped working loses its ranking.

Checkpointer as Accidental Long-Term Memory

LangGraph's checkpointer persists the entire graph state per thread (persistence docs). Teams sometimes rely on it as long-term memory by reusing thread_id forever. That works until the thread state grows past the context window. Use checkpoints for resumability within a thread and a separate store (the LangGraph Store API or your own vector DB) for cross-thread memory, as described in the LangGraph memory concepts.

Frequently Asked Questions

What's the difference between agent memory and RAG?

RAG is semantic memory only - retrieving relevant documents from a knowledge base. Agent memory includes RAG plus episodic memory (past experiences), procedural memory (learned skills), and working memory (reasoning traces). RAG answers "what do I know?", while full memory systems answer "what have I experienced?" and "what have I learned works?"

How much memory should an agent have?

For production AI agents:

Short-term: 10-20 messages (4,000-8,000 tokens)
Semantic: Unlimited via vector database
Episodic: 500-1,000 most relevant episodes per user
Procedural: Top 100 learned patterns
Working: Minimal, task-specific only

Balance memory size against retrieval latency and cost.

Should I store all conversations in memory?

Store summaries, not raw conversations. Raw conversation storage grows unbounded and slows retrieval. Instead:

Maintain raw conversation in short-term memory
Extract key facts, decisions, and learnings
Store extracted insights in semantic/episodic memory
Archive raw conversations separately for compliance

How do I prevent outdated memories?

Implement memory decay:

Time-based: Weight recent memories higher in retrieval
Relevance-based: Prune low-relevance memories periodically
Explicit updates: Allow updating/deleting specific memories
Version control: Track when knowledge was added/updated

Mark memories with timestamps and last-accessed dates for pruning.

What's the best vector database for agent memory?

Choose based on scale:

Prototypes: Chroma, FAISS (local, simple)
Production <1M vectors: Pinecone, Weaviate (managed)
Production >1M vectors: Milvus, Qdrant (self-hosted, scalable)
Enterprise: Pgvector (if already using PostgreSQL)

All work with LangChain. See LangGraph framework comparison for integration patterns.

How do I implement memory privacy?

Implement per-user memory isolation:

python

# User-specific memory namespaces
await semantic_memory.add_knowledge(
    text=knowledge,
    metadata={"user_id": user_id, "private": True}
)

# Retrieve only user's memories
results = await semantic_memory.retrieve(
    query=query,
    filter={"user_id": user_id}
)

Never mix user memories without explicit consent. Implement GDPR-compliant deletion.

Should episodic memory store successes or failures?

Store both. Successes show what works; failures prevent repeated mistakes. Weight successes higher in retrieval but include relevant failures as warnings. Our pattern:

Success episodes: Recommend similar approaches
Failure episodes: Warn against repeating mistakes

How do I test agent memory systems?

Test at multiple levels:

Unit tests: Individual memory operations (add, retrieve, update)
Integration tests: Memory systems working together
Retrieval tests: Relevant memories retrieved for queries
Performance tests: Retrieval latency at scale
Consistency tests: Memories persist across sessions

See our guide on testing AI agents.

What's the latency impact of memory retrieval?

Typical latency:

Short-term: <1ms (in-memory)
Semantic: 50-200ms (vector search)
Episodic: 100-300ms (vector + DB query)
Procedural: 20-50ms (DB query)

Optimize with caching, parallel retrieval, and result limiting. Budget 200-400ms total for comprehensive memory retrieval.

How do I implement memory in multi-agent systems?

Implement shared semantic memory (common knowledge) plus agent-specific episodic/procedural memory (individual experiences):

python

# Shared knowledge
shared_memory = SemanticMemory("shared-knowledge")

# Agent-specific memory
agent_memory = {
    "agent_1": EpisodicMemory(db, namespace="agent_1"),
    "agent_2": EpisodicMemory(db, namespace="agent_2")
}

See multi-agent orchestration patterns.

Conclusion

Production AI agents require comprehensive memory systems spanning short-term conversation context, long-term knowledge, episodic experiences, procedural skills, and working memory. The five memory types working together create agents that learn, adapt, and improve over time.

Key implementation patterns:

Short-term: Message buffer with summarization for context management
Semantic: Vector database with hybrid retrieval for knowledge
Episodic: Experience log with similarity search for learning
Procedural: Pattern storage for optimization and personalization
Working: Scratch space for multi-step reasoning

Memory is not optional for production agents. While simple chatbots can work with short-term memory alone, sophisticated agents require all five types for human-like intelligence.

Start with semantic memory (RAG), add episodic memory for learning, layer procedural memory for optimization, and integrate with LangGraph's stateful architecture for production systems.

Ready to build agents with sophisticated memory? Contact our team or read about our AI agent development services.

Free consultation

Book a free consultation call on AI agent memory systems

30-minute call with the HinterBuild team. Discuss your project, architecture questions, or next steps — no obligation.

Book a meeting

Resources:

Keep reading

AI SystemsSeptember 7, 2026 · 12 min read

AI Agent Memory: Short-Term vs Long-Term Architecture

Build AI agent memory that works: token-budgeted short-term context, vector-backed long-term storage, and the hybrid architecture we ship in production.

Read post

AI SystemsSeptember 14, 2026 · 12 min read

Agentic RAG: Iterative Retrieval & Self-Refinement Guide

Agentic RAG explained — iterative retrieval, query refinement, and self-correction loops with production Python code, costs, and guardrails.

Read post

AI SystemsSeptember 14, 2026 · 10 min read

Reliable Tool Calling: Production AI Agent Error Handling &

Reliable Tool Calling guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable production implementation.

Read post

AI SystemsSeptember 14, 2026 · 12 min read

ReAct vs Plan-and-Execute: Agent Reasoning Patterns Compared

Learn react vs plan-and-execute through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.

Read post

Why Agents Need Memory

Agent Memory Types Compared

1. Short-Term Memory (Conversation Buffer)

2. Long-Term Semantic Memory (Knowledge Base)

3. Episodic Memory (Experience Log)

4. Procedural Memory (Learned Behaviors)

5. Working Memory (Scratch Space)

Short-Term Memory

LangGraph Short-Term Memory

Long-Term Semantic Memory

Semantic Memory with Metadata Filtering

Episodic Memory

Episodic Memory in Decision-Making

Procedural Memory

Working Memory

Memory Retrieval Strategies

Hybrid Search (Semantic + Keyword)

Recency-Weighted Retrieval

Memory Consolidation

Memory Storage Architecture

Memory in LangGraph

Performance Optimization

Caching

Batch Operations

Lazy Loading

Choosing an Agent Memory Architecture

Agent Memory Failure Modes

Context Pollution From Over-Retrieval

Summarization Drift

Stale Semantic Memory

Procedural Lock-In

Checkpointer as Accidental Long-Term Memory

Frequently Asked Questions

What's the difference between agent memory and RAG?

How much memory should an agent have?

Should I store all conversations in memory?

How do I prevent outdated memories?

What's the best vector database for agent memory?

How do I implement memory privacy?

Should episodic memory store successes or failures?

How do I test agent memory systems?

What's the latency impact of memory retrieval?

How do I implement memory in multi-agent systems?

Conclusion

Related articles

AI Agent Memory: Short-Term vs Long-Term Architecture

Agentic RAG: Iterative Retrieval & Self-Refinement Guide

Reliable Tool Calling: Production AI Agent Error Handling &

ReAct vs Plan-and-Execute: Agent Reasoning Patterns Compared