How to add checkpointing to LangGraph
Quick answer
Add checkpointing to
LangGraph by using a checkpointer like MemorySaver when compiling the graph with graph.compile(checkpointer=MemorySaver()). This enables state persistence across invocations, allowing your AI agent to resume from saved states.PREREQUISITES
Python 3.8+pip install langgraphBasic knowledge of Python typing and functions
Setup
Install the langgraph package via pip and import necessary classes. Ensure you have Python 3.8 or newer.
pip install langgraph output
Collecting langgraph Downloading langgraph-0.1.0-py3-none-any.whl (10 kB) Installing collected packages: langgraph Successfully installed langgraph-0.1.0
Step by step
Create a StateGraph with a typed state, define nodes as functions, add edges, and compile the graph with a MemorySaver checkpointer to enable checkpointing. Invoke the compiled app multiple times to see state persistence.
from langgraph.graph import StateGraph, END
from langgraph.checkpointers import MemorySaver
from typing import TypedDict
class State(TypedDict):
messages: list[str]
# Define a node function that appends a response
def my_node(state: State) -> State:
new_messages = state["messages"] + ["response"]
return {"messages": new_messages}
# Create the graph and add nodes and edges
graph = StateGraph(State)
graph.add_node("my_node", my_node)
graph.set_entry_point("my_node")
graph.add_edge("my_node", END)
# Compile with MemorySaver for checkpointing
app = graph.compile(checkpointer=MemorySaver())
# Initial invocation
result1 = app.invoke({"messages": ["Hello"]})
print("First call messages:", result1["messages"])
# Subsequent invocation resumes from checkpoint
result2 = app.invoke({})
print("Second call messages:", result2["messages"]) output
First call messages: ['Hello', 'response'] Second call messages: ['Hello', 'response', 'response']
Common variations
You can use other checkpointers or implement custom ones by subclassing the Checkpointer interface. Async invocation is supported by calling await app.invoke_async() if your nodes are async functions. Checkpointing works with complex state types as long as they are serializable.
import asyncio
async def async_node(state: State) -> State:
new_messages = state["messages"] + ["async response"]
return {"messages": new_messages}
graph.add_node("async_node", async_node)
graph.add_edge("my_node", "async_node")
graph.add_edge("async_node", END)
app_async = graph.compile(checkpointer=MemorySaver())
async def run_async():
result = await app_async.invoke_async({"messages": ["Start"]})
print("Async call messages:", result["messages"])
asyncio.run(run_async()) output
Async call messages: ['Start', 'response', 'async response']
Troubleshooting
- If checkpointing does not persist state, ensure you pass
checkpointer=MemorySaver()when compiling the graph. - For custom state types, verify they are JSON serializable.
- If state resets unexpectedly, confirm your environment supports file writes or use a persistent checkpointer.
Key Takeaways
- Use
MemorySaveras a checkpointer to enable checkpointing inLangGraph. - Pass the checkpointer to
graph.compile()to persist and resume state. - Checkpointing supports both synchronous and asynchronous node functions.
- Ensure your state data is serializable for checkpointing to work correctly.