Pause a long-running agent run, hand off to a human for review, then resume — and survive crashes mid-run.
Saving a video game. You reach a checkpoint, can stop there, come back tomorrow, resume from where you left off.
state -> checkpoint(v1) ... checkpoint(v2) -> state
|
v
resume from v2 after crash
Checkpointing persists the agent state — messages, tool results, intermediate variables — after every node. After a crash or human approval, the agent resumes from the latest checkpoint. LangGraph uses thread_id + checkpointer (memory or Postgres). Cron-driven agents run in pieces; long workflows survive reboots; HITL flows pause at any node.
from langgraph.checkpoint.sqlite import SqliteSaver
memory = SqliteSaver.from_conn_string(':memory:')
app = graph.compile(checkpointer=memory)
config = {'configurable': {'thread_id': '42'}}
print(app.invoke({'messages': []}, config=config))
SqliteSaver or PostgresSaver attaches state to thread_id; pause/resume without writing your own persistence layer.
# resume after crash
for step in app.stream({'messages': []}, config=config):
print(step)
if crash_now: break
# later, when restored:
for step in app.stream(None, config=config):
print(step)
Passing stream=None tells the graph to continue from the most recent checkpoint.
# inspect / rewind by thread
for state in app.get_state_history(config):
print('checkpoint:', state.checkpoint_id, 'next:', state.next)
get_state_history exposes all checkpoints — useful for debugging and rewinding.
Building your own saving logic instead of using a checkpointer. You’ll re-implement thread_id, atomic writes, garbage collection — poorly.
Use LangGraph checkpointer. Pass thread_id in config; stream None to resume; get_state_history to inspect or roll back.