TopolearnPersistence powers memory, human-in-the-loop, time travel and crash recovery; nearly every real app needs it.
A checkpointer saves a snapshot of the graph's state (a checkpoint) after every step. Attach one when compiling:
graph = builder.compile(checkpointer=InMemorySaver())Checkpoints are grouped into threads. A thread is one conversation or one job, named by a thread_id you choose and pass in the config:
config = {"configurable": {"thread_id": "user-42-chat-1"}}
graph.invoke({"messages": [...]}, config)Call again with the same thread_id and the graph starts from the saved state, so a chatbot remembers earlier messages; you send only the new message. A different thread_id starts fresh. This is short-term memory: scoped to one thread.
Inspect a thread with graph.get_state(config). It returns a snapshot with .values (the state), .next (nodes that would run next; empty when finished) and more.
Which checkpointer:
InMemorySaverfor tests and experiments (lost when the process ends),SqliteSaver(packagelanggraph-checkpoint-sqlite) for local apps,PostgresSaver(langgraph-checkpoint-postgres) and others such as MongoDB or Redis for production. Call.setup()once to create tables.
When you deploy with LangSmith Deployment, persistence is provided for you and you don't pass a checkpointer.
State is serialized when saved, so keep it to data types the serializer understands (plain data, LangChain messages, Pydantic models).
Example
from langchain_core.messages import AIMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END, MessagesState
def bot(state: MessagesState) -> dict:
return {"messages": [AIMessage(f"I have seen {len(state['messages'])} messages")]}
b = StateGraph(MessagesState)
b.add_node(bot)
b.add_edge(START, "bot")
b.add_edge("bot", END)
graph = b.compile(checkpointer=InMemorySaver())
chat1 = {"configurable": {"thread_id": "chat-1"}}
graph.invoke({"messages": [{"role": "user", "content": "hi"}]}, chat1)
out = graph.invoke({"messages": [{"role": "user", "content": "again"}]}, chat1)
print(out["messages"][-1].content) # I have seen 3 messages
chat2 = {"configurable": {"thread_id": "chat-2"}}
print(graph.invoke({"messages": [{"role": "user", "content": "hi"}]}, chat2)["messages"][-1].content)
# I have seen 1 messages
print(graph.get_state(chat1).next) # () -> finishedRemember
- compile(checkpointer=...) saves state after every step
- thread_id in config['configurable'] selects the conversation
- Same thread continues; new thread starts fresh
- InMemorySaver for tests; Postgres/SQLite/etc. for real apps
Watch out for
- Forgetting thread_id when a checkpointer is set; LangGraph raises an error asking for it.