Using InMemorySaver to Give LangGraph Agents Persistent Conversation Memory
The article explains LangGraph’s checkpoint system that lets agents retain dialogue context, detailing the InMemorySaver for development and PostgresSaver for production, how to use checkpointer.put/get, thread_id for session isolation, manual state manipulation, and time‑travel replay, with full Python examples.
Checkpoint: Core of State Persistence
LangGraph’s checkpoint mechanism records a snapshot after each agent step, allowing automatic restoration of context on the next turn.
Two core operations are used: checkpointer.put(config, checkpoint) – saves the snapshot. checkpointer.get(config) – retrieves it.
InMemorySaver: In‑memory Checkpoint for Development
In development environments the InMemorySaver class stores checkpoints in RAM; production uses PostgresSaver.
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
# ... define ConversationState, chat_node ...
builder = StateGraph(ConversationState)
builder.add_node("chat", chat_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)
memory = InMemorySaver()
graph = builder.compile(checkpointer=memory)thread_id: Isolating Separate Conversations
The configurable thread_id key distinguishes independent sessions, e.g., “user‑123” vs “user‑456”.
# Session A
config_a = {"configurable": {"thread_id": "user-123"}}
# Session B
config_b = {"configurable": {"thread_id": "user-456"}}
result_a = graph.invoke({"messages": ["你好"], "counter": 0}, config_a)
result_b = graph.invoke({"messages": ["你好"], "counter": 0}, config_b)Manual State Control with get_state / update_state
Developers can fetch the current snapshot via graph.get_state(config) and modify fields directly with graph.update_state(config, {...}).
snapshot = graph.get_state(config_a)
print(snapshot.values) # {'messages': ['你好', '回复: 你好'], 'counter': 1}
graph.update_state(config_a, {"counter": 100})Time‑Travel Replay
Checkpoints retain execution history, enabling replay of any point by iterating over memory.list(config) and re‑executing.
for checkpoint in memory.list(config_a):
print(f"ID: {checkpoint.id}")
print(f"Time: {checkpoint.metadata}")
print(f"Value: {checkpoint.values}")Practical Use Cases
Recovering from errors by restoring a checkpoint.
Debugging specific steps via replay.
Implementing undo operations.
Full Example: A Conversational Agent with Memory
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent
@tool
def multiply(a: int, b: int) -> int:
"""Multiply two numbers."""
return a * b
model = ChatAnthropic(model="claude-sonnet-4-6")
memory = InMemorySaver()
agent = create_react_agent(model, [multiply], checkpointer=memory)
config = {"configurable": {"thread_id": "conversation-1"}}
result1 = agent.invoke({"messages": [{"role": "user", "content": "3乘4等于多少?"}]}, config)
print(result1["messages"][-1].content) # "3乘4等于12"
result2 = agent.invoke({"messages": [{"role": "user", "content": "再加5呢?"}]}, config)
print(result2["messages"][-1].content)Production with PostgresSaver
from langgraph.checkpoint.postgres import PostgresSaver
saver = PostgresSaver.from_conn_string(
"postgresql://user:password@localhost:5432/langgraph"
)
saver.setup() # run once to create tables
graph = builder.compile(checkpointer=saver)Signed-in readers can open the original source through BestHub's protected redirect.
This article has been distilled and summarized from source material, then republished for learning and reference. If you believe it infringes your rights, please contactand we will review it promptly.
Tech Ocean
Focused on AI programming, sharing ready-to-use development efficiency solutions.
How this landed with the community
Was this worth your time?
0 Comments
Thoughtful readers leave field notes, pushback, and hard-won operational detail here.
