	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > LangGraph > LangGraph Persistence with Checkpointers

LangGraph Persistence with Checkpointers

Author: Venkata Sudhakar

By default a LangGraph agent has no memory between calls - each invocation starts fresh with an empty state. Checkpointers add persistence by saving the full state of the graph after every node execution. This enables two critical capabilities: multi-turn conversations where the agent remembers what was said in previous turns, and resumable execution where a long-running agent can be paused and resumed later without losing progress. Both use the same mechanism - a thread_id identifies the conversation, and the checkpointer stores and retrieves the state for that thread.

LangGraph provides two built-in checkpointers. MemorySaver stores state in a Python dictionary in memory - perfect for development and testing but lost when the process restarts. SqliteSaver (from langgraph-checkpoint-sqlite) stores state in a SQLite file - persists across restarts, ideal for local applications. For production deployments, use PostgresSaver (langgraph-checkpoint-postgres) to store state in PostgreSQL shared across multiple server instances. The checkpointer is passed at compile time and is transparent to the rest of the graph code.

The below example shows a simple chat agent with MemorySaver that remembers the full conversation history across multiple calls using a thread_id.

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, Annotated, Sequence
import operator

llm = ChatOpenAI(model="gpt-4o-mini", api_key="your-key")

class State(TypedDict):
    messages: Annotated[Sequence, operator.add]

def chat_node(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

graph = StateGraph(State)
graph.add_node("chat", chat_node)
graph.set_entry_point("chat")
graph.add_edge("chat", END)

# Compile with checkpointer - this is all that is needed for persistence
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# thread_id identifies the conversation - same ID = same memory
config = {"configurable": {"thread_id": "user-venkata-session-1"}}

# Turn 1
result = app.invoke(
    {"messages": [HumanMessage(content="My name is Venkata. I work on data migration.")]},
    config=config
)
print("Turn 1:", result["messages"][-1].content)

# Turn 2 - agent remembers turn 1 because same thread_id
result = app.invoke(
    {"messages": [HumanMessage(content="What is my name and what do I work on?")]},
    config=config
)
print("Turn 2:", result["messages"][-1].content)

It gives the following output,

Turn 1: Nice to meet you, Venkata! Data migration is a fascinating field.
How can I help you today?

Turn 2: Your name is Venkata, and you work on data migration.

# The agent remembered the information from Turn 1
# without it being passed explicitly in Turn 2
# The full message history is stored in the checkpointer keyed by thread_id

# SqliteSaver: persists across process restarts
from langgraph.checkpoint.sqlite import SqliteSaver

# State is saved to migration_agent.db SQLite file
with SqliteSaver.from_conn_string("migration_agent.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "session-abc123"}}

# First run
    app.invoke(
        {"messages": [HumanMessage(content="We have 50 tables to migrate.")]},
        config=config
    )

# Process restarts here - state is in SQLite file...

with SqliteSaver.from_conn_string("migration_agent.db") as checkpointer:
    app = graph.compile(checkpointer=checkpointer)

# Resume same thread - agent still has the previous message
    result = app.invoke(
        {"messages": [HumanMessage(content="How many tables did I say we have?")]},
        config=config  # same thread_id
    )
    print(result["messages"][-1].content)

# Inspect the saved state at any time
with SqliteSaver.from_conn_string("migration_agent.db") as checkpointer:
    app2 = graph.compile(checkpointer=checkpointer)
    state = app2.get_state(config)
    print("Messages in state:", len(state.values["messages"]))

It gives the following output after process restart,

You mentioned you have 50 tables to migrate.

Messages in state: 4
# 4 messages: 2 human + 2 AI, persisted across restarts in SQLite

Each thread_id is a completely independent conversation with its own state history. Use a unique thread_id per user session, per task, or per conversation. For production web applications, generate a UUID as the thread_id when a new chat session starts and store it in the user session. The checkpointer also enables the human-in-the-loop pattern covered in the LangGraph Human-in-the-Loop tutorial - interrupt_before works by saving the state at the interruption point so it can be resumed after human review.

Send your comments, suggestions or queries regarding this site to [email protected].