LangGraph persistence — checkpointers and threads

Your LangGraph chatbot works perfectly in dev. You ship it. A user sends three messages, gets responses, closes the tab, comes back an hour later, and asks “what did I just ask you about?” The agent has no idea. The Kubernetes pod that handled turns 1-3 was recycled. Even if it weren’t, your in-memory state evaporated the moment .invoke() returned. This is the moment every LangGraph developer discovers checkpointers.

Why production agents need persistence

A compiled LangGraph runs once and forgets everything. That’s fine for demos. It is not how production agents work. Real agents need:

• Multi-turn conversations — the user’s third message should see history from messages one and two.
• Resumability — if the process restarts mid-run, you don’t lose the conversation.
• Time-travel debugging — when something goes wrong on turn 17, you want to inspect state at turn 12 and rerun from there.
• Human-in-the-loop pauses — pause the graph, walk away, resume hours later.

All four come from one feature: checkpointers (a durable store that saves the full graph state after every node executes, keyed by thread ID).

Because state survives outside the process, the graph can stop mid-run and be resumed later — even from a different process or pod. That “stop and resume” property is also exactly what human-in-the-loop pauses rely on.

Wiring up a checkpointer

You add a checkpointer at compile time:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()  # in-process; great for tests
app = graph.compile(checkpointer=checkpointer)

For production, swap in a persistent backend:

• from langgraph.checkpoint.sqlite import SqliteSaver
• from langgraph.checkpoint.postgres import PostgresSaver

Same interface. The graph’s behaviour doesn’t change — only where state ends up.

Thread IDs — what they actually are

Every invocation of a compiled-with-checkpointer graph needs a config with a thread_id. The thread ID names a conversation; all state for that conversation is stored under it.

config = {"configurable": {"thread_id": "user-42-chat-1"}}
app.invoke({"messages": [HumanMessage("Hi")]}, config=config)
app.invoke({"messages": [HumanMessage("And again")]}, config=config)

Same thread_id → second call sees the state from the first. Different thread_id → fresh conversation. That’s the whole API for multi-turn.

In a chat app, the thread ID is usually the conversation ID. In a long-running agent, it might be the job ID. The important thing is: it’s how LangGraph knows “this is the same conversation.”

A two-turn conversation

# Mocked LangGraph with a checkpointer. The thread_id / config /
# checkpointer pattern mirrors real LangGraph code.

from typing import TypedDict

START, END = "__start__", "__end__"

def add_messages(old, new):
    return (old or []) + (new or [])

class MemorySaver:
    def __init__(self):
        self.store = {}            # thread_id -> [state_history]
    def get(self, thread_id):
        history = self.store.get(thread_id, [])
        return history[-1] if history else None
    def put(self, thread_id, state):
        self.store.setdefault(thread_id, []).append(state)
    def history(self, thread_id):
        return list(self.store.get(thread_id, []))

class StateGraph:
    def __init__(self, schema, reducers=None):
        self.schema, self.reducers = schema, reducers or {}
        self.nodes, self.edges = {}, {}
    def add_node(self, name, fn): self.nodes[name] = fn
    def add_edge(self, src, dst): self.edges[src] = dst
    def compile(self, checkpointer=None): return CompiledGraph(self, checkpointer)

class CompiledGraph:
    def __init__(self, g, cp): self.g, self.cp = g, cp
    def _merge(self, state, update):
        merged = dict(state)
        for k, v in update.items():
            if k in self.g.reducers:
                merged[k] = self.g.reducers[k](state.get(k), v)
            else:
                merged[k] = v
        return merged
    def invoke(self, new_input, config):
        thread = config["configurable"]["thread_id"]
        prior = self.cp.get(thread) if self.cp else None
        state = self._merge(prior or {"messages": []}, new_input)
        current = START
        while True:
            if current in self.g.edges:
                current = self.g.edges[current]
                if current == END:
                    return state   # END state == last node's state; no extra ckpt
                update = self.g.nodes[current](state)
                state = self._merge(state, update)
                if self.cp: self.cp.put(thread, state)   # one checkpoint per node
            else:
                return state
    def get_state(self, config):
        return self.cp.get(config["configurable"]["thread_id"])

# --- the graph ---
class State(TypedDict):
    messages: list

