Durable, Restart-Safe Agents

The agents from the previous Chapter “Building a ReAct Agent with LangGraph + Ollama” have one important limitation: they forget everything the moment the Python process exits. If you build a chatbot on top of create_react_agent, every user turn starts from scratch, with no memory of the previous message, let alone yesterday’s conversation. That is a script, not a service.

LangGraph fixes this with one concept: the checkpointer. You pass a checkpointer to .compile(), you add a thread_id through your invoke config, and now every step of your graph (every state update from every node) gets serialized to whatever storage the checkpointer manages. Multi-turn conversations remember previous turns. Interrupted work resumes where it stopped. Crashed processes come back up mid-transaction with no lost state.

This chapter builds four small scripts that demonstrate this progressively: in-process memory, on-disk SQLite persistence, cross-process restart, and inspecting the recorded checkpoint history. All four share one uncompiled state graph; the checkpointer is the only thing that varies.

Three flavors of checkpointer

LangGraph ships several checkpointer implementations. Only the first one is included in the core package; the others are separate PyPI packages.

MemorySaver: from langgraph.checkpoint.memory, included in the core langgraph install. Stores checkpoints in a Python dict. Fast, zero configuration, forgotten on process exit. Use in tests, notebooks, and request-scoped web endpoints where conversation state only needs to live as long as the request.
SqliteSaver: from langgraph.checkpoint.sqlite, in the separate langgraph-checkpoint-sqlite package. Persists checkpoints to a single SQLite file. Zero infrastructure: no server, no daemon, no config file. Perfect for personal projects, small tools, and single-node services where a SQLite file is enough storage.
PostgresSaver: from langgraph.checkpoint.postgres, in the separate langgraph-checkpoint-postgres package. For production services with multiple workers or high concurrency. You bring your own Postgres.

All three implement the same BaseCheckpointSaver interface, so swapping between them is a one-line change. Nothing in the graph itself changes.

The shared graph

All four scripts in source-code/langgraph_durable/ use the same uncompiled state graph that is defined in the file _graph.py:

 1 from typing import Annotated, TypedDict
 2 
 3 from langchain_core.messages import BaseMessage, SystemMessage
 4 from langchain_ollama import ChatOllama
 5 from langgraph.graph import END, START, StateGraph
 6 from langgraph.graph.message import add_messages
 7 
 8 SYSTEM_PROMPT = SystemMessage(
 9     content=(
10         "You are a helpful assistant. You keep answers short. "
11         "You freely refer back to earlier turns in the conversation."
12     )
13 )
14 
15 
16 class State(TypedDict):
17     messages: Annotated[list[BaseMessage], add_messages]
18 
19 
20 _model = ChatOllama(model="qwen3.5:4b", temperature=0)
21 
22 
23 def call_model(state: State) -> dict:
24     messages = state["messages"]
25     if not messages or not isinstance(messages[0], SystemMessage):
26         messages = [SYSTEM_PROMPT] + list(messages)
27     return {"messages": [_model.invoke(messages)]}
28 
29 
30 def build_graph() -> StateGraph:
31     graph = StateGraph(State)
32     graph.add_node("model", call_model)
33     graph.add_edge(START, "model")
34     graph.add_edge("model", END)
35     return graph

This is an example of the smallest useful graph: MessagesState-style transcript, one node that calls the model on it, using no tools. The tool-calling ReAct loop from the last Chapter “Building a ReAct Agent with LangGraph + Ollama” works exactly the same way once you add a checkpointer, but starting simple lets us focus on what the checkpointer does.

The call to build_graph() returns an uncompiled StateGraph. Each script compiles it with a different checkpointer.

Example 1: `MemorySaver` and `thread_id`

Here is the file example file 01_memory_saver.py:

 1 from langchain_core.messages import HumanMessage
 2 from langgraph.checkpoint.memory import MemorySaver
 3 
 4 from _graph import build_graph
 5 
 6 checkpointer = MemorySaver()
 7 agent = build_graph().compile(checkpointer=checkpointer)
 8 
 9 config = {"configurable": {"thread_id": "demo"}}
