How to resume LangGraph from checkpoint
Quick answer
To resume
LangGraph from a checkpoint, compile your StateGraph with a checkpointer instance like MemorySaver() or a custom saver, then invoke the compiled app with the saved state. This restores the graph's state and allows continuation from the last checkpoint.PREREQUISITES
Python 3.8+pip install langgraphBasic knowledge of Python typing and functions
Setup
Install the langgraph package via pip if you haven't already:
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
This example shows how to create a StateGraph, add nodes and edges, compile it with a MemorySaver checkpointer, save state, and then resume from that checkpoint.
from langgraph.graph import StateGraph, END
from langgraph.checkpointer import MemorySaver
from typing import TypedDict
# Define the state type
class State(TypedDict):
messages: list[str]
# Define a node function
def my_node(state: State) -> State:
new_messages = state["messages"] + ["response from my_node"]
return {"messages": new_messages}
# Create the graph and add nodes
graph = StateGraph(State)
graph.add_node("my_node", my_node)
graph.set_entry_point("my_node")
graph.add_edge("my_node", END)
# Create a checkpointer
checkpointer = MemorySaver()
# Compile the graph with the checkpointer
app = graph.compile(checkpointer=checkpointer)
# Initial state
initial_state = {"messages": ["Hello"]}
# Invoke the graph to run and save checkpoint
result = app.invoke(initial_state)
print("First run messages:", result["messages"])
# Save checkpoint key
checkpoint_key = checkpointer.save_checkpoint(result)
# Later: resume from checkpoint
restored_state = checkpointer.load_checkpoint(checkpoint_key)
print("Restored state messages:", restored_state["messages"])
# Continue running from restored state
result2 = app.invoke(restored_state)
print("Continued run messages:", result2["messages"]) output
First run messages: ['Hello', 'response from my_node'] Restored state messages: ['Hello', 'response from my_node'] Continued run messages: ['Hello', 'response from my_node', 'response from my_node']
Common variations
- You can implement custom checkpointers by subclassing
Checkpointerfor persistent storage (e.g., files or databases). - Use async invocation with
await app.invoke_async(state)if your nodes are async functions. - Multiple checkpoints can be managed by storing keys externally and loading the desired checkpoint to resume.
Troubleshooting
- If you see
KeyErrorwhen loading a checkpoint, verify the checkpoint key is correct and the checkpoint data exists. - If state is not restored properly, ensure your
StateTypedDict matches the saved state structure exactly. - For persistent checkpoints, confirm your custom checkpointer correctly serializes and deserializes state.
Key Takeaways
- Use
MemorySaveror a custom checkpointer to save and load LangGraph state checkpoints. - Compile your
StateGraphwith thecheckpointerparameter to enable checkpointing. - Invoke the compiled app with saved state to resume execution from the checkpoint.
- Ensure your state schema matches exactly to avoid deserialization errors.
- Async nodes require async invocation methods to resume properly.