langgraph-checkpointing-and-persistence

LangGraph Memory and Persistence

LangGraph provides built-in persistence through checkpointers, enabling workflows to maintain state across interactions, support multiple conversation threads, and recover from failures.

Checkpointer Types

InMemorySaver - Development/Testing

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_anthropic import ChatAnthropic

# Create checkpointer for in-memory persistence
checkpointer = InMemorySaver()

# Define the graph
model = ChatAnthropic(model="claude-sonnet-4-5-20250929")

def chatbot(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

# Build and compile with checkpointer
builder = StateGraph(MessagesState)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)
graph = builder.compile(checkpointer=checkpointer)

# Use thread_id to maintain conversation context
config = {"configurable": {"thread_id": "user-123"}}

# Conversation turn 1
result = graph.invoke({"messages": [("user", "Hi, I'm Alice")]}, config)

# Conversation turn 2 - remembers context from turn 1
result = graph.invoke({"messages": [("user", "What's my name?")]}, config)
# Response: "Your name is Alice"

Pros: Zero setup, fast iteration Cons: Data lost on restart - use only for development

SqliteSaver - Local Persistence

# Requires: pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-sonnet-4-5-20250929")

def chatbot(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(MessagesState)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

# Use context manager for proper connection handling
with SqliteSaver.from_conn_string("checkpoints.sqlite") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "session-456"}}
    result = graph.invoke({"messages": [("user", "Hello!")]}, config)