Checkpoints

Video games save your progress, so that when you die, you don't start from the beginning. LangGraph does the same thing for agents. After every step, it can save the whole state. That one feature unlocks memory, pause and resume, recovery, and even time travel.

• Games save your progress
• Agents can too
• One feature unlocks a lot

Threads

We compile the graph with a checkpointer. Then every run needs a thread ID, which works like a separate save file. Priya gets one, and Sam gets another. We run Priya's twice, and Sam's once. Run it. Priya's count goes up to two, because her state was saved. Sam starts from one. Each customer gets their own conversation, with no extra code. And we can read any thread's saved state.

from typing import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    count: int

g = StateGraph(State)
g.add_node("visit", lambda s: {"count": s.get("count", 0) + 1})
g.add_edge(START, "visit")
g.add_edge("visit", END)
app = g.compile(checkpointer=InMemorySaver())

priya = {"configurable": {"thread_id": "priya"}}
sam = {"configurable": {"thread_id": "sam"}}

print(app.invoke({}, priya))
print(app.invoke({}, priya))
print(app.invoke({}, sam))
print("Saved for Priya:", app.get_state(priya).values)

{'count': 1}
{'count': 2}
{'count': 1}
Saved for Priya: {'count': 2}

Surviving a crash

An in memory saver is lost when your program stops. For production, you need a real database. Here's the same graph, saving to SQLite. We build the app, run it once, then build a brand new app, as if the server restarted, and run the same thread. Run it. It carried on from where it stopped. The second run saw the first run's count. In production you'd use Postgres. Your agent can now survive restarts, deployments, and crashes.

import sqlite3
from typing import TypedDict
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    count: int

def build():
    g = StateGraph(State)
    g.add_node("visit", lambda s: {"count": s.get("count", 0) + 1})
    g.add_edge(START, "visit")
    g.add_edge("visit", END)
    db = sqlite3.connect("agent.db", check_same_thread=False)
    return g.compile(checkpointer=SqliteSaver(db))

config = {"configurable": {"thread_id": "ticket-881"}}
print("Before the crash:", build().invoke({}, config))
print("After a restart: ", build().invoke({}, config))

Before the crash: {'count': 1}
After a restart:  {'count': 2}

What checkpoints give you

Look what we've just got. Conversation memory, because threads persist. Crash recovery, because a run can continue. Pause and resume, because a run can wait for days. And inspection, as you can read the state at any step. Next, we'll use that to put a human in the loop.

Summary