feat(langgraph): DeltaChannel: store sentinel in blobs, reconstruct from checkpoint_writes (#7586)

# 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>
2026-09-28 20:45:05 +02:00 · 2026-04-29 17:26:17 -04:00
parent a48a045596
commit 5c18bde0f8
26 changed files with 3292 additions and 64 deletions
@@ -4,15 +4,17 @@ import threading
 from collections import defaultdict
 from collections.abc import Iterator, Sequence
 from contextlib import contextmanager
-from typing import Any
+from typing import Any, cast

 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
+    DELTA_SENTINEL,
    WRITES_IDX_MAP,
    ChannelVersions,
    Checkpoint,
    CheckpointMetadata,
    CheckpointTuple,
+    _ChannelWritesHistory,
    get_checkpoint_id,
    get_serializable_checkpoint_metadata,
 )
@@ -23,7 +25,11 @@ from psycopg.types.json import Jsonb
 from psycopg_pool import ConnectionPool

 from langgraph.checkpoint.postgres import _internal
-from langgraph.checkpoint.postgres.base import BasePostgresSaver
+from langgraph.checkpoint.postgres.base import (
+    SELECT_DELTA_COMBINED_SQL,
+    BasePostgresSaver,
+    _DeltaCombinedRow,
+)
 from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver

 Conn = _internal.Conn  # For backward compatibility
@@ -430,6 +436,48 @@ class PostgresSaver(BasePostgresSaver):
                with conn.cursor(binary=True, row_factory=dict_row) as cur:
                    yield cur

+    def _get_channel_writes_history(
+        self, config: RunnableConfig, channel: str
+    ) -> _ChannelWritesHistory:
+        """Fast-path override of `BaseCheckpointSaver._get_channel_writes_history`.
+
+        One combined UNION ALL query (`SELECT_DELTA_COMBINED_SQL`) fetches rows
+        from `checkpoints`, `checkpoint_writes`, and `checkpoint_blobs` in a
+        single roundtrip; the ancestor walk runs in Python.
+        """
+        thread_id = config["configurable"]["thread_id"]
+        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
+        checkpoint_id = get_checkpoint_id(config)
+        if checkpoint_id is None:
+            # Caller didn't specify a target — resolve to the latest
+            # checkpoint on the thread. `get_tuple` without `checkpoint_id`
+            # returns the newest; its config carries the resolved id.
+            target = self.get_tuple(config)
+            if target is None:
+                return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+            checkpoint_id = target.config["configurable"]["checkpoint_id"]
+        with self._cursor() as cur:
+            cur.execute(
+                SELECT_DELTA_COMBINED_SQL,
+                (
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
+                ),
+            )
+            rows = cur.fetchall()
+        return self._build_delta_channel_writes_history(
+            channel=channel,
+            target_id=checkpoint_id,
+            rows=cast("list[_DeltaCombinedRow]", rows),
+        )
+
    def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
        """
        Convert a database row into a CheckpointTuple object.
@@ -4,15 +4,17 @@ import asyncio
 from collections import defaultdict
 from collections.abc import AsyncIterator, Iterator, Sequence
 from contextlib import asynccontextmanager
-from typing import Any
+from typing import Any, cast

 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
+    DELTA_SENTINEL,
    WRITES_IDX_MAP,
    ChannelVersions,
    Checkpoint,
    CheckpointMetadata,
    CheckpointTuple,
+    _ChannelWritesHistory,
    get_checkpoint_id,
    get_serializable_checkpoint_metadata,
 )
@@ -23,7 +25,11 @@ from psycopg.types.json import Jsonb
 from psycopg_pool import AsyncConnectionPool

 from langgraph.checkpoint.postgres import _ainternal
-from langgraph.checkpoint.postgres.base import BasePostgresSaver
+from langgraph.checkpoint.postgres.base import (
+    SELECT_DELTA_COMBINED_SQL,
+    BasePostgresSaver,
+    _DeltaCombinedRow,
+)
 from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver

 Conn = _ainternal.Conn  # For backward compatibility
@@ -391,6 +397,46 @@ class AsyncPostgresSaver(BasePostgresSaver):
                async with conn.cursor(binary=True, row_factory=dict_row) as cur:
                    yield cur

+    async def _aget_channel_writes_history(
+        self, config: RunnableConfig, channel: str
+    ) -> _ChannelWritesHistory:
+        """Fast-path override of `BaseCheckpointSaver._aget_channel_writes_history`.
+
+        One combined UNION ALL query (`SELECT_DELTA_COMBINED_SQL`) fetches rows
+        from `checkpoints`, `checkpoint_writes`, and `checkpoint_blobs` in a
+        single roundtrip; rows are assembled by the shared pure helper on
+        `BasePostgresSaver`.
+        """
+        thread_id = config["configurable"]["thread_id"]
+        checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
+        checkpoint_id = get_checkpoint_id(config)
+        if checkpoint_id is None:
+            target = await self.aget_tuple(config)
+            if target is None:
+                return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+            checkpoint_id = target.config["configurable"]["checkpoint_id"]
+        async with self._cursor() as cur:
+            await cur.execute(
+                SELECT_DELTA_COMBINED_SQL,
+                (
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
+                    thread_id,
+                    checkpoint_ns,
+                    channel,
+                ),
+            )
+            rows = await cur.fetchall()
+        return self._build_delta_channel_writes_history(
+            channel=channel,
+            target_id=checkpoint_id,
+            rows=cast("list[_DeltaCombinedRow]", rows),
+        )
+
    async def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
        """
        Convert a database row into a CheckpointTuple object.
