Agent Memory Architectures with LangGraph: Complete Guide
Building intelligent agents that can maintain context across interactions is one of the most challenging problems in modern AI development. LangGraph, an extension of LangChain designed for building stateful, multi-actor applications, provides powerful primitives for implementing sophisticated memory architectures. In this guide, we'll explore how to design and implement agent memory systems that scale from simple conversation buffers to complex, persistent knowledge stores.
What Is Agent Memory?
Agent memory refers to the mechanisms by which an AI agent stores, retrieves, and uses information across interactions. Unlike a stateless LLM call that processes each prompt in isolation, a memory-enabled agent can recall past conversations, remember user preferences, and build upon previous reasoning steps. Memory transforms an agent from a reactive tool into a context-aware assistant that learns and adapts over time.
In LangGraph specifically, memory is implemented through state management. The framework treats memory as a graph state that flows through nodes and edges, allowing developers to define exactly what information persists, how it updates, and when it gets retrieved.
Why Memory Architectures Matter
Without proper memory architecture, agents suffer from several critical limitations:
- Context loss: Each interaction starts from scratch, forcing users to repeat information constantly.
- Inconsistent behavior: Agents cannot maintain personality, preferences, or ongoing tasks across sessions.
- Scalability issues: Naive approaches like stuffing entire conversation histories into prompts quickly hit token limits.
- Poor personalization: Without remembering user-specific details, agents deliver generic responses.
- Limited reasoning: Complex multi-step tasks require agents to track intermediate results and decisions.
A well-designed memory architecture addresses all these challenges while balancing latency, cost, and accuracy. LangGraph's stateful graph model makes it particularly well-suited for this work because it provides first-class support for checkpointing, state persistence, and conditional routing based on stored information.
Understanding LangGraph's Memory Model
LangGraph manages memory through three primary mechanisms: graph state, checkpointers, and store backends. Understanding how these interact is essential before diving into implementation.
Graph State: The Foundation
Every LangGraph application is built around a state schema. This schema defines what data flows through your graph and what gets persisted between steps. The state is typically defined using Python's TypedDict or Pydantic models, and each node in the graph receives the current state and returns updates to it.
from typing import TypedDict, Annotated, List
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
user_id: str
preferences: dict
summary: str
turn_count: int
In this example, the messages field uses the add_messages reducer, which appends new messages rather than replacing the entire list. This is crucial for conversation memory because it lets each node contribute messages without needing to manage the full history manually.
Checkpointers: Persistence Layer
Checkpointers are LangGraph's mechanism for persisting graph state between executions. When you attach a checkpointer to your graph, the framework automatically saves state after each node execution. This enables thread-based memory, where each conversation thread maintains its own independent state.
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver
# In-memory checkpointing (development only)
memory_saver = MemorySaver()
# SQLite for local persistence
import sqlite3
conn = sqlite3.connect("agent_memory.db", check_same_thread=False)
sqlite_saver = SqliteSaver(conn)
# PostgreSQL for production
# postgres_saver = PostgresSaver.from_conn_string("postgresql://...")
The choice of checkpointer depends on your deployment context. MemorySaver is perfect for development and testing but loses all data on restart. SqliteSaver provides local persistence suitable for single-instance deployments. For production multi-instance setups, PostgresSaver or Redis-based checkpointers are the right choice.
Store: Cross-Thread Long-Term Memory
While checkpointers provide per-thread memory, LangGraph's BaseStore interface enables cross-thread, long-term memory. This is where you store information that should persist across different conversations, such as user profiles, learned facts, or shared knowledge bases.
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
store = InMemoryStore()
# Store data with a namespace and key
store.put(
namespace=("user_profiles", "user_123"),
key="preferences",
value={
"language": "python",
"expertise": "intermediate",
"preferred_style": "concise"
}
)
# Retrieve stored data
item = store.get(("user_profiles", "user_123"), "preferences")
print(item.value) # {'language': 'python', 'expertise': 'intermediate', ...}
# Search across a namespace
results = store.search(("user_profiles", "user_123"))
for item in results:
print(item.key, item.value)
Building a Memory-Enabled Agent
Let's build a complete agent that combines short-term conversation memory with long-term user preferences. This example demonstrates the full pattern you'll use in production applications.
Setting Up the State and Tools
from typing import TypedDict, Annotated, List, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
# Define the state schema
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
user_id: str
summary: str
turn_count: int
# Define a simple tool
@tool
def save_preference(key: str, value: str) -> str:
"""Save a user preference for future conversations."""
return f"Saved preference: {key}={value}"
@tool
def recall_preference(key: str) -> str:
"""Recall a previously saved user preference."""
return f"Preference for {key}: not found"
tools = [save_preference, recall_preference]
# Initialize the LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
llm_with_tools = llm.bind_tools(tools)
Implementing Conversation Summarization
One of the most effective memory strategies is conversation summarization. Instead of keeping every message in context, you periodically compress older messages into a summary. This keeps token usage manageable while preserving important context.
SUMMARIZE_THRESHOLD = 10
def should_summarize(state: AgentState) -> str:
"""Determine whether to summarize the conversation."""
if len(state["messages"]) > SUMMARIZE_THRESHOLD:
return "summarize"
return "continue"
def summarize_conversation(state: AgentState) -> dict:
"""Compress older messages into a summary."""
existing_summary = state.get("summary", "")
messages_to_summarize = state["messages"][:-4] # Keep last 4 messages
summary_prompt = f"""Review the conversation and create a concise summary.
Existing summary:
{existing_summary}
Conversation to incorporate:
{chr(10).join([f"{m.type}: {m.content}" for m in messages_to_summarize])}
Provide an updated summary that captures key facts, decisions, and context."""
response = llm.invoke([HumanMessage(content=summary_prompt)])
# Keep only the most recent messages plus the summary
return {
"summary": response.content,
"messages": [], # Don't add new messages, just update summary
}
def call_model(state: AgentState) -> dict:
"""Main agent node that processes user input."""
system_content = f"""You are a helpful assistant with memory capabilities.
Conversation summary so far:
{state.get('summary', 'No prior context.')}
Turn count: {state.get('turn_count', 0)}
User ID: {state.get('user_id', 'unknown')}"""
messages = [SystemMessage(content=system_content)] + state["messages"]
response = llm_with_tools.invoke(messages)
return {
"messages": [response],
"turn_count": state.get("turn_count", 0) + 1
}
Assembling the Graph
# Build the graph
workflow = StateGraph(AgentState)
# Add nodes
workflow.add_node("agent", call_model)
workflow.add_node("summarize", summarize_conversation)
# Set entry point
workflow.add_edge(START, "agent")
# Add conditional routing for summarization
workflow.add_conditional_edges(
"agent",
should_summarize,
{
"summarize": "summarize",
"continue": END,
}
)
workflow.add_edge("summarize", END)
# Compile with memory
memory = MemorySaver()
store = InMemoryStore()
app = workflow.compile(checkpointer=memory, store=store)
Running the Agent
import uuid
# Configure thread for this conversation
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}
# First interaction
response = app.invoke(
{
"messages": [HumanMessage(content="Hi, I'm Alice and I prefer Python.")],
"user_id": "alice_001",
"summary": "",
"turn_count": 0,
},
config=config
)
print("Agent:", response["messages"][-1].content)
# Second interaction - agent remembers context
response = app.invoke(
{"messages": [HumanMessage(content="What programming language do I prefer?")]},
config=config
)
print("Agent:", response["messages"][-1].content)
print("Turn count:", response["turn_count"])
print("Summary:", response.get("summary", ""))
Because we pass the same thread_id in the config, the checkpointer automatically loads the previous state. The agent remembers Alice's name and preference without needing them repeated.
Advanced Memory Patterns
Episodic Memory with Vector Search
Episodic memory stores specific past interactions that can be retrieved based on semantic similarity. This is particularly useful for agents that need to recall relevant past experiences when facing new situations.
from langchain_openai import OpenAIEmbeddings
import numpy as np
class EpisodicMemory:
def __init__(self, embeddings_model: str = "text-embedding-3-small"):
self.embeddings = OpenAIEmbeddings(model=embeddings_model)
self.episodes: List[dict] = []
def add_episode(self, content: str, metadata: dict = None):
"""Store a new episodic memory."""
embedding = self.embeddings.embed_query(content)
self.episodes.append({
"content": content,
"embedding": embedding,
"metadata": metadata or {}
})
def retrieve(self, query: str, k: int = 3) -> List[dict]:
"""Retrieve the k most relevant episodes."""
if not self.episodes:
return []
query_embedding = self.embeddings.embed_query(query)
# Compute cosine similarity
scores = []
for episode in self.episodes:
sim = np.dot(query_embedding, episode["embedding"])
scores.append((sim, episode))
scores.sort(key=lambda x: x[0], reverse=True)
return [ep for _, ep in scores[:k]]
# Integrate into agent state
class AgentStateWithEpisodic(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
user_id: str
summary: str
episodic_memory: EpisodicMemory
relevant_episodes: List[str]
def agent_with_recall(state: AgentStateWithEpisodic) -> dict:
"""Agent node that retrieves relevant past episodes."""
last_message = state["messages"][-1].content
# Retrieve relevant memories
episodes = state["episodic_memory"].retrieve(last_message, k=3)
episode_context = "\n".join([f"- {e['content']}" for e in episodes])
system_msg = f"""You are a helpful assistant with episodic memory.
Relevant past experiences:
{episode_context if episode_context else 'No relevant memories found.'}
Conversation summary:
{state.get('summary', '')}"""
messages = [SystemMessage(content=system_msg)] + state["messages"]
response = llm.invoke(messages)
# Store this interaction as a new episode
state["episodic_memory"].add_episode(
content=f"User: {last_message}\nAgent: {response.content}",
metadata={"turn": state.get("turn_count", 0)}
)
return {"messages": [response]}
Semantic Memory with the Store Backend
Semantic memory stores general knowledge and facts that persist across all conversations. LangGraph's store backend is ideal for this, especially when combined with vector search for semantic retrieval.
from langgraph.store.memory import InMemoryStore
from langchain_openai import OpenAIEmbeddings
# Create a store with vector indexing
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
store = InMemoryStore(
index={
"embed": embeddings,
"dims": 1536,
"fields": ["content"]
}
)
def store_user_fact(state: AgentState, store: BaseStore) -> dict:
"""Extract and store important facts about the user."""
user_id = state["user_id"]
last_message = state["messages"][-1].content
# Use LLM to extract facts
extraction_prompt = f"""Extract any important facts about the user from this message.
Return a JSON list of facts, or an empty list if none found.
Message: {last_message}"""
response = llm.invoke([HumanMessage(content=extraction_prompt)])
# Store each fact
import json
try:
facts = json.loads(response.content)
for fact in facts:
store.put(
namespace=("user_facts", user_id),
key=f"fact_{uuid.uuid4().hex[:8]}",
value={"content": fact, "timestamp": str(datetime.now())}
)
except json.JSONDecodeError:
pass
return {}
def retrieve_relevant_facts(state: AgentState, store: BaseStore) -> dict:
"""Retrieve facts relevant to the current query."""
user_id = state["user_id"]
query = state["messages"][-1].content
# Semantic search across stored facts
results = store.search(
namespace=("user_facts", user_id),
query=query,
limit=5
)
facts = [item.value["content"] for item in results]
return {"relevant_facts": facts}
Working Memory with Sliding Window
Working memory holds the immediate context the agent is actively using. A sliding window approach keeps only the most recent N messages in full detail while older context is compressed or discarded.
from collections import deque
class SlidingWindowMemory:
def __init__(self, window_size: int = 20):
self.window_size = window_size
self.messages = deque(maxlen=window_size)
self.evicted_summary = ""
def add(self, message: BaseMessage):
"""Add a message, evicting old ones if necessary."""
if len(self.messages) == self.window_size:
# Evict oldest message
evicted = self.messages[0]
self._update_summary(evicted)
self.messages.append(message)
def _update_summary(self, evicted_message: BaseMessage):
"""Incorporate evicted message into running summary."""
prompt = f"""Update the conversation summary with this evicted message.
Current summary: {self.evicted_summary}
Evicted message: {evicted_message.type}: {evicted_message.content}
Provide an updated summary."""
response = llm.invoke([HumanMessage(content=prompt)])
self.evicted_summary = response.content
def get_context(self) -> List[BaseMessage]:
"""Get the current working memory context."""
summary_msg = SystemMessage(
content=f"Previous context summary: {self.evicted_summary}"
)
return [summary_msg] + list(self.messages)
Putting It All Together: A Complete Memory Architecture
Now let's combine all the memory types into a single, cohesive agent architecture. This example shows how short-term, working, episodic, and semantic memory can work together.
from typing import TypedDict, Annotated, List, Optional
from datetime import datetime
import uuid
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.store.memory import InMemoryStore
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
class ComprehensiveAgentState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages]
user_id: str
thread_id: str
conversation_summary: str
turn_count: int
retrieved_facts: List[str]
retrieved_episodes: List[str]
# Initialize persistent storage
import sqlite3
conn = sqlite3.connect("agent.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
store = InMemoryStore(
index={
"embed": embeddings,
"dims": 1536,
"fields": ["content"]
}
)
def retrieve_long_term_memory(state: ComprehensiveAgentState) -> dict:
"""Retrieve relevant facts and episodes from long-term storage."""
user_id = state["user_id"]
query = state["messages"][-1].content if state["messages"] else ""
# Retrieve user facts
fact_results = store.search(
namespace=("facts", user_id),
query=query,
limit=5
)
facts = [item.value.get("content", "") for item in fact_results]
# Retrieve episodic memories
episode_results = store.search(
namespace=("episodes", user_id),
query=query,
limit=3
)
episodes = [item.value.get("content", "") for item in episode_results]
return {
"retrieved_facts": facts,
"retrieved_episodes": episodes
}
def call_agent(state: ComprehensiveAgentState) -> dict:
"""Main agent reasoning node."""
facts_str = "\n".join([f"- {f}" for f in state.get("retrieved_facts", [])])
episodes_str = "\n".join([f"- {e}" for e in state.get("retrieved_episodes", [])])
summary = state.get("conversation_summary", "No prior context.")
system_prompt = f"""You are an intelligent assistant with multiple memory systems.
## Known Facts About the User
{facts_str if facts_str else "No stored facts yet."}
## Relevant Past Experiences
{episodes_str if episodes_str else "No relevant past experiences."}
## Conversation Summary
{summary}
Use all available context to provide helpful, personalized responses.
When you learn important new facts about the user, mention them explicitly."""
messages = [SystemMessage(content=system_prompt)] + state["messages"]
response = llm.invoke(messages)
return {
"messages": [response],
"turn_count": state.get("turn_count", 0) + 1
}
def persist_memories(state: ComprehensiveAgentState) -> dict:
"""Store new information in long-term memory."""
user_id = state["user_id"]
messages = state["messages"]
if len(messages) < 2:
return {}
# Store the interaction as an episode
last_human = [m for m in messages if isinstance(m, HumanMessage)][-1]
last_ai = [m for m in messages if isinstance(m, AIMessage)][-1]
episode_content = f"User asked: {last_human.content}\nAssistant responded: {last_ai.content}"
store.put(
namespace=("episodes", user_id),
key=f"ep_{uuid.uuid4().hex[:12]}",
value={
"content": episode_content,
"timestamp": datetime.now().isoformat(),
"turn": state.get("turn_count", 0)
}
)
# Extract and store facts
extraction_prompt = f"""Extract any new facts about the user from this exchange.
Return a JSON array of fact strings. Return [] if no new facts.
User: {last_human.content}
Assistant: {last_ai.content}"""
extraction_response = llm.invoke([HumanMessage(content=extraction_prompt)])
import json
try:
facts = json.loads(extraction_response.content)
for fact in facts:
store.put(
namespace=("facts", user_id),
key=f"fact_{uuid.uuid4().hex[:8]}",
value={
"content": fact,
"timestamp": datetime.now().isoformat()
}
)
except (json.JSONDecodeError, TypeError):
pass
return {}
def maybe_summarize(state: ComprehensiveAgentState) -> str:
"""Route to summarization if conversation is getting long."""
if len(state["messages"]) > 15:
return "summarize"
return "end"
def summarize(state: ComprehensiveAgentState) -> dict:
"""Compress conversation history into a summary."""
existing = state.get("conversation_summary", "")
old_messages = state["messages"][:-6]
prompt = f"""Update the conversation summary.
Current summary: {existing}
New messages to incorporate:
{chr(10).join([f"{m.type}: {m.content}" for m in old_messages])}
Provide a concise updated summary preserving key information."""
response = llm.invoke([HumanMessage(content=prompt)])
return {"conversation_summary": response.content}
# Build the complete graph
workflow = StateGraph(ComprehensiveAgentState)
workflow.add_node("retrieve", retrieve_long_term_memory)
workflow.add_node("agent", call_agent)
workflow.add_node("persist", persist_memories)
workflow.add_node("summarize", summarize)
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "agent")
workflow.add_edge("agent", "persist")
workflow.add_conditional_edges(
"persist",
maybe_summarize,
{"summarize": "summarize", "end": END}
)
workflow.add_edge("summarize", END)
app = workflow.compile(checkpointer=checkpointer, store=store)
# Usage example
config = {"configurable": {"thread_id": "conversation_001"}}
result = app.invoke(
{
"messages": [HumanMessage(content="I'm a data scientist working on NLP projects.")],
"user_id": "user_42",
"thread_id": "conversation_001",
"conversation_summary": "",
"turn_count": 0,
"retrieved_facts": [],
"retrieved_episodes": []
},
config=config
)
print("Response:", result["messages"][-1].content)
# Continue the conversation
result = app.invoke(
{"messages": [HumanMessage(content="Can you recommend a good transformer library?")]},
config=config
)
print("Response:", result["messages"][-1].content)
Best Practices for Agent Memory
Choose the Right Memory Type for Each Use Case
Not every agent needs all memory types. A simple customer service bot may only need short-term conversation memory with summarization. A personal assistant benefits from long-term semantic memory for user preferences. A research agent needs episodic memory to recall past findings. Start simple and add complexity only when you observe clear benefits.
Manage Token Budgets Explicitly
Memory directly impacts token usage, which affects both cost and latency. Always track how many tokens your memory system consumes and set hard limits. A common pattern is to allocate a fixed token budget for context, then distribute it across summary, retrieved facts, and recent messages.
def build_context_within_budget(
summary: str,
recent_messages: List[BaseMessage],
retrieved_facts: List[str],
max_tokens: int = 4000
) -> List[BaseMessage]:
"""Build context while respecting a token budget."""
from langchain_core.messages import SystemMessage
token_estimate = lambda text: len(text) // 4 # Rough estimate
budget = max_tokens
context_parts = []
# Always include summary first
if summary:
summary_msg = f"Conversation summary: {summary}"
budget -= token_estimate(summary_msg)
context_parts.append(SystemMessage(content=summary_msg))
# Add retrieved facts if budget allows
if retrieved_facts and budget > 200:
facts_text = "Known facts:\n" + "\n".join(f"- {f}" for f in retrieved_facts)
facts_tokens = token_estimate(facts_text)
if facts_tokens < budget - 500: # Reserve space for recent messages
budget -= facts_tokens
context_parts.append(SystemMessage(content=facts_text))
# Add recent messages within remaining budget
for msg in reversed(recent_messages):
msg_tokens = token_estimate(str(msg.content))
if msg_tokens > budget:
break
budget -= msg_tokens
context_parts.insert(-1, msg) # Insert before system messages
return context_parts
Implement Memory Decay
Not all memories are equally valuable over time. Implement decay mechanisms that gradually reduce the relevance of old information. This prevents stale facts from polluting your agent's context and keeps retrieval focused on current relevance.
from datetime import datetime, timedelta
def get_memories_with_decay(
store: InMemoryStore,
namespace: tuple,
query: str,
limit: int = 5,
half_life_days: float = 30.0
) -> List[dict]:
"""Retrieve memories with time-based decay scoring."""
results = store.search(namespace=namespace, query=query, limit=limit * 2)
now = datetime.now()
scored_results = []
for item in results:
# Parse timestamp
ts_str = item.value.get("timestamp", "")
try:
ts = datetime.fromisoformat(ts_str)
age_days = (now - ts).total_seconds() / 86400
except (ValueError, TypeError):
age_days = 0
# Apply exponential decay
decay_factor = 0.5 ** (age_days / half_life_days)
scored_results.append({
"content": item.value.get("content", ""),
"decay_score": decay_factor,
"age_days": age_days
})
# Sort by decay-adjusted relevance
scored_results.sort(key=lambda x: x["decay_score"], reverse=True)
return scored_results[:limit]
Handle Memory Conflicts
Users may provide contradictory information over time. Your memory system should handle updates gracefully, either by overwriting old facts or by maintaining a versioned history. A simple approach is to use the store's update capability with conflict detection.
def update_user_fact(
store: InMemoryStore,
user_id: str,
fact_key: str,
new_value: str,
confidence: float = 1.0
) -> dict:
"""Update a user fact with conflict handling."""
namespace = ("facts", user_id)
# Check for existing fact
existing = store.get(namespace, fact_key)
if existing:
old_value = existing.value.get("content", "")
if old_value == new_value:
return {"status": "unchanged", "value": old_value}
# Store the old value in history
history_key = f"{fact_key}_history_{uuid.uuid4().hex[:6]}"
store.put(namespace, history_key, {
"content": old_value,
"superseded_by": new_value,
"timestamp": datetime.now().isoformat()
})
# Update with new value
store.put(namespace, fact_key, {
"content": new_value,
"confidence": confidence,
"updated_at": datetime.now().isoformat(),
"previous_value": old_value
})
return {"status": "updated", "old": old_value, "new": new_value}
else:
store.put(namespace, fact_key, {
"content": new_value,
"confidence": confidence,
"created_at": datetime.now().isoformat()
})
return {"status": "created", "value": new_value}
Test Memory Behavior Explicitly
Memory bugs are subtle and often only manifest across multiple interactions. Write integration tests that simulate multi-turn conversations and verify that the agent correctly recalls and uses stored information.
def test_memory_persistence():
"""Test that facts persist across conversation threads."""
store = InMemoryStore()
user_id = "test_user"
# Thread 1: User shares information
config1 = {"configurable": {"thread_id": "thread_1"}}
app.invoke(
{
"messages": [HumanMessage(content="I live in Tokyo.")],
"user_id": user_id,
"thread_id": "thread_1",
"conversation_summary": "",
"turn_count": 0,
"retrieved_facts": [],
"retrieved_episodes": []
},
config=config1
)
# Thread 2: Different conversation, same user
config2 = {"configurable": {"thread_id": "thread_2"}}
result = app.invoke(
{
"messages": [HumanMessage(content="What city do I live in?")],
"user_id": user_id,
"thread_id": "thread_2",
"conversation_summary": "",
"turn_count": 0,
"retrieved_facts": [],
"retrieved_episodes": []
},
config=config2
)
response = result["messages"][-1].content.lower()
assert "tokyo" in response, f"Agent should remember user lives in Tokyo. Got: {response}"
print("Memory persistence test passed!")
Monitor and Debug Memory Usage
In production, instrument your memory system with logging and metrics. Track how often memories are retrieved, what fraction is actually useful, and how token costs grow over time. LangGraph's built-in tracing can help visualize how state flows through your graph.
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agent_memory")
def logged_retrieve(state: ComprehensiveAgentState) -> dict:
"""Memory retrieval with logging."""
user_id = state["user_id"]
query = state["messages"][-1].content if state["messages"] else ""
logger.info(f"Retrieving memories for user={user_id}, query_length={len(query)}")
facts = store.search(("facts", user_id), query=query, limit=5)
episodes = store.search(("episodes", user_id), query=query, limit=3)
logger.info(f"Retrieved {len(facts)} facts and {len(episodes)} episodes")
# Log retrieval quality metrics
if facts:
avg_score = sum(getattr(f, 'score', 0) for f in facts) / len(facts)
logger.info(f"Average fact relevance score: {avg_score:.3f}")
return {
"retrieved_facts": [f.value.get("content", "") for f in facts],
"retrieved_episodes": [e.value.get("content", "") for e in episodes]
}
Production Considerations
Choosing a Checkpointer Backend
For production deployments, your checkpointer choice has significant implications. SQLite works for single-instance applications but cannot handle concurrent writes from multiple processes. PostgreSQL is the recommended production backend because it handles concurrent access, provides ACID guarantees, and scales horizontally. Redis-based checkpointers offer lower latency but with different consistency trade-offs.
Security and Privacy
Memory systems store potentially sensitive user information. Always encrypt stored data at rest, implement proper access controls keyed to user identity, and provide mechanisms for users to view and delete their stored memories. Consider implementing a retention policy that automatically purges old data unless explicitly preserved.
def purge_old_memories(
store: InMemoryStore,
user_id: str,
max_age_days: int = 90
) -> int:
"""Delete memories older than the retention period."""
namespace = ("facts", user_id)
cutoff = datetime.now() - timedelta(days=max_age_days)
all_items = store.search(namespace, limit=10000)
purged = 0
for item in all_items:
ts_str = item.value.get("timestamp", item.value.get("created_at", ""))
try:
ts = datetime.fromisoformat(ts_str)
if ts < cutoff:
store.delete(namespace, item.key)
purged += 1
except (ValueError, TypeError):
continue
logger.info(f"Purged {purged} old memories for user {user_id}")
return purged
Performance Optimization
Memory retrieval can become a bottleneck as your memory store grows. Use vector indexes for semantic search, implement caching for frequently accessed memories, and consider pre-computing embeddings for stored content. Batch embedding operations when storing multiple facts to reduce API calls.
from functools import lru_cache
class CachedMemoryStore:
"""Wrapper around the store with caching for frequent retrievals."""
def __init__(self, store: InMemoryStore):
self.store = store
self._cache = {}
def search(self, namespace: tuple, query: str, limit: int = 5):
cache_key = (namespace, query[:100], limit)
if cache_key in self._cache:
return self._cache[cache_key]
results = self.store.search(namespace, query=query, limit=limit)
self._cache[cache_key] = results
# Simple cache eviction
if len(self._cache) > 1000:
self._cache.clear()
return results
def put(self, namespace: tuple, key: str, value: dict):
# Invalidate cache on writes
keys_to_remove = [k for k in self._cache if k[0] == namespace]
for k in keys_to_remove:
del self._cache[k]
self.store.put(namespace, key, value)
Conclusion
Agent memory is the bridge between stateless LLM calls and truly intelligent, context-aware assistants. LangGraph provides a robust foundation for building memory architectures through its state management, checkpointing, and store primitives. By combining short-term conversation memory with summarization, long-term semantic storage for user facts, and episodic memory for past experiences, you can create agents that genuinely learn and adapt over time. The key is to start with the simplest memory system that meets your needs, instrument it thoroughly, and iteratively add sophistication as your application demands. Remember that more memory is not always better — every byte of context costs tokens, increases latency, and introduces potential for confusion. The best memory architectures are thoughtful about what to remember, what to forget, and when to retrieve, creating agents that feel both knowledgeable and responsive without being overwhelmed by their own history.