10 
11 turns = [
12     "My name is Mark. What is 12 * 12?",
13     "What did I just ask you to compute?",
14     "What is my name?",
15 ]
16 
17 for user_turn in turns:
18     print(f"USER: {user_turn}")
19     result = agent.invoke({"messages": [HumanMessage(content=user_turn)]}, config=config)
20     reply = result["messages"][-1]
21     print(f"AGENT: {reply.content.strip()}\n")

Two things are new relative to the “LangGraph 1.0 Fundamentals” chapter’s example code 04_llm_in_a_node.py.

The checkpointer= argument to .compile(). Without a checkpointer, each .invoke() call starts from a fresh empty state. With one, .invoke() looks up the state for the configured thread, appends the new input, runs the graph, and saves the resulting state before returning.

The thread_id in the invoke config. thread_id is the conversation identifier. Two invocations with the same thread_id share history; two invocations with different thread_ids do not. In a chat application this is typically the user’s session ID or conversation ID. In a background job it might be a task ID. In a personal script like this one it can be any string.

Expected output:

1 $ uv run 01_memory_saver.py
2 USER: My name is Mark. What is 12 * 12?
3 AGENT: 12 * 12 is 144, Mark.
4 
5 USER: What did I just ask you to compute?
6 AGENT: You asked me to compute 12 * 12.
7 
8 USER: What is my name?
9 AGENT: Your name is Mark.

Three invocations, three growing transcripts, one thread. On the third invocation the model is given all six prior messages plus the new question, so of course it can answer. That is the whole trick: the system works at a cost of growing context size.

Example 2: `SqliteSaver`, spread across two processes

Now we replace the checkpointer and watch the same mechanism survive a process restart. Here is the next example file 02_sqlite_first_run.py:

 1 from langchain_core.messages import HumanMessage
 2 from langgraph.checkpoint.sqlite import SqliteSaver
 3 
 4 from _graph import build_graph
 5 
 6 CHECKPOINT_DB = "checkpoints.db"
 7 THREAD_ID = "sqlite-demo"
 8 
 9 with SqliteSaver.from_conn_string(CHECKPOINT_DB) as checkpointer:
10     agent = build_graph().compile(checkpointer=checkpointer)
11 
12     config = {"configurable": {"thread_id": THREAD_ID}}
13 
14     user_turn = "My name is Mark and I live in Sedona, Arizona. Please remember that."
15     print(f"USER: {user_turn}")
16 
17     result = agent.invoke({"messages": [HumanMessage(content=user_turn)]}, config=config)
18 
19     print(f"AGENT: {result['messages'][-1].content.strip()}")
20     print(f"\nSaved to {CHECKPOINT_DB!r} on thread {THREAD_ID!r}.")
21     print("Now run 03_sqlite_second_run.py in a fresh process.")

The object SqliteSaver.from_conn_string(path) is a context manager that opens the SQLite file, creates the checkpoint tables on first use, and closes cleanly on exit. Everything else (thread config, .compile(), .invoke()) is identical to the MemorySaver version.

Let’s run this example:

1 $ uv run 02_sqlite_first_run.py
2 USER: My name is Mark and I live in Sedona, Arizona. Please remember that.
3 AGENT: Got it, Mark - noted that you live in Sedona, Arizona.
4 
5 Saved to 'checkpoints.db' on thread 'sqlite-demo'.
6 Now run 03_sqlite_second_run.py in a fresh process.

The script exits and the Python runtime environment is gone so now all we have is checkpoints.db on disk.

The next example 03_sqlite_second_run.py is nearly identical to example 2, except it asks a question that requires remembering and uses the database created by the last example:

 1 from langchain_core.messages import HumanMessage
 2 from langgraph.checkpoint.sqlite import SqliteSaver
 3 
 4 from _graph import build_graph
 5 
 6 CHECKPOINT_DB = "checkpoints.db"
 7 THREAD_ID = "sqlite-demo"
 8 
 9 with SqliteSaver.from_conn_string(CHECKPOINT_DB) as checkpointer:
10     agent = build_graph().compile(checkpointer=checkpointer)
11 
12     config = {"configurable": {"thread_id": THREAD_ID}}
13 
14     user_turn = "What is my name and where do I live?"
15     print(f"USER: {user_turn}")
16 
17     result = agent.invoke({"messages": [HumanMessage(content=user_turn)]}, config=config)
18 
19     print(f"AGENT: {result['messages'][-1].content.strip()}")
20     print(f"\nFull transcript for thread {THREAD_ID!r}:")
21     for m in result["messages"]:
22         print(f"  {type(m).__name__}: {m.content[:120]}")