@@ -4,13 +4,16 @@ import random
 import warnings
 from collections.abc import Sequence
 from importlib.metadata import version as get_version
-from typing import Any, cast
+from typing import Any, TypedDict, cast

 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
+    DELTA_SENTINEL,
    WRITES_IDX_MAP,
    BaseCheckpointSaver,
    ChannelVersions,
+    PendingWrite,
+    _ChannelWritesHistory,
    get_checkpoint_id,
 )
 from langgraph.checkpoint.serde.types import TASKS
@@ -153,6 +156,62 @@ INSERT_CHECKPOINT_WRITES_SQL = """
 """


+class _DeltaCombinedRow(TypedDict, total=False):
+    """One row from `SELECT_DELTA_COMBINED_SQL` (a UNION ALL of three tables).
+
+    Every row carries `_kind` ("p" / "w" / "b") plus whichever columns are
+    relevant for that kind; irrelevant columns are NULL and typed as `None`.
+    """
+
+    _kind: str  # always present: "p", "w", or "b"
+    # checkpoint row ("p")
+    checkpoint_id: str | None
+    parent_checkpoint_id: str | None
+    ver: str | None
+    # write / blob rows ("w", "b")
+    type: str | None
+    blob: bytes | None
+    # write row only ("w")
+    task_id: str | None
+    idx: int | None
+    # blob row only ("b")
+    version: str | None
+
+
+# DeltaChannel reconstruction: one UNION ALL query fetches checkpoints,
+# writes, and blobs for `channel` in one roundtrip; the ancestor walk runs
+# in Python in `_build_delta_channel_writes_history`.
+#
+# Parameter order: (channel, thread_id, checkpoint_ns,
+#                   thread_id, checkpoint_ns, channel,
+#                   thread_id, checkpoint_ns, channel)
+SELECT_DELTA_COMBINED_SQL = """
+    SELECT 'p'::text       AS _kind,
+           checkpoint_id,
+           parent_checkpoint_id,
+           checkpoint -> 'channel_versions' ->> %s AS ver,
+           NULL::text       AS type,
+           NULL::bytea      AS blob,
+           NULL::text       AS task_id,
+           NULL::int        AS idx,
+           NULL::text       AS version
+    FROM checkpoints
+    WHERE thread_id = %s AND checkpoint_ns = %s
+    UNION ALL
+    SELECT 'w',
+           checkpoint_id, NULL, NULL,
+           type, blob, task_id, idx, NULL
+    FROM checkpoint_writes
+    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s
+    UNION ALL
+    SELECT 'b',
+           NULL, NULL, NULL,
+           type, blob, NULL, NULL, version
+    FROM checkpoint_blobs
+    WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s
+"""
+
+
 class BasePostgresSaver(BaseCheckpointSaver[str]):
    SELECT_SQL = SELECT_SQL
    SELECT_PENDING_SENDS_SQL = SELECT_PENDING_SENDS_SQL
