mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-17 21:25:46 +02:00
5c18bde0f8
# DeltaChannel: sentinel-based checkpoint blobs + write-replay
reconstruction
## Summary
`DeltaChannel` is a new fold-reducer channel that stores only a
zero-byte sentinel in checkpoint blobs instead of the full accumulated
value. On restore, the runtime replays ancestor writes through the
reducer to reconstruct state. For long-running threads with large
accumulating state (e.g. message histories), this delivers dramatically
smaller checkpoint blobs with configurable read-depth bounds.
```python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.message import _messages_delta_reducer
class State(TypedDict):
# blob per step: ~60 bytes (sentinel) instead of growing full list
messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
# bound read depth to 10 steps via periodic snapshots
messages_bounded: Annotated[list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=10)]
```
---
## Storage benchmarks (InMemory, ~400 char/msg)
**Messages blob storage** (`checkpoint_blobs` bytes for the messages
channel):
| turns | add\_messages | delta(inf) | delta(freq=50) | delta(freq=10) |
delta(freq=5) |
|------:|-------------:|-----------:|---------------:|---------------:|--------------:|
| 10 | 91.0 KB | 60 B (1517x) | 60 B (1517x) | 14.4 KB (6x) | 32.6 KB
(3x) |
| 50 | 2.20 MB | 300 B (7347x) | 67.1 KB (33x) | 423 KB (5x) | 864 KB
(3x) |
| 100 | 8.78 MB | 600 B (14636x) | 310 KB (28x) | 1.72 MB (5x) | 3.48 MB
(3x) |
| 250 | 54.80 MB | 1.5 KB (36536x) | 2.09 MB (26x) | 10.87 MB (5x) |
21.84 MB (3x) |
| 500 | 219.19 MB | 3.0 KB (73063x) | 8.56 MB (26x) | 43.67 MB (5x) |
87.50 MB (3x) |
**Total checkpoint storage** (blobs + writes + metadata):
| turns | add\_messages | delta(inf) | delta(freq=50) | delta(freq=10) |
delta(freq=5) |
|------:|-------------:|-----------:|---------------:|---------------:|--------------:|
| 10 | 129.7 KB | 38.7 KB (3.4x) | 38.7 KB (3.4x) | 53.1 KB (2.4x) |
71.2 KB (1.8x) |
| 50 | 2.40 MB | 196 KB (12x) | 263 KB (9x) | 620 KB (3.9x) | 1.06 MB
(2.3x) |
| 100 | 9.18 MB | 394 KB (23x) | 703 KB (13x) | 2.12 MB (4.3x) | 3.87 MB
(2.4x) |
| 250 | 55.79 MB | 987 KB (57x) | 3.07 MB (18x) | 11.86 MB (4.7x) |
22.82 MB (2.4x) |
| 500 | 221.16 MB | 1.98 MB (112x) | 10.53 MB (21x) | 45.64 MB (4.9x) |
89.48 MB (2.5x) |
**Write-phase peak heap**:
| turns | add\_messages | delta(inf) | delta(freq=50) | delta(freq=10) |
delta(freq=5) |
|------:|-------------:|-----------:|---------------:|---------------:|--------------:|
| 10 | 456 KB | 199 KB (2.3x) | 199 KB (2.3x) | 212 KB (2.2x) | 232 KB
(2.0x) |
| 50 | 3.04 MB | 742 KB (4.1x) | 805 KB (3.8x) | 1.21 MB (2.5x) | 1.67
MB (1.8x) |
| 100 | 10.70 MB | 1.41 MB (7.6x) | 1.82 MB (5.9x) | 3.42 MB (3.1x) |
5.25 MB (2.0x) |
| 250 | 60.44 MB | 3.36 MB (18x) | 5.67 MB (11x) | 14.87 MB (4.1x) |
26.31 MB (2.3x) |
**Read-phase avg `get_state` latency** (5 calls, InMemory):
| turns | add\_messages | delta(inf) | delta(freq=50) | delta(freq=10) |
delta(freq=5) |
|------:|-------------:|-----------:|---------------:|---------------:|--------------:|
| 10 | 0.7 ms | 1.1 ms (0.6x) | 1.1 ms (0.6x) | 0.8 ms (0.9x) | 0.6 ms
(1.1x) |
| 50 | 2.7 ms | 5.3 ms (0.5x) | 3.5 ms (0.8x) | 2.7 ms (1.0x) | 2.7 ms
(1.0x) |
| 100 | 5.5 ms | 11.1 ms (0.5x) | 6.0 ms (0.9x) | 5.2 ms (1.1x) | 5.4 ms
(1.0x) |
| 250 | 12.9 ms | 27.2 ms (0.5x) | 13.6 ms (0.9x) | 12.9 ms (1.0x) |
13.0 ms (1.0x) |
**Postgres `get_tuple` read latency** (~100 tok/msg per step):
| steps | full-list | delta(inf) | delta(freq=50) | delta(freq=10) |
delta(freq=5) |
|------:|----------:|-----------:|---------------:|---------------:|--------------:|
| 10 | 0.29 ms | 0.21 ms (1.4x) | 0.19 ms (1.6x) | 0.19 ms (1.5x) | 0.19
ms (1.6x) |
| 50 | 0.19 ms | 0.15 ms (1.3x) | 0.19 ms (1.0x) | 0.22 ms (0.8x) | 0.29
ms (0.7x) |
| 100 | 0.27 ms | 0.17 ms (1.6x) | 0.22 ms (1.2x) | 0.23 ms (1.2x) |
0.21 ms (1.3x) |
| 500 | 0.60 ms | 0.30 ms (2.0x) | 0.66 ms (0.9x) | 0.56 ms (1.1x) |
0.69 ms (0.9x) |
**Takeaway:** `snapshot_frequency=10` matches full-list read latency
while still saving 5x on blob storage and ~4x on total storage.
---
## How it works
### Checkpoint blobs
`checkpoint()` always returns `DELTA_SENTINEL` (a zero-byte msgpack ext
marker) instead of the accumulated value. On restore, the saver's
`_get_channel_writes_history` walks the ancestor chain collecting
`checkpoint_writes` entries and replays them through the reducer:
```python
# blob stored per step: ~1 byte (sentinel)
# vs. full list growing O(N) every step with BinaryOperatorAggregate
```
### Reducer interface
`DeltaChannel` takes a **batch reducer** `(state, list[writes]) ->
state` — all writes for a step arrive in one call, enabling single-pass
implementations:
```python
# ❌ Don't use add_messages directly — it's a binary operator, not a batch reducer
messages: Annotated[list, DeltaChannel(add_messages)] # wrong
# ✅ Use _messages_delta_reducer — single pass, dedup by ID, RemoveMessage support
messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
# ✅ Or write your own batch reducer for custom types
def my_dict_reducer(state: dict, writes: list[dict]) -> dict:
result = dict(state)
for w in writes:
result.update(w)
return result
files: Annotated[dict, DeltaChannel(my_dict_reducer)]
```
### Snapshot frequency
`snapshot_frequency=N` writes a full `_DeltaSnapshot` blob every N
pregel steps, bounding replay depth regardless of thread length.
Snapshots are eager — written even if the channel had no update that
step, so the depth bound always holds:
```python
# Replay walks at most 10 ancestors before hitting a snapshot
messages: Annotated[list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=10)]
```
### Migration from `BinaryOperatorAggregate`
Pre-existing threads written under `BinaryOperatorAggregate` work
transparently after swapping the annotation — the saver detects a
plain-value ancestor blob and uses it as the reconstruction seed:
```python
# Before: BinaryOperatorAggregate stores full list every step
items: Annotated[list, add_messages]
# After: DeltaChannel — existing checkpoints still readable, new steps use sentinel
items: Annotated[list, DeltaChannel(_messages_delta_reducer)]
```
### Async write-ordering safety
In `durability="async"` mode (default), `put_writes` calls are
fire-and-forget. `AsyncPregelLoop` tracks in-flight `aput_writes`
futures for DeltaChannel channels in `_delta_write_futs` and drains them
via `await asyncio.gather()` in `_checkpointer_put_after_previous`
before `aput()` — ensuring `checkpoint_writes` are durable before the
sentinel blob is committed.
---
## What's in scope
- **`libs/langgraph/langgraph/channels/delta.py`** — `DeltaChannel`
implementation
- **`libs/langgraph/langgraph/graph/message.py`** —
`_messages_delta_reducer` (experimental)
- **`libs/checkpoint/`** — `_get_channel_writes_history` ancestor-walk
API on `BaseCheckpointSaver`, `InMemorySaver` optimized override
- **`libs/checkpoint-postgres/`** — `PostgresSaver` /
`AsyncPostgresSaver` single-roundtrip UNION ALL override
- **`libs/langgraph/langgraph/pregel/`** — `channels_from_checkpoint` /
`create_checkpoint` wiring, async write-ordering safety
---
## Follow-ups
- **Batch reconstruction**: each DeltaChannel field issues its own
`_get_channel_writes_history` call; a single walk collecting all
sentinel channels would reduce roundtrips proportionally to the number
of DeltaChannel fields.
- **Sync write ordering**: `BackgroundExecutor.__exit__` guarantees
completion before `invoke()` returns, but within a run there's no
explicit ordering between `put_writes` and `put`. Two-phase commit for
sync would close this gap.
- **`ShallowPostgresSaver` compatibility**: shallow savers keep only the
latest checkpoint and have no parent chain to walk; DeltaChannel is
currently incompatible and should raise or warn at compile time.
- Updating the writes table w/ delta epoch ids for more efficient reads
- follow up w/ LSD checkpointer implementations to support delta
channel! and update prune
---------
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: ccurme <chester.curme@gmail.com>
Co-authored-by: Will Fu-Hinthorn <will@langchain.dev>
104 lines
1.2 KiB
Plaintext
104 lines
1.2 KiB
Plaintext
.vs/
|
|
.vscode/
|
|
.idea/
|
|
# Byte-compiled / optimized / DLL files
|
|
__pycache__/
|
|
*.py[cod]
|
|
*$py.class
|
|
|
|
# Distribution / packaging
|
|
.Python
|
|
build/
|
|
develop-eggs/
|
|
dist/
|
|
downloads/
|
|
eggs/
|
|
.eggs/
|
|
lib/
|
|
lib64/
|
|
parts/
|
|
sdist/
|
|
var/
|
|
wheels/
|
|
pip-wheel-metadata/
|
|
share/python-wheels/
|
|
*.egg-info/
|
|
.installed.cfg
|
|
*.egg
|
|
MANIFEST
|
|
|
|
# PyInstaller
|
|
# Usually these files are written by a python script from a template
|
|
# before PyInstaller builds the exe, so as to inject date/other infos into it.
|
|
*.manifest
|
|
*.spec
|
|
|
|
# Installer logs
|
|
pip-log.txt
|
|
pip-delete-this-directory.txt
|
|
|
|
# Unit test / coverage reports
|
|
htmlcov/
|
|
.tox/
|
|
.nox/
|
|
.coverage
|
|
.coverage.*
|
|
.cache
|
|
nosetests.xml
|
|
coverage.xml
|
|
*.cover
|
|
*.py,cover
|
|
.hypothesis/
|
|
.pytest_cache/
|
|
|
|
# Django stuff:
|
|
*.log
|
|
local_settings.py
|
|
db.sqlite3
|
|
db.sqlite3-journal
|
|
|
|
# PyBuilder
|
|
target/
|
|
|
|
# Jupyter Notebook
|
|
.ipynb_checkpoints
|
|
notebooks/
|
|
|
|
# IPython
|
|
profile_default/
|
|
ipython_config.py
|
|
|
|
# pyenv
|
|
.python-version
|
|
|
|
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
|
|
__pypackages__/
|
|
|
|
# Environments
|
|
.env
|
|
.envrc
|
|
.venv
|
|
.venvs
|
|
env/
|
|
venv/
|
|
ENV/
|
|
env.bak/
|
|
venv.bak/
|
|
|
|
# mypy
|
|
.mypy_cache/
|
|
.dmypy.json
|
|
dmypy.json
|
|
|
|
# Pyre type checker
|
|
.pyre/
|
|
|
|
# macOS display setting files
|
|
.DS_Store
|
|
|
|
.vercel
|
|
.turbo
|
|
.editorconfig
|
|
.scratch
|
|
.worktrees/
|