Run example 3 after example 2 has finished running:

1 $ uv run 03_sqlite_second_run.py
2 USER: What is my name and where do I live?
3 AGENT: Your name is Mark, and you live in Sedona, Arizona.
4 
5 Full transcript for thread 'sqlite-demo':
6   HumanMessage: My name is Mark and I live in Sedona, Arizona. Please remember that.
7   AIMessage: Got it, Mark - noted that you live in Sedona, Arizona.
8   HumanMessage: What is my name and where do I live?
9   AIMessage: Your name is Mark, and you live in Sedona, Arizona.

The full transcript from both processes is intact. No code has been written to serialize messages, load them, or thread them into the prompt. The checkpointer did all of it.

Delete checkpoints.db to reset and start fresh.

Example 3: inspecting checkpoint history

Every state update from every node in every invocation is a checkpoint. The graph exposes two methods to inspect them:

agent.get_state(config) returns the current state as a StateSnapshot.
agent.get_state_history(config) yields every checkpoint recorded for the thread, newest first.

Here is our fourth example 04_state_history.py:

 1 from langgraph.checkpoint.sqlite import SqliteSaver
 2 
 3 from _graph import build_graph
 4 
 5 CHECKPOINT_DB = "checkpoints.db"
 6 THREAD_ID = "sqlite-demo"
 7 
 8 with SqliteSaver.from_conn_string(CHECKPOINT_DB) as checkpointer:
 9     agent = build_graph().compile(checkpointer=checkpointer)
10 
11     config = {"configurable": {"thread_id": THREAD_ID}}
12 
13     print(f"=== Current state for thread {THREAD_ID!r} ===")
14     current = agent.get_state(config)
15     for m in current.values["messages"]:
16         print(f"  {type(m).__name__}: {m.content[:100]}")
17 
18     print(f"\n=== Checkpoint history (newest first) ===")
19     for i, snapshot in enumerate(agent.get_state_history(config)):
20         n_messages = len(snapshot.values.get("messages", []))
21         meta = snapshot.metadata or {}
22         source = meta.get("source", "?")
23         step = meta.get("step", "?")
24         print(f"  [{i}] step={step} source={source} messages_len={n_messages}")

After running scripts 2 and 3, this shows the current transcript and then a summary of every checkpoint the graph has recorded for the thread. Each StateSnapshot carries a .config field you can pass back to .invoke() to time-travel: restart the graph from that historical state and run it forward with a different input. That is the mechanism the next chapter uses for human-in-the-loop editing: pause the graph, inspect its state, edit it, then resume.

Two design points worth internalizing

**Setting a value for thread_id is your problem, not the framework’s. The graph does not invent one for you. Any invocation that omits thread_id from the config gets a fresh empty state, which is almost never what you want in a durable app. In practice you generate a stable thread_id per conversation at your application layer (session cookie, user ID plus timestamp, chat room ID) and thread it through every invoke and stream call.

Checkpoints are automatic and per-step. You do not call .save_state() or .load_state() anywhere. Every time a node produces a state update, the checkpointer writes it. Every time you invoke the graph, the checkpointer loads the latest state for the thread. This is what makes the migration from “prototype in a notebook with MemorySaver” to “production service with PostgresSaver” a two-line change instead of a rewrite.

What we covered

A checkpointer passed to .compile() gives a graph durable per-thread state.
MemorySaver for in-process, SqliteSaver for single-file on-disk, PostgresSaver for production. Same interface, same graph, different storage.
A thread_id in the invoke config identifies the conversation. Same thread ID means shared history; different means isolated.
The state survives full Python process restarts with no extra code.
.get_state() and .get_state_history() expose the recorded state for inspection and time-travel.

The next chapter Chapter “Human-in-the-Loop Patterns” uses this same mechanism: pause the graph mid-run with interrupt(), hand control back to your calling code (or a human), edit the checkpoint, then resume from where you stopped. Every one of those capabilities is a direct consequence of the checkpointer machinery in this chapter.

Up next

Human-in-the-Loop Patterns