WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-10 03:37:51 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(delta-channel): store sentinel in blobs, reconstruct from checkpoint_writes

Browse Source 
DeltaChannel.checkpoint() now returns a zero-byte DeltaChannelSentinel
instead of duplicating delta data in checkpoint_blobs. Reconstruction
walks the parent checkpoint chain via checkpoint_writes (which already
holds per-step writes) and replays them through the operator.

In-memory benchmark (100 turns, ~20K tokens):
  storage: 10.2 MB → 40.5 KB (251x reduction)
  read:    0.6ms → 7.9ms (reconstruction cost, amortized by storage savings)

InMemorySaver and PostgresSaver override get_channel_writes() with
efficient implementations (Python dict walk and recursive CTE respectively).
The base class fallback uses self.list() with a thread-local recursion guard.
This commit is contained in:
Sydney Runkle
2026-04-30 14:49:05 -04:00
parent 5d72a3f390
commit 4a6ddbb0e3
12 changed files with 410 additions and 603 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py
+16 -8
View File
@@ -4,7 +4,7 @@ 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 (
@@ -442,14 +442,22 @@ class PostgresSaver(BasePostgresSaver):
including its configuration, metadata, parent checkpoint (if any),
and pending writes.
"""
with self._cursor() as cur:
channel_values = self._load_blobs(
value["channel_values"],
thread_id=value["thread_id"],
checkpoint_ns=value["checkpoint_ns"],
checkpoint_id=value["checkpoint_id"],
cur=cur,
from langgraph.checkpoint.base import DeltaChannelSentinel
channel_values = self._load_blobs(value["channel_values"])
if any(isinstance(v, DeltaChannelSentinel) for v in channel_values.values()):
cp_config = cast(
RunnableConfig,
{
"configurable": {
"thread_id": value["thread_id"],
"checkpoint_ns": value["checkpoint_ns"],
"checkpoint_id": value["checkpoint_id"],
}
},
)
with self._cursor() as cur:
self._resolve_delta_channels(cp_config, channel_values, cur)
return CheckpointTuple(
{
"configurable": {
libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py
+33 -57
View File
@@ -13,8 +13,7 @@ from langgraph.checkpoint.base import (
Checkpoint,
CheckpointMetadata,
CheckpointTuple,
DeltaChainValue,
DeltaValue,
DeltaChannelSentinel,
get_checkpoint_id,
get_serializable_checkpoint_metadata,
)
@@ -393,80 +392,55 @@ class AsyncPostgresSaver(BasePostgresSaver):
async with conn.cursor(binary=True, row_factory=dict_row) as cur:
yield cur
async def _aload_delta_chain(
async def _aget_channel_writes_cur(
self,
thread_id: str,
checkpoint_ns: str,
checkpoint_id: str,
channel: str,
cur: Any,
) -> DeltaChainValue:
"""Fetch the full delta chain for a channel in one recursive CTE query (async)."""
) -> list[Any]:
"""Fetch writes for `channel` across the checkpoint ancestor chain, oldest→newest (async)."""
await cur.execute(
"""
WITH RECURSIVE chain AS (
SELECT
c.checkpoint_id,
c.parent_checkpoint_id,
cb.version,
cb.type,
cb.blob
FROM checkpoints c
JOIN checkpoint_blobs cb
ON cb.thread_id = %s
AND cb.checkpoint_ns = %s
AND cb.channel = %s
AND cb.version = (c.checkpoint->'channel_versions'->>%s)::text
WHERE c.thread_id = %s
AND c.checkpoint_ns = %s
AND c.checkpoint_id = %s
WITH RECURSIVE chain(cid, depth) AS (
SELECT parent_checkpoint_id, 0
FROM checkpoints
WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id = %s
UNION ALL
SELECT
c.checkpoint_id,
c.parent_checkpoint_id,
cb.version,
cb.type,
cb.blob
FROM chain prev
JOIN checkpoints c ON c.checkpoint_id = prev.parent_checkpoint_id
JOIN checkpoint_blobs cb
ON cb.thread_id = %s
AND cb.checkpoint_ns = %s
AND cb.channel = %s
AND cb.version = (c.checkpoint->'channel_versions'->>%s)::text
WHERE prev.parent_checkpoint_id IS NOT NULL
AND prev.type = 'delta'
SELECT c.parent_checkpoint_id, ch.depth + 1
FROM checkpoints c
JOIN chain ch ON c.checkpoint_id = ch.cid
WHERE ch.cid IS NOT NULL
)
SELECT DISTINCT ON (version) type, blob
FROM chain
ORDER BY version ASC
SELECT cw.type, cw.blob
FROM checkpoint_writes cw
JOIN chain ON cw.checkpoint_id = chain.cid
WHERE cw.thread_id = %s AND cw.checkpoint_ns = %s AND cw.channel = %s
ORDER BY chain.depth DESC, cw.task_id, cw.idx
""",
(
thread_id,
checkpoint_ns,
channel,
channel,
thread_id,
checkpoint_ns,
checkpoint_id,
thread_id,
checkpoint_ns,
channel,
channel,
),
)
rows = await cur.fetchall()
base = None
deltas: list[list[Any]] = []
for row in rows:
blob = self.serde.loads_typed((row["type"], row["blob"]))
if isinstance(blob, DeltaValue):
deltas.append(blob.delta)
else:
base = blob
return DeltaChainValue(base=base, deltas=deltas)
return [self.serde.loads_typed((row["type"], row["blob"])) for row in rows]
async def aget_channel_writes(
self, config: RunnableConfig, channel: str
) -> list[Any]:
thread_id = config["configurable"]["thread_id"]
checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
checkpoint_id = config["configurable"]["checkpoint_id"]
async with self._cursor() as cur:
return await self._aget_channel_writes_cur(
thread_id, checkpoint_ns, checkpoint_id, channel, cur
)
async def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
"""
@@ -489,12 +463,14 @@ class AsyncPostgresSaver(BasePostgresSaver):
if blob_values:
channel_values = self._load_blobs(blob_values)
delta_channels = [
k.decode() for k, t, _ in blob_values if t.decode() == "delta"
ch
for ch, v in channel_values.items()
if isinstance(v, DeltaChannelSentinel)
]
if delta_channels:
async with self._cursor() as cur:
for channel in delta_channels:
channel_values[channel] = await self._aload_delta_chain(
channel_values[channel] = await self._aget_channel_writes_cur(
thread_id, checkpoint_ns, checkpoint_id, channel, cur
)
libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py
+39 -73
View File
@@ -11,8 +11,7 @@ from langgraph.checkpoint.base import (
WRITES_IDX_MAP,
BaseCheckpointSaver,
ChannelVersions,
DeltaChainValue,
DeltaValue,
DeltaChannelSentinel,
get_checkpoint_id,
)
from langgraph.checkpoint.serde.types import TASKS
@@ -188,105 +187,72 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
def _load_blobs(
self,
blob_values: list[tuple[bytes, bytes, bytes]],
*,
thread_id: str = "",
checkpoint_ns: str = "",
checkpoint_id: str = "",
cur: Any = None,
blob_values: Any,
) -> dict[str, Any]:
if not blob_values:
return {}
result: dict[str, Any] = {}
delta_channels: list[str] = []
for k, t, v in blob_values:
channel = k.decode()
type_tag = t.decode()
if type_tag == "delta":
delta_channels.append(channel)
elif type_tag != "empty":
result[channel] = self.serde.loads_typed((type_tag, v))
if delta_channels and cur is not None and checkpoint_id:
for channel in delta_channels:
result[channel] = self._load_delta_chain(
thread_id, checkpoint_ns, checkpoint_id, channel, cur
)
if type_tag != "empty":
result[k.decode()] = self.serde.loads_typed((type_tag, v))
return result
def _load_delta_chain(
def _resolve_delta_channels(
self,
config: RunnableConfig,
channel_values: dict[str, Any],
cur: Any,
) -> None:
for channel, value in list(channel_values.items()):
if isinstance(value, DeltaChannelSentinel):
channel_values[channel] = self._get_channel_writes_cur(
config["configurable"]["thread_id"],
config["configurable"].get("checkpoint_ns", ""),
config["configurable"]["checkpoint_id"],
channel,
cur,
)
def _get_channel_writes_cur(
self,
thread_id: str,
checkpoint_ns: str,
checkpoint_id: str,
channel: str,
cur: Any,
) -> DeltaChainValue:
"""Fetch the full delta chain for a channel in one recursive CTE query."""
) -> list[Any]:
"""Fetch writes for `channel` across the checkpoint ancestor chain, oldest→newest."""
cur.execute(
"""
WITH RECURSIVE chain AS (
SELECT
c.checkpoint_id,
c.parent_checkpoint_id,
cb.version,
cb.type,
cb.blob
FROM checkpoints c
JOIN checkpoint_blobs cb
ON cb.thread_id = %s
AND cb.checkpoint_ns = %s
AND cb.channel = %s
AND cb.version = (c.checkpoint->'channel_versions'->>%s)::text
WHERE c.thread_id = %s
AND c.checkpoint_ns = %s
AND c.checkpoint_id = %s
WITH RECURSIVE chain(cid, depth) AS (
SELECT parent_checkpoint_id, 0
FROM checkpoints
WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id = %s
UNION ALL
SELECT
c.checkpoint_id,
c.parent_checkpoint_id,
cb.version,
cb.type,
cb.blob
FROM chain prev
JOIN checkpoints c ON c.checkpoint_id = prev.parent_checkpoint_id
JOIN checkpoint_blobs cb
ON cb.thread_id = %s
AND cb.checkpoint_ns = %s
AND cb.channel = %s
AND cb.version = (c.checkpoint->'channel_versions'->>%s)::text
WHERE prev.parent_checkpoint_id IS NOT NULL
AND prev.type = 'delta'
SELECT c.parent_checkpoint_id, ch.depth + 1
FROM checkpoints c
JOIN chain ch ON c.checkpoint_id = ch.cid
WHERE ch.cid IS NOT NULL
)
SELECT DISTINCT ON (version) type, blob
FROM chain
ORDER BY version ASC
SELECT cw.type, cw.blob
FROM checkpoint_writes cw
JOIN chain ON cw.checkpoint_id = chain.cid
WHERE cw.thread_id = %s AND cw.checkpoint_ns = %s AND cw.channel = %s
ORDER BY chain.depth DESC, cw.task_id, cw.idx
""",
(
thread_id,
checkpoint_ns,
channel,
channel,
thread_id,
checkpoint_ns,
checkpoint_id,
thread_id,
checkpoint_ns,
channel,
channel,
),
)
rows = cur.fetchall()
base = None
deltas: list[list[Any]] = []
for row in rows:
blob = self.serde.loads_typed((row["type"], row["blob"]))
if isinstance(blob, DeltaValue):
deltas.append(blob.delta)
else:
base = blob
return DeltaChainValue(base=base, deltas=deltas)
return [
self.serde.loads_typed((row["type"], row["blob"])) for row in cur.fetchall()
]
def _dump_blobs(
self,