@@ -195,6 +254,83 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
            if t.decode() != "empty"
        }

+    def _build_delta_channel_writes_history(
+        self,
+        *,
+        channel: str,
+        target_id: str,
+        rows: Sequence[_DeltaCombinedRow],
+    ) -> _ChannelWritesHistory:
+        """Reconstruct one delta channel's history from the combined UNION ALL rows.
+
+        Pure data transform shared by sync (`PostgresSaver`) and async
+        (`AsyncPostgresSaver`); both paths run `SELECT_DELTA_COMBINED_SQL`
+        and feed the tagged rows here.
+
+        Walk is newest → oldest from the target's parent. A non-sentinel
+        blob in `checkpoint_blobs` (a pre-delta snapshot) terminates the
+        walk and is returned as the seed so replay starts from it.
+
+        Writes stored at `target_id` itself are pending writes for the next
+        step and are excluded — the walk begins at the target's parent.
+        """
+        parent_of: dict[str, str | None] = {}
+        ver_of: dict[str, str | None] = {}
+        writes_by_cid: dict[str, list[tuple[str, bytes, str, int]]] = {}
+        blob_by_ver: dict[str, tuple[str, bytes]] = {}
+
+        for r in rows:
+            kind = r["_kind"]
+            if kind == "p":
+                cid = cast(str, r["checkpoint_id"])
+                parent_of[cid] = r["parent_checkpoint_id"]
+                ver_of[cid] = r["ver"]
+            elif kind == "w":
+                cid = cast(str, r["checkpoint_id"])
+                writes_by_cid.setdefault(cid, []).append(
+                    cast(
+                        "tuple[str, bytes, str, int]",
+                        (r["type"], r["blob"], r["task_id"], r["idx"]),
+                    )
+                )
+            else:  # kind == "b"
+                blob_by_ver[cast(str, r["version"])] = cast(
+                    "tuple[str, bytes]", (r["type"], r["blob"])
+                )
+
+        # newest write first per ancestor (task_id DESC, idx DESC)
+        for ws in writes_by_cid.values():
+            ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
+
+        ancestors: list[str] = []
+        cur_cid: str | None = parent_of.get(target_id)
+        while cur_cid is not None:
+            ancestors.append(cur_cid)
+            cur_cid = parent_of.get(cur_cid)
+        if not ancestors:
+            return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=[])
+
+        collected: list[PendingWrite] = []  # newest first; reversed at the end
+        for cid in ancestors:
+            # Collect writes first — they encode the transition FROM this
+            # ancestor's state to its child's and must be included even if
+            # this ancestor is also the seed checkpoint.
+            for type_tag, write_blob, task_id, _idx in writes_by_cid.get(cid, []):
+                val = self.serde.loads_typed((type_tag, write_blob))
+                collected.append((task_id, channel, val))
+            # Then check seed terminator.
+            ver = ver_of.get(cid)
+            if ver is not None:
+                seed_blob = blob_by_ver.get(ver)
+                if seed_blob is not None and seed_blob[0] != "empty":
+                    blob_value = self.serde.loads_typed(seed_blob)
+                    if blob_value is not DELTA_SENTINEL:
+                        collected.reverse()
+                        return _ChannelWritesHistory(seed=blob_value, writes=collected)
+
+        collected.reverse()  # oldest → newest
+        return _ChannelWritesHistory(seed=DELTA_SENTINEL, writes=collected)
+
    def _dump_blobs(
        self,
        thread_id: str,