Checkpoints allow LangGraph agents to persist their state within and across multiple interactions. A checkpoint is a snapshot of the graph state at a given point in time, identified by a unique, monotonically increasing ID.
Learn more: For conceptual guides and tutorials on persistence, see the Persistence documentation.
| Concept | Description |
|---|---|
| Checkpoint | Snapshot of graph state at a point in time, including channel values, channel versions, and version tracking per node |
| Thread | A sequence of checkpoints identified by a unique thread_id; optionally narrow to a specific checkpoint with checkpoint_id |
| Serde | Serialization protocol used to encode/decode checkpoint data; defaults to JsonPlusSerializer |
| Pending writes | Writes from successful nodes preserved when other nodes fail, enabling resumption without re-running completed work |
LangGraph provides several checkpoint saver implementations:
| Saver | Package | Async Support | Use Case |
|---|---|---|---|
InMemorySaver | langgraph-checkpoint | Yes | Debugging and testing only |
SqliteSaver | langgraph-checkpoint-sqlite | No | Lightweight demos and small projects |
AsyncSqliteSaver | langgraph-checkpoint-sqlite | Yes | Async SQLite; not recommended for production |
PostgresSaver | langgraph-checkpoint-postgres | No | Production workloads with full checkpoint history |
AsyncPostgresSaver | langgraph-checkpoint-postgres | Yes | Async production workloads with full checkpoint history |
from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer) from langgraph.checkpoint.postgres import PostgresSaver with PostgresSaver.from_conn_string("postgresql://user:pass@localhost/db") as checkpointer: checkpointer.setup() graph = builder.compile(checkpointer=checkpointer) from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver async with AsyncPostgresSaver.from_conn_string("postgresql://user:pass@localhost/db") as checkpointer: await checkpointer.setup() graph = builder.compile(checkpointer=checkpointer) All checkpoint savers use a serialization protocol to encode and decode checkpoint data. The default serializer is JsonPlusSerializer, which uses ormsgpack with a fallback to an extended JSON format that handles LangChain/LangGraph types, datetimes, enums, and more.
For sensitive data, wrap any serializer with EncryptedSerializer to encrypt checkpoint contents at rest:
from langgraph.checkpoint.serde.encrypted import EncryptedSerializer from langgraph.checkpoint.postgres import PostgresSaver encrypted_serde = EncryptedSerializer.from_pycryptodome_aes(encryption_key) checkpointer = PostgresSaver(conn, serde=encrypted_serde) Subclass BaseCheckpointSaver and implement the following methods:
get_tuple — Retrieve a specific checkpointlist — List checkpoints, optionally filteredput — Store a checkpointput_writes — Store pending writesdelete_thread — Delete all checkpoints for a threadFor async usage, implement the corresponding async variants (aget_tuple, alist, aput, aput_writes, adelete_thread).
Metadata associated with a checkpoint.
Base class for creating a graph checkpointer.
Protocol for serialization and deserialization of objects.
Protocol for encryption and decryption of data.
Serializer that uses ormsgpack, with optional fallbacks.
Serializer that encrypts and decrypts data using an encryption protocol.
An in-memory checkpoint saver.
Persistent dictionary with an API compatible with shelve and anydbm.
A checkpoint saver that stores checkpoints in a SQLite database.
An asynchronous checkpoint saver that stores checkpoints in a SQLite database.
Checkpointer that stores checkpoints in a Postgres database.
Asynchronous checkpointer that stores checkpoints in a Postgres database.