def assistant(state):
    last = state["messages"][-1]["content"].lower()
    if "joins" in last:
        reply = "A JOIN combines rows from two tables based on a key."
    elif "remember" in last or "what did i" in last:
        # uses prior turns!
        prior = " | ".join(m["content"] for m in state["messages"] if m["role"] == "user")
        reply = f"You've asked about: {prior}"
    else:
        reply = "[mock reply]"
    return {"messages": [{"role": "assistant", "content": reply}]}

graph = StateGraph(State, reducers={"messages": add_messages})
graph.add_node("assistant", assistant)
graph.add_edge(START, "assistant")
graph.add_edge("assistant", END)

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# Turn 1
config = {"configurable": {"thread_id": "demo-1"}}
app.invoke({"messages": [{"role": "user", "content": "What are SQL joins?"}]}, config=config)

# Turn 2 — the agent sees history from turn 1 because thread_id is the same
result = app.invoke(
    {"messages": [{"role": "user", "content": "Remember what I asked?"}]},
    config=config,
)

print("--- full conversation (from state) ---")
for m in result["messages"]:
    print(f"[{m['role']}] {m['content']}")

print("\n--- checkpoint history (one entry per step) ---")
for i, snap in enumerate(checkpointer.history("demo-1")):
    print(f"step {i}: {len(snap['messages'])} messages")

--- full conversation (from state) ---
[user] What are SQL joins?
[assistant] A JOIN combines rows from two tables based on a key.
[user] Remember what I asked?
[assistant] You've asked about: What are SQL joins? | Remember what I asked?

--- checkpoint history (one entry per step) ---
step 0: 2 messages
step 1: 4 messages

Two .invoke() calls, same thread_id.

The second call’s assistant node can list both prior user questions because the checkpointer persisted turn 1’s state. Turn 2 starts from a 2-message history and grows it to 4.

This is the whole multi-turn pattern: there’s no chat-loop class, no session manager, just state keyed by thread.

Inspecting and time-travelling

A compiled graph with a checkpointer exposes some extra methods:

• app.get_state(config) — current state for a thread.
• app.get_state_history(config) — all checkpoints, newest first.
• app.update_state(config, update) — manually patch state.

You can fork from any historical checkpoint by grabbing its config (which includes a checkpoint_id) and re-invoking. That’s how “rerun from turn 12 with a different message” works.

history = list(app.get_state_history(config))
old_snapshot = history[5]   # state from 5 steps ago
app.invoke(new_input, config=old_snapshot.config)   # branch from there

This is enormously useful when debugging. You can iterate on a single problematic step without re-running the whole conversation.

Streaming intermediate state

.invoke() runs the graph to completion and gives you the final state. .stream() yields each intermediate update as the graph runs — perfect for UI feedback.

for update in app.stream(input_state, config=config):
    # update is a dict: {node_name: state_after_this_node}
    print(update)

For a chat UI, this is how you show progress as the graph progresses through nodes:

• “thinking…”
• “searching docs…”
• “drafting reply…”

There are several stream modes ("values", "updates", "messages"); pick based on what your UI needs.

Production considerations

A few things that bite teams:

• Checkpointer write latency. Every node execution writes a checkpoint. For a 10-node graph hitting Postgres on the other side of the country, that’s 10 round trips. Use connection pooling and prefer regional databases.
• State size. Checkpoints include the full state. If your state carries large retrieved documents or long message histories, your storage and serialization costs add up fast. Trim or paginate.
• Schema evolution. State is serialized; if you change the TypedDict, old checkpoints may not deserialize cleanly. Version your state shape and write a migration path before changing it.
• Garbage collection. Threads accumulate forever by default. Plan a TTL or cleanup job.

In one breath

• A bare compiled graph runs once and forgets everything; checkpointers save the full state after every node, keyed by thread_id.
• That one feature buys multi-turn conversations, resumability, time-travel debugging, and human-in-the-loop pauses — same thread_id resumes a conversation, a new one starts fresh.
• Swap the backend (MemorySaver → SqliteSaver → PostgresSaver) without touching graph code — persistence is a deployment detail.
• .stream() yields state per node for live UI feedback; get_state_history + re-invoke from an old checkpoint is how time-travel works.
• Watch write latency, state size, schema evolution, and GC — and bind thread IDs to the authenticated user, since they key all of a conversation’s state.

Quick check

Next

Now you can persist state and resume conversations. The last LangGraph piece is human-in-the-loop: pausing the graph at a node so a human can review, edit, or approve before it continues.