## Summary Builds on #7586. Adds `snapshot_frequency: int | None` to `DeltaChannel`, letting users trade storage for bounded read depth. Also promotes `channels/_delta.py` from private to public (`channels/delta.py`). ### How it works Every Nth **pregel step**, `create_checkpoint` writes a `_DeltaSnapshot` blob instead of `DELTA_SENTINEL`. The ancestor walk in `_get_channel_writes_history` terminates at the snapshot rather than walking the full chain, bounding replay to at most N steps. Snapshots are **eager**: fired even on steps where the channel had no write (via a `get_next_version` version bump), so the depth bound holds unconditionally — no risk of the cadence drifting if a channel happens to be silent at a snapshot step. ### Storage formula | Mode | Blob storage | Read depth | |------|-------------|------------| | `snapshot_frequency=None` (pure delta) | O(N) — sentinels only | O(N) steps | | `snapshot_frequency=K` | O(N²/K) — periodic snapshots of growing size | O(K) steps | | add_messages / BinOp | O(N²) — full blob every step | O(1) | At N turns with ~400 char/msg messages, total snapshot storage ≈ N²/(2K) × avg_msg_size, since each snapshot blob grows linearly with accumulated messages. ### Key design decisions - **Step-based**: `snapshot_frequency=K` means "snapshot every K pregel steps." `create_checkpoint` has the step number; the channel itself doesn't need to track writes. - **Eager**: version-bumped via `get_next_version` even on non-write steps so `put()` always stores the blob. - **`_DeltaSnapshot` NamedTuple + msgpack ext type** (`EXT_DELTA_SNAPSHOT = 7`): serde type tag dispatches in `from_checkpoint` — no dict key inspection, no collision risk. - **`from_checkpoint` semantics**: `_DeltaSnapshot` → restore value directly (no replay needed); `DELTA_SENTINEL` / `MISSING` → replay from ancestor writes; plain value → pre-migration BinOp blob. - **InMemorySaver and PostgresSaver updated**: `_get_channel_writes_history` collects the snapshot ancestor's pending_writes before terminating (they encode the *next* step's transition, unlike pre-delta migration blobs which subsume their own writes). - **`snapshot_frequency=None`** is the pure-delta default (replaces `math.inf`). ### Benchmark results (InMemory, ~400 char/msg) **Storage** | turns | ctx | freq=1 | freq=5 | freq=10 | freq=50 | freq=inf | |------:|----:|-------:|-------:|--------:|--------:|---------:| | 50 | ~10K tok | 5.9 MB | 1.2 MB | 601.3 KB | 119.8 KB | 29.5 KB | | 100 | ~20K tok | 23.7 MB | 4.8 MB | 2.4 MB | 475.8 KB | 58.4 KB | | 200 | ~40K tok | 94.6 MB | 19.0 MB | 9.5 MB | 1.9 MB | 116.4 KB | | 500 | ~100K tok | 591.5 MB | 118.4 MB | 59.2 MB | 11.8 MB | 290.3 KB | **Read latency** (avg of 5 `get_state` calls) | turns | ctx | freq=1 | freq=5 | freq=10 | freq=50 | freq=inf | |------:|----:|-------:|-------:|--------:|--------:|---------:| | 50 | ~10K tok | 0.4ms | 0.4ms | 0.7ms | 0.9ms | 1.8ms | | 100 | ~20K tok | 0.7ms | 0.9ms | 1.0ms | 1.7ms | 5.7ms | | 200 | ~40K tok | 1.5ms | 1.7ms | 4.5ms | 3.7ms | 20.1ms | | 500 | ~100K tok | 3.6ms | 4.2ms | 4.4ms | 9.0ms | 110.3ms | **Per-invoke write latency** | turns | ctx | freq=1 | freq=5 | freq=10 | freq=50 | freq=inf | |------:|----:|-------:|-------:|--------:|--------:|---------:| | 50 | ~10K tok | 1.5ms | 1.1ms | 1.1ms | 1.3ms | 1.7ms | | 100 | ~20K tok | 2.5ms | 1.6ms | 1.5ms | 1.7ms | 3.3ms | | 200 | ~40K tok | 3.4ms | 2.3ms | 2.2ms | 2.5ms | 8.3ms | | 500 | ~100K tok | 6.2ms | 4.2ms | 3.6ms | 4.1ms | 39.2ms | ## Test plan - [x] `make format` / `make lint` clean across `langgraph`, `checkpoint`, `checkpoint-postgres` - [x] `tests/test_channels.py` — 37 passing including step-based and eager-snapshot tests - [x] `tests/test_delta_channel_migration.py` — all passing - [x] Full suite: 1387 passing, 6 pre-existing failures unrelated to this branch --------- Co-authored-by: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
LangGraph Checkpoint Postgres
Implementation of LangGraph CheckpointSaver that uses Postgres.
Dependencies
By default langgraph-checkpoint-postgres installs psycopg (Psycopg 3) without any extras. However, you can choose a specific installation that best suits your needs here (for example, psycopg[binary]).
Security
Important
Set
LANGGRAPH_STRICT_MSGPACK=trueor pass an explicitallowed_msgpack_moduleslist when creating your checkpointer. This restricts checkpoint deserialization to known-safe types, preventing code execution if the database is compromised. See the langgraph-checkpoint README for details.
Usage
Important
When using Postgres checkpointers for the first time, make sure to call
.setup()method on them to create required tables. See example below.
Important
When manually creating Postgres connections and passing them to
PostgresSaverorAsyncPostgresSaver, make sure to includeautocommit=Trueandrow_factory=dict_row(from psycopg.rows import dict_row). See a full example in this how-to guide.Why these parameters are required:
autocommit=True: Required for the.setup()method to properly commit the checkpoint tables to the database. Without this, table creation may not be persisted.row_factory=dict_row: Required because the PostgresSaver implementation accesses database rows using dictionary-style syntax (e.g.,row["column_name"]). The defaulttuple_rowfactory returns tuples that only support index-based access (e.g.,row[0]), which will causeTypeErrorexceptions when the checkpointer tries to access columns by name.Example of incorrect usage:
# ❌ This will fail with TypeError during checkpointer operations with psycopg.connect(DB_URI) as conn: # Missing autocommit=True and row_factory=dict_row checkpointer = PostgresSaver(conn) checkpointer.setup() # May not persist tables properly # Any operation that reads from database will fail with: # TypeError: tuple indices must be integers or slices, not str
from langgraph.checkpoint.postgres import PostgresSaver
write_config = {"configurable": {"thread_id": "1", "checkpoint_ns": ""}}
read_config = {"configurable": {"thread_id": "1"}}
DB_URI = "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
# call .setup() the first time you're using the checkpointer
checkpointer.setup()
checkpoint = {
"v": 4,
"ts": "2024-07-31T20:14:19.804150+00:00",
"id": "1ef4f797-8335-6428-8001-8a1503f9b875",
"channel_values": {
"my_key": "meow",
"node": "node"
},
"channel_versions": {
"__start__": 2,
"my_key": 3,
"start:node": 3,
"node": 3
},
"versions_seen": {
"__input__": {},
"__start__": {
"__start__": 1
},
"node": {
"start:node": 2
}
},
}
# store checkpoint
checkpointer.put(write_config, checkpoint, {}, {})
# load checkpoint
checkpointer.get(read_config)
# list checkpoints
list(checkpointer.list(read_config))
Async
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
async with AsyncPostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpoint = {
"v": 4,
"ts": "2024-07-31T20:14:19.804150+00:00",
"id": "1ef4f797-8335-6428-8001-8a1503f9b875",
"channel_values": {
"my_key": "meow",
"node": "node"
},
"channel_versions": {
"__start__": 2,
"my_key": 3,
"start:node": 3,
"node": 3
},
"versions_seen": {
"__input__": {},
"__start__": {
"__start__": 1
},
"node": {
"start:node": 2
}
},
}
# store checkpoint
await checkpointer.aput(write_config, checkpoint, {}, {})
# load checkpoint
await checkpointer.aget(read_config)
# list checkpoints
[c async for c in checkpointer.alist(read_config)]