– Fix LangGraph Agent Memory: Redis Persistence & Coercion Errors.
Master AI agent persistence in LangGraph. Complete guide to setting up RedisSaver for thread state, RedisStore for long-term memory, and JsonPlusSerializer.
Table of Contents
The Problem: Your Agent Has No Memory
Short-Term Memory: The Current Chat
Long-Term Memory: The Permanent Storage
The Sync Problem: Making Both Work Together
MESSAGE_COERCION_FAILURE: The Error That Breaks Everything
The JSON+ Serializer: The Real Fix
Namespaces: Keeping Users Separate
Cross-Thread Memory: Remembering Across Sessions
The Complete Working Code
Production Notes: What the Guide Missed
FAQs
The Bottom Line
1. The Problem: Your Agent Has No Memory
You’re building an AI agent. You talk to it. It responds. You ask a follow-up question. It stares blankly like you’ve never met.
This is not a bug. This is how LangGraph works by default.
Every conversation is a fresh start. The agent doesn’t remember what you said 2 minutes ago, let alone yesterday. It’s like talking to someone with severe short-term memory loss—but worse, because they don’t even have a notebook to write things down.
Why This Happens ?
LangGraph’s default checkpointer (MemorySaver) stores everything in RAM. The moment your Python script stops, every conversation vanishes into thin air.
# This agent has NO persistent memory graph = builder.compile() # No checkpointer # After this runs, everything is forgotten forever result = graph.invoke({"messages": [("human", "Hi! My name is Priya.")]})
The Solution: Redis
Redis gives your agent a memory that survives restarts, crashes, and even server migrations. It’s like giving your agent a permanent notebook that never gets thrown away.
But here’s the thing most tutorials don’t tell you: Your agent actually needs TWO types of memory, and they live in completely different places in Redis.
2. Short-Term Memory: The Current Chat
What It Is
Short-term memory is everything that happened in the current conversation. It includes:
The user’s messages
The agent’s responses
Which tools were called
The current state of the graph
It only lasts for the duration of the session. Close the chat, and it’s gone.
How LangGraph Handles It ?
LangGraph uses checkpointers to save the graph state at each step. Think of it like a save point in a video game. If something crashes, you can pick up exactly where you left off.
The Code
Here’s how to set up Redis for short-term memory. I’ll explain every single line.
from langgraph.checkpoint.redis import RedisSaver from langgraph.graph import StateGraph, MessagesState # Step 1: Tell Redis where to connect # "redis://localhost:6379" means "the Redis server running on my computer" REDIS_URI = "redis://localhost:6379" # Step 2: Create the checkpointer # This object is responsible for saving and loading your agent's state checkpointer = RedisSaver.from_conn_string(REDIS_URI) # Step 3: One-time setup # This creates the necessary Redis structures checkpointer.setup() # Step 4: Build your graph with the checkpointer builder = StateGraph(MessagesState) # ... add your nodes and edges graph = builder.compile(checkpointer=checkpointer) # Step 5: Run with a thread_id # The thread_id is like a conversation ID # Without this, Redis doesn't know which conversation to save config = {"configurable": {"thread_id": "conversation_123"}} result = graph.invoke( {"messages": [("human", "Hi! My name is Priya.")]}, config )
The Most Important Line :
The thread_id is the most important thing in this entire setup.
# ❌ WRONG: No thread_id result = graph.invoke({"messages": [("human", "Hello")]}) # ✅ CORRECT: With thread_id config = {"configurable": {"thread_id": "session_abc"}} result = graph.invoke({"messages": [("human", "Hello")]}, config)
Why? Redis uses thread_id as the key to find your conversation. Without it, Redis doesn’t know where to save or retrieve the state.
3. Long-Term Memory: The Permanent Storage
What It Is
Long-term memory is everything your agent remembers across conversations. It includes:
User preferences (“Priya likes dark mode”)
Past experiences (“Priya visited Paris last year”)
General knowledge (“Priya is a Python developer”)
It lasts forever. Or until you delete it.
How Redis Handles It
Long-term memory lives in Redis Store, not RedisSaver.
RedisSaver = Your agent’s scratchpad for the current conversation
RedisStore = Your agent’s diary that it writes in every day
The Code:
from langgraph.store.redis import RedisStore # Create the store store = RedisStore.from_conn_string("redis://localhost:6379") # One-time setup store.setup()
Saving a Memory:
import uuid from datetime import datetime def save_memory(user_id, content, memory_type): """Save a memory about a user.""" memory_id = str(uuid.uuid4()) memory_key = f"memory:{user_id}:{memory_id}" # Generate embedding for semantic search embedding = vectorizer.embed(content) store.put(memory_key, { "content": content, "type": memory_type, # "preference", "experience", "fact" "user_id": user_id, "embedding": embedding, "timestamp": datetime.now().isoformat() }) return memory_id
Retrieving a Memory:
def recall_memories(query, user_id, limit=5): """Find relevant memories using semantic search.""" query_vector = vectorizer.embed(query) # Search the store with vector similarity results = store.search( query=query_vector, num_results=limit, vector_field_name="embedding", distance_threshold=0.1 ) # Filter by user user_memories = [r.value for r in results if r.value.get("user_id") == user_id] return user_memories
Types of Memories:
Episodic Memories – Personal experiences
"Priya had a bad experience with United Airlines" "Priya prefers window seats on flights"
Semantic Memories – General facts
"Priya is a software engineer" "Priya lives in Mumbai"
4. The Sync Problem: Making Both Work Together
Now here’s the tricky part. You have two different memory systems. How do they work together?
The Manual Approach (Beginners)
You control everything. After each conversation, you extract memories. Before each conversation, you inject them.
def chat_with_memory(user_input, user_id, thread_id): # Step 1: Get relevant memories memories = recall_memories(user_input, user_id) # Step 2: Inject into prompt if memories: context = "\n".join([m["content"] for m in memories]) augmented_input = f"Context:\n{context}\n\nUser: {user_input}" else: augmented_input = user_input # Step 3: Run the agent config = {"configurable": {"thread_id": thread_id}} result = graph.invoke( {"messages": [("human", augmented_input)]}, config ) # Step 4: Extract and store new memories # In production, this would use an LLM with structured output new_memories = extract_memories(result["messages"], user_id) for memory in new_memories: save_memory(user_id, memory["content"], memory["type"]) return result
The Agentic Approach (Advanced)
You give the agent tools to save and recall memories. The agent decides when to use them.
@tool def save_memory(content: str): """Save important information about the user.""" save_memory(user_id, content, "preference") return f"Memory saved: {content}" @tool def recall_memory(query: str): """Find relevant information about the user.""" memories = recall_memories(query, user_id) return "\n".join([m["content"] for m in memories]) # Bind tools to the agent tools = [save_memory, recall_memory] agent_with_tools = agent.bind_tools(tools)
The Extraction Node (Production)
In a true LangGraph architecture, memory extraction is handled by a dedicated node:
User Input → Agent → Response
↓
Extractor Node (runs in background)
↓
RedisStore (updates long-term memory)This node uses an LLM with structured output to identify and store important information automatically.
5. MESSAGE_COERCION_FAILURE: The Error That Breaks Everything
You’re going to run into this error. Everyone does. It’s the most common frustration for beginners.
The Error Message
ValueError: Message dict must contain 'role' and 'content' keys, got {
'lc': 1,
'type': 'constructor',
'id': ['langchain','schema','messages','HumanMessage'],
'kwargs': {'content': 'hey','type': 'human','id': '...'}
}What’s Actually Happening
Your Redis checkpointer is trying to restore a message, but it can’t figure out what kind of message it is.
Picture this: You asked Redis to store a message. It stored the wrapping paper instead of the gift inside.
Why This Happens ?
LangChain messages have a special format. When you store a HumanMessage object, it looks like this internally:
HumanMessage(content="Hello")
But when Redis stores it, it might look like this:
{ 'lc': 1, 'type': 'constructor', 'id': ['langchain','schema','messages','HumanMessage'], 'kwargs': {'content': 'Hello', ...} }
This is the LangChain JSON (“lc”) format. It’s how LangChain serializes objects. But Redis doesn’t know how to convert this back to a HumanMessage object.
6. The JSON+ Serializer: The Real Fix
What It Is
The JsonPlusSerializer is the translator between LangChain objects and Redis storage.
What It Actually Does ?
| Operation | With JSON+ Serializer | Without JSON+ Serializer |
|---|---|---|
| Saving | HumanMessage → JSON (works) | HumanMessage → “lc” JSON (breaks) |
| Loading | JSON → HumanMessage (works) | “lc” JSON → “lc” JSON (still breaks) |
The Code:
from langgraph.checkpoint.redis import RedisSaver from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer # Step 1: Create the serializer serde = JsonPlusSerializer() # Step 2: Use it with RedisSaver checkpointer = RedisSaver.from_conn_string( "redis://localhost:6379", serde=serde # DON'T FORGET THIS ) # Step 3: Setup and compile checkpointer.setup() graph = builder.compile(checkpointer=checkpointer)
The Three Rules
# ✅ GOOD: BaseMessage object initial_state = {"messages": [HumanMessage(content="Hello!")]} # ✅ GOOD: Dict with role/content initial_state = {"messages": [{"role": "user", "content": "Hello!"}]} # ❌ BAD: to_dict() format - GUARANTEES MESSAGE_COERCION_FAILURE initial_state = {"messages": [HumanMessage(content="Hello!").to_dict()]}
Security Note
For production, use the strict allowlist with JsonPlusSerializer:
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer from langgraph.checkpoint.serde import serde # This prevents arbitrary code execution vulnerabilities serde = JsonPlusSerializer( allowlist=[serde.TYPES_ALLOWLIST] # Only allow known types )
7. Namespaces: Keeping Your Users Separate
The Problem
You have User A and User B. Without proper isolation, User A might accidentally see User B’s memories.
This is a privacy disaster.
The Solution: Namespaces
Redis supports namespaces through key prefixes. You store User A’s data under memory:user_A:... and User B’s under memory:user_B:....
def get_user_namespace(user_id): """Get the namespace for a user.""" return f"memory:{user_id}" def save_user_memory(user_id, content): """Save a memory in the user's namespace.""" namespace = get_user_namespace(user_id) memory_id = str(uuid.uuid4()) store.put(f"{namespace}:{memory_id}", { "content": content, "user_id": user_id })
Metadata Filtering
Even better than namespaces, you can use metadata filters on vector queries:
def search_user_memories(query, user_id): """Search for memories, but only for this user.""" query_vector = vectorizer.embed(query) results = store.search( query=query_vector, num_results=5, vector_field_name="embedding", distance_threshold=0.1 ) # Filter by user_id in the results return [r.value for r in results if r.value.get("user_id") == user_id]
The Thread ID vs User ID Distinction
thread_id: Isolates conversations (short-term memory)
user_id: Isolates users (long-term memory)
You need both. Thread IDs prevent conversation mixing. User IDs prevent user mixing.
8. Cross-Thread Memory: Remembering Across Sessions
What It Is
Cross-thread memory is when your agent remembers something across different conversations.
Session 1: “I’m a Python developer.”
Session 2 (next day): “What language should I use?” → “Python”
That’s cross-thread memory in action.
How to Implement It
# Store in the store store.put(f"user:{user_id}:preferences:language", "Python") # Retrieve across sessions def get_user_language(user_id): result = store.get(f"user:{user_id}:preferences:language") return result.value if result else None
The Update Pattern
How do I update long-term profile data without corrupting the current thread’s state?
Keep them in completely different places:
# User profile lives in the store def update_user_profile(user_id, data): store.put(f"profile:{user_id}", data) # Thread state lives in the checkpointer def update_thread_state(thread_id, state): checkpointer.put(thread_id, state) # These don't interact. Updating one doesn't break the other.
9. The Complete Working Code
Here’s everything together. Copy, paste, and it works.
Setup:
# Start Redis Stack docker run -d --name redis-memory -p 6379:6379 -p 8001:8001 redis/redis-stack:latest # Install dependencies pip install langgraph langchain-openai redis redisvl sentence-transformers
The Complete Agent:
import os import uuid from datetime import datetime from typing import Annotated, TypedDict from langgraph.checkpoint.redis import RedisSaver from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.store.redis import RedisStore from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, BaseMessage # Setup os.environ["OPENAI_API_KEY"] = "your-key-here" REDIS_URI = "redis://localhost:6379" # 1. Create the checkpointer (short-term memory) serde = JsonPlusSerializer() checkpointer = RedisSaver.from_conn_string(REDIS_URI, serde=serde) checkpointer.setup() # 2. Create the store (long-term memory) store = RedisStore.from_conn_string(REDIS_URI) store.setup() # 3. Create the vectorizer for embeddings from sentence_transformers import SentenceTransformer model = SentenceTransformer('all-MiniLM-L6-v2') def get_embedding(text): return model.encode(text).tolist() # 4. Define the state class AgentState(TypedDict): messages: Annotated[list[BaseMessage], add_messages] # 5. Define the agent model = ChatOpenAI(model="gpt-4o-mini") def call_model(state: AgentState): response = model.invoke(state["messages"]) return {"messages": [response]} # 6. Build the graph builder = StateGraph(AgentState) builder.add_node("agent", call_model) builder.add_edge(START, "agent") builder.add_edge("agent", END) graph = builder.compile(checkpointer=checkpointer) # 7. Helper functions def save_memory(user_id, content, memory_type): memory_id = str(uuid.uuid4()) key = f"memory:{user_id}:{memory_id}" # Generate embedding for semantic search embedding = get_embedding(content) # Use put() for RedisStore store.put(key, { "content": content, "type": memory_type, "user_id": user_id, "embedding": embedding, "timestamp": datetime.now().isoformat() }) return memory_id def recall_memories(query, user_id, limit=5): # Generate embedding for the query query_vector = get_embedding(query) # Search the store results = store.search( query=query_vector, num_results=limit, vector_field_name="embedding", distance_threshold=0.1 ) # Filter by user and return content return [r.value["content"] for r in results if r.value.get("user_id") == user_id] def chat_with_memory(user_input, user_id, thread_id): # 1. Retrieve relevant long-term memories memories = recall_memories(user_input, user_id) # 2. Inject into prompt if any found if memories: memory_context = "\n".join([f"- {m}" for m in memories]) augmented_input = f"Context about the user:\n{memory_context}\n\nUser question: {user_input}" else: augmented_input = user_input # 3. Run the agent with short-term memory config = {"configurable": {"thread_id": thread_id}} result = graph.invoke( {"messages": [HumanMessage(content=augmented_input)]}, config ) # 4. Extract and store new memories using LLM # In production, use a dedicated Extractor Node # Here's a simplified version: extract_prompt = f""" Extract important facts from this conversation that should be remembered about the user. Return as a list of facts. Conversation: {result['messages']} """ extraction = model.invoke([HumanMessage(content=extract_prompt)]) # For demo, we'll use a simple approach # In production, use structured output for fact in extraction.content.split('\n'): if fact.strip(): save_memory(user_id, fact.strip(), "extracted") return result # 8. Run it if __name__ == "__main__": user_id = "priya_123" # The user_id is typically passed via config # In production, you'd do: config = {"configurable": {"user_id": user_id, "thread_id": thread_id}} # Then access in your nodes via state or config thread_id_1 = "session_456" print(chat_with_memory("Hi! My name is Priya. I'm a Python developer.", user_id, thread_id_1)) thread_id_2 = "session_789" print(chat_with_memory("What language should I use for my project?", user_id, thread_id_2))
10. Production Notes: What the Guide Missed
1. The Context Manager Pattern
For production, use the context manager to ensure proper cleanup:
with RedisSaver.from_conn_string(REDIS_URI, serde=serde) as checkpointer: checkpointer.setup() graph = builder.compile(checkpointer=checkpointer) # ... use the graph # Automatically cleans up when the block exits
2. Async Versions
For high-throughput production systems, use AsyncRedisSaver:
from langgraph.checkpoint.redis.aio import AsyncRedisSaver async with AsyncRedisSaver.from_conn_string(REDIS_URI, serde=serde) as checkpointer: await checkpointer.asetup() graph = builder.compile(checkpointer=checkpointer) # ... async operations
3. Redis Stack Requirement
Important: Vector search and JSON storage require Redis Stack, not plain Redis.
The Docker command from Section 1 uses redis/redis-stack, not redis. If you use plain Redis, these features won’t work.
4. The User ID Injection Pattern
In production, the user_id should be passed in the config, not hardcoded:
# Pass user_id in config config = { "configurable": { "thread_id": thread_id, "user_id": user_id # Custom metadata } } # Access in your nodes def call_model(state: AgentState, config: dict): user_id = config.get("configurable", {}).get("user_id") # ... use user_id
5. Strict Serialization
For production, use the strict allowlist:
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer from langgraph.checkpoint.serde import serde serde = JsonPlusSerializer( allowlist=[serde.TYPES_ALLOWLIST] # Prevents arbitrary code execution )
6. Memory Extraction with LLM
In production, use a dedicated extraction node with structured output:
from pydantic import BaseModel from typing import List class MemoryExtraction(BaseModel): memories: List[str] memory_types: List[str] # "preference", "experience", "fact" def extract_memories_node(state: AgentState): """Dedicated node for memory extraction.""" llm = ChatOpenAI(model="gpt-4o") structured_llm = llm.with_structured_output(MemoryExtraction) response = structured_llm.invoke([ SystemMessage("Extract important facts about the user from this conversation."), HumanMessage(str(state["messages"])) ]) for memory in response.memories: save_memory(user_id, memory.content, memory.memory_type) return state
11. FAQs
Why is my agent forgetting everything?
You’re either:
Not using a checkpointer at all
Not passing a
thread_idUsing
MemorySaver(RAM only) instead ofRedisSaver
What’s MESSAGE_COERCION_FAILURE?
Redis stored your messages in LangChain JSON (“lc”) format and can’t convert them back. Use JsonPlusSerializer.
Why does Redis restore messages as raw JSON instead of BaseMessage?
You forgot serde=JsonPlusSerializer(). Add it to your RedisSaver.from_conn_string() call.
How do I use short-term and long-term memory together?
Use RedisSaver for short-term and RedisStore for long-term. Keep them in separate namespaces.
How do I update profile data without corrupting thread state?
Store profile data in the store, not the checkpointer. They use different keyspaces.
What’s the best way to handle cross-thread memory?
Use RedisStore with metadata filters. Never store user data in the checkpointer.
How can an agent look up persistent global state?
Use RedisStore with semantic search. Generate an embedding for the query and find similar stored memories.
Why do I need Redis Stack instead of plain Redis?
Redis Stack adds vector search and JSON support. Plain Redis doesn’t have these features. The Docker command redis/redis-stack gives you everything you need.
12. The Bottom Line:
Your AI agent needs memory. Redis gives it memory.
Two types of memory:
Short-term (RedisSaver) – Remembers the current conversation
Long-term (RedisStore) – Remembers across conversations
One error to avoid:
MESSAGE_COERCION_FAILURE – Fixed with JsonPlusSerializer
Three rules to never break:
Always pass a
thread_idAlways use
JsonPlusSerializerKeep short-term and long-term memory in separate namespaces
The One-Line Takeaway :
Your agent forgets because you didn’t give it Redis. Fix that, and it’ll remember everything.
