WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-26 17:42:24 +02:00
Code Issues Packages Projects Releases Wiki Activity

chore: rename DiffChannel/DiffDelta/DiffChainValue to Delta* across libs

Browse Source 
Renames the diff-channel types to DeltaChannel, DeltaValue, and DeltaChainValue
for consistency with the settled naming convention.
This commit is contained in:
Sydney Runkle
2026-04-22 14:03:37 -04:00
parent ed9711fd33
commit b799b95138
16 changed files with 313 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py
+2 -2
View File
@@ -438,7 +438,7 @@ class PostgresSaver(BasePostgresSaver):
*,
cur: Any = None,
) -> dict[str, Any]:
from langgraph.checkpoint.base import DiffChainValue
from langgraph.checkpoint.base import DeltaChainValue
result: dict[str, Any] = {}
for channel, current_payload in diff_channel_payloads.items():
@@ -469,7 +469,7 @@ class PostgresSaver(BasePostgresSaver):
break
payloads.reverse()
result[channel] = DiffChainValue(
result[channel] = DeltaChainValue(
base=base, deltas=[p["d"] for p in payloads]
)
return result
libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py
+2 -2
View File
@@ -399,7 +399,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
*,
cur: Any,
) -> dict[str, Any]:
from langgraph.checkpoint.base import DiffChainValue
from langgraph.checkpoint.base import DeltaChainValue
result: dict[str, Any] = {}
for channel, current_payload in diff_channel_payloads.items():
@@ -430,7 +430,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
break
payloads.reverse()
result[channel] = DiffChainValue(
result[channel] = DeltaChainValue(
base=base, deltas=[p["d"] for p in payloads]
)
return result
libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py
+1 -1
View File
@@ -223,7 +223,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
*,
cur: Any = None,
) -> dict[str, Any]:
"""Override in sync/async subclasses. Resolves diff-chain blobs to DiffChainValue."""
"""Override in sync/async subclasses. Resolves diff-chain blobs to DeltaChainValue."""
raise NotImplementedError
def _dump_blobs(
libs/checkpoint-postgres/tests/test_async.py
+5 -5
View File
@@ -374,22 +374,22 @@ async def test_get_checkpoint_no_channel_values(
@pytest.mark.parametrize("saver_name", ["base", "pool", "pipe"])
async def test_diff_channel_chain_reconstruction(saver_name: str) -> None:
"""AsyncPostgresSaver reconstructs DiffChannel chain via point-lookup traversal."""
async def test_delta_channel_chain_reconstruction(saver_name: str) -> None:
"""AsyncPostgresSaver reconstructs DeltaChannel chain via point-lookup traversal."""
pytest.importorskip(
"langgraph.channels.diff", reason="langgraph core not installed"
"langgraph.channels.delta", reason="langgraph core not installed"
)
from typing import Annotated
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
from typing_extensions import TypedDict
class State(TypedDict):
messages: Annotated[list, DiffChannel(add_messages)]
messages: Annotated[list, DeltaChannel(add_messages)]
def respond(state: State) -> dict:
n = len(state["messages"])
libs/checkpoint/langgraph/checkpoint/base/__init__.py
+4 -4
View File
@@ -32,16 +32,16 @@ PendingWrite = tuple[str, str, Any]
@dataclasses.dataclass
class DiffDelta:
"""Returned by DiffChannel.checkpoint(). Represents one step's writes."""
class DeltaValue:
"""Returned by DeltaChannel.checkpoint(). Represents one step's writes."""
delta: list[Any]
prev_version: str | None # version of previous diff blob; None = chain root
@dataclasses.dataclass
class DiffChainValue:
"""Passed to DiffChannel.from_checkpoint(). Assembled by saver _load_blobs()."""
class DeltaChainValue:
"""Passed to DeltaChannel.from_checkpoint(). Assembled by saver _load_blobs()."""
base: list[Any] | None # starting accumulated value; None = start from empty
deltas: list[list[Any]] # per-step write-sets, ordered oldest → newest
libs/checkpoint/langgraph/checkpoint/memory/__init__.py
+4 -4
View File
@@ -123,7 +123,7 @@ class InMemorySaver(
def _load_blobs(
self, thread_id: str, checkpoint_ns: str, versions: ChannelVersions
) -> dict[str, Any]:
from langgraph.checkpoint.base import DiffChainValue
from langgraph.checkpoint.base import DeltaChainValue
channel_values: dict[str, Any] = {}
diff_channels: dict[str, str] = {}
@@ -146,7 +146,7 @@ class InMemorySaver(
while version is not None:
if version in visited:
logger.warning(
"DiffChannel chain cycle detected at version %r for channel %r; breaking",
"DeltaChannel chain cycle detected at version %r for channel %r; breaking",
version,
k,
)
@@ -155,7 +155,7 @@ class InMemorySaver(
kk = (thread_id, checkpoint_ns, k, version)
if kk not in self.blobs:
logger.warning(
"DiffChannel chain is broken: blob for channel %r version %r not found; "
"DeltaChannel chain is broken: blob for channel %r version %r not found; "
"partial history will be returned",
k,
version,
@@ -172,7 +172,7 @@ class InMemorySaver(
base = self.serde.loads_typed(vv)
break
chain_deltas.reverse()
channel_values[k] = DiffChainValue(base=base, deltas=chain_deltas)
channel_values[k] = DeltaChainValue(base=base, deltas=chain_deltas)
return channel_values
libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py
+2 -2
View File
@@ -65,9 +65,9 @@ def _warn_once(
def _is_diff_delta(obj: Any) -> bool:
from langgraph.checkpoint.base import DiffDelta # lazy import avoids circular dep
from langgraph.checkpoint.base import DeltaValue # lazy import avoids circular dep
return isinstance(obj, DiffDelta)
return isinstance(obj, DeltaValue)
class JsonPlusSerializer(SerializerProtocol):
libs/checkpoint/tests/test_jsonplus.py
+4 -4
View File
@@ -1000,11 +1000,11 @@ def test_msgpack_nested_pydantic_serializes_as_dict(
def test_diff_delta_serde_round_trip() -> None:
from langgraph.checkpoint.base import DiffDelta
from langgraph.checkpoint.base import DeltaValue
serde = JsonPlusSerializer()
prev = "00000000000000000000000000000001.1234567890123456"
delta = DiffDelta(
delta = DeltaValue(
delta=[HumanMessage(content="hello", id="msg-1")],
prev_version=prev,
)
@@ -1019,10 +1019,10 @@ def test_diff_delta_serde_round_trip() -> None:
def test_diff_delta_serde_root_blob() -> None:
from langgraph.checkpoint.base import DiffDelta
from langgraph.checkpoint.base import DeltaValue
serde = JsonPlusSerializer()
delta = DiffDelta(delta=[], prev_version=None)
delta = DeltaValue(delta=[], prev_version=None)
type_tag, blob = serde.dumps_typed(delta)
assert type_tag == "diff"
libs/checkpoint/tests/test_memory.py
+11 -11
View File
@@ -322,10 +322,10 @@ def test_memory_saver_with_allowlist_proxy_isolated() -> None:
assert direct.checkpoint["channel_values"]["foo"] == expected
class TestInMemorySaverDiffChannel:
def test_diff_channel_chain_reconstruction(self) -> None:
"""_load_blobs follows the diff chain and returns DiffChainValue."""
from langgraph.checkpoint.base import DiffChainValue, DiffDelta
class TestInMemorySaverDeltaChannel:
def test_delta_channel_chain_reconstruction(self) -> None:
"""_load_blobs follows the diff chain and returns DeltaChainValue."""
from langgraph.checkpoint.base import DeltaChainValue, DeltaValue
saver = InMemorySaver()
serde = JsonPlusSerializer()
@@ -337,8 +337,8 @@ class TestInMemorySaverDiffChannel:
v1 = "00000000000000000000000000000001.1234567890000000"
v2 = "00000000000000000000000000000002.1234567890000000"
delta1 = DiffDelta(delta=["msg1"], prev_version=None)
delta2 = DiffDelta(delta=["msg2"], prev_version=v1)
delta1 = DeltaValue(delta=["msg1"], prev_version=None)
delta2 = DeltaValue(delta=["msg2"], prev_version=v1)
saver.blobs[(thread_id, ns, "messages", v1)] = serde.dumps_typed(delta1)
saver.blobs[(thread_id, ns, "messages", v2)] = serde.dumps_typed(delta2)
@@ -347,13 +347,13 @@ class TestInMemorySaverDiffChannel:
assert "messages" in channel_values
result = channel_values["messages"]
assert isinstance(result, DiffChainValue)
assert isinstance(result, DeltaChainValue)
assert result.base is None
assert result.deltas == [["msg1"], ["msg2"]]
def test_diff_channel_mixed_old_and_new_blobs(self) -> None:
def test_delta_channel_mixed_old_and_new_blobs(self) -> None:
"""When chain hits an old non-diff blob, it becomes base."""
from langgraph.checkpoint.base import DiffChainValue, DiffDelta
from langgraph.checkpoint.base import DeltaChainValue, DeltaValue
saver = InMemorySaver()
serde = JsonPlusSerializer()
@@ -367,11 +367,11 @@ class TestInMemorySaverDiffChannel:
# Old-style full-list blob
saver.blobs[(thread_id, ns, "messages", v_old)] = serde.dumps_typed(["old_msg"])
# New diff blob chained to old
delta = DiffDelta(delta=["new_msg"], prev_version=v_old)
delta = DeltaValue(delta=["new_msg"], prev_version=v_old)
saver.blobs[(thread_id, ns, "messages", v_new)] = serde.dumps_typed(delta)
channel_values = saver._load_blobs(thread_id, ns, {"messages": v_new})
result = channel_values["messages"]
assert isinstance(result, DiffChainValue)
assert isinstance(result, DeltaChainValue)
assert result.base == ["old_msg"]
assert result.deltas == [["new_msg"]]
libs/langgraph/langgraph/channels/__init__.py
+2 -2
View File
@@ -1,7 +1,7 @@
from langgraph.channels.any_value import AnyValue
from langgraph.channels.base import BaseChannel
from langgraph.channels.binop import BinaryOperatorAggregate
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.channels.ephemeral_value import EphemeralValue
from langgraph.channels.last_value import LastValue, LastValueAfterFinish
from langgraph.channels.named_barrier_value import (
@@ -21,7 +21,7 @@ __all__ = (
"UntrackedValue",
"EphemeralValue",
"BinaryOperatorAggregate",
"DiffChannel",
"DeltaChannel",
"NamedBarrierValue",
"NamedBarrierValueAfterFinish",
# topics
libs/langgraph/langgraph/channels/base.py
+1 -1
View File
@@ -125,6 +125,6 @@ class BaseChannel(Generic[Value, Update, Checkpoint], ABC):
from_checkpoint() with the current channel version.
No-op by default. Override in channels that track their own version
for incremental checkpointing (e.g. DiffChannel).
for incremental checkpointing (e.g. DeltaChannel).
"""
pass
libs/langgraph/langgraph/channels/diff.py → libs/langgraph/langgraph/channels/delta.py
+20 -20
View File
@@ -4,7 +4,7 @@ import collections.abc
from collections.abc import Callable, Sequence
from typing import Any, Generic
from langgraph.checkpoint.base import DiffChainValue, DiffDelta
from langgraph.checkpoint.base import DeltaChainValue, DeltaValue
from typing_extensions import Self
from langgraph._internal._typing import MISSING
@@ -12,10 +12,10 @@ from langgraph.channels.base import BaseChannel, Value
from langgraph.channels.binop import _get_overwrite, _strip_extras
from langgraph.errors import EmptyChannelError
__all__ = ("DiffChannel",)
__all__ = ("DeltaChannel",)
class DiffChannel(Generic[Value], BaseChannel[list[Value], Value, DiffDelta]):
class DeltaChannel(Generic[Value], BaseChannel[list[Value], Value, DeltaValue]):
"""A channel that stores only per-step write deltas in checkpoints.
Reconstructs the full accumulated list at load time by replaying the
@@ -28,13 +28,13 @@ class DiffChannel(Generic[Value], BaseChannel[list[Value], Value, DiffDelta]):
Usage::
class State(TypedDict):
messages: Annotated[list[AnyMessage], DiffChannel(add_messages)]
messages: Annotated[list[AnyMessage], DeltaChannel(add_messages)]
"""
__slots__ = (
"value",
"operator",
"rehydrate_every",
"snapshot_every",
"_pending",
"_base_version",
"_overwritten",
@@ -46,7 +46,7 @@ class DiffChannel(Generic[Value], BaseChannel[list[Value], Value, DiffDelta]):
operator: Callable[[list[Value], Any], list[Value]],
typ: type = list,
*,
rehydrate_every: int | None = None,
snapshot_every: int | None = None,
) -> None:
typ = _strip_extras(typ)
if typ in (
@@ -56,7 +56,7 @@ class DiffChannel(Generic[Value], BaseChannel[list[Value], Value, DiffDelta]):
typ = list
super().__init__(typ)
self.operator = operator
self.rehydrate_every = rehydrate_every
self.snapshot_every = snapshot_every
try:
self.value: list[Value] = typ()
except Exception:
@@ -67,9 +67,9 @@ class DiffChannel(Generic[Value], BaseChannel[list[Value], Value, DiffDelta]):
self._steps_since_rehydrate: int = 0
def __eq__(self, other: object) -> bool:
if not isinstance(other, DiffChannel):
if not isinstance(other, DeltaChannel):
return False
if self.rehydrate_every != other.rehydrate_every:
if self.snapshot_every != other.snapshot_every:
return False
if (
self.operator.__name__ != "<lambda>"
@@ -87,7 +87,7 @@ class DiffChannel(Generic[Value], BaseChannel[list[Value], Value, DiffDelta]):
return self.typ | list[self.typ] # type: ignore[name-defined]
def copy(self) -> Self:
new = DiffChannel(self.operator, self.typ, rehydrate_every=self.rehydrate_every)
new = DeltaChannel(self.operator, self.typ, snapshot_every=self.snapshot_every)
new.key = self.key
new.value = self.value[:]
new._pending = self._pending[:]
@@ -97,11 +97,11 @@ class DiffChannel(Generic[Value], BaseChannel[list[Value], Value, DiffDelta]):
return new
def from_checkpoint(self, checkpoint: Any) -> Self:
new = DiffChannel(self.operator, self.typ, rehydrate_every=self.rehydrate_every)
new = DeltaChannel(self.operator, self.typ, snapshot_every=self.snapshot_every)
new.key = self.key
if checkpoint is MISSING:
new.value = []
elif isinstance(checkpoint, DiffChainValue):
elif isinstance(checkpoint, DeltaChainValue):
accumulated: list[Value] = list(checkpoint.base) if checkpoint.base else []
for step_writes in checkpoint.deltas:
for write in step_writes:
@@ -110,9 +110,9 @@ class DiffChannel(Generic[Value], BaseChannel[list[Value], Value, DiffDelta]):
# Seed the counter from actual chain depth so rehydration fires at
# the right time regardless of how many prior invocations there were.
new._steps_since_rehydrate = len(checkpoint.deltas)
elif isinstance(checkpoint, DiffDelta):
elif isinstance(checkpoint, DeltaValue):
raise ValueError(
"DiffChannel received a raw DiffDelta from the checkpoint saver. "
"DeltaChannel received a raw DeltaValue from the checkpoint saver. "
"Your saver does not support incremental channel storage. "
"Use InMemorySaver or PostgresSaver."
)
@@ -164,14 +164,14 @@ class DiffChannel(Generic[Value], BaseChannel[list[Value], Value, DiffDelta]):
def checkpoint(self) -> Any:
if (
self.rehydrate_every is not None
and self._steps_since_rehydrate >= self.rehydrate_every
self.snapshot_every is not None
and self._steps_since_rehydrate >= self.snapshot_every
):
# Emit a full snapshot to cap chain depth at rehydrate_every.
# Emit a full snapshot to cap chain depth at snapshot_every.
# The saver stores this as a plain (non-diff) blob, so future
# deltas will chain back to it and traversal depth resets to 1.
return list(self.value)
return DiffDelta(
return DeltaValue(
delta=self._pending[:],
prev_version=None if self._overwritten else self._base_version,
)
@@ -182,8 +182,8 @@ class DiffChannel(Generic[Value], BaseChannel[list[Value], Value, DiffDelta]):
# First call after from_checkpoint — anchor the base version
# without counting a step (the counter was seeded by from_checkpoint).
pass
elif self.rehydrate_every is not None:
if self._steps_since_rehydrate >= self.rehydrate_every:
elif self.snapshot_every is not None:
if self._steps_since_rehydrate >= self.snapshot_every:
self._steps_since_rehydrate = 0
else:
self._steps_since_rehydrate += 1
libs/langgraph/tests/test_channels.py
+42 -42
View File
@@ -119,20 +119,20 @@ def test_untracked_value() -> None:
new_channel.get()
def test_diff_channel_basic_two_steps() -> None:
def test_delta_channel_basic_two_steps() -> None:
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.checkpoint.base import DiffDelta
from langgraph.checkpoint.base import DeltaValue
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.message import add_messages
ch = DiffChannel(add_messages).from_checkpoint(MISSING)
ch = DeltaChannel(add_messages).from_checkpoint(MISSING)
ch.after_checkpoint(None)
# Step 1: one message added
ch.update([HumanMessage(content="hi", id="h1")])
d1 = ch.checkpoint()
assert isinstance(d1, DiffDelta)
assert isinstance(d1, DeltaValue)
assert len(d1.delta) == 1
assert d1.prev_version is None # first ever step
ch.after_checkpoint("v1")
@@ -150,13 +150,13 @@ def test_diff_channel_basic_two_steps() -> None:
assert ch.get()[1].content == "hello"
def test_diff_channel_after_checkpoint_no_op_when_unchanged() -> None:
def test_delta_channel_after_checkpoint_no_op_when_unchanged() -> None:
from langchain_core.messages import HumanMessage
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.message import add_messages
ch = DiffChannel(add_messages).from_checkpoint(MISSING)
ch = DeltaChannel(add_messages).from_checkpoint(MISSING)
ch.after_checkpoint(None)
ch.update([HumanMessage(content="hi", id="h1")])
ch.after_checkpoint("v1")
@@ -167,15 +167,15 @@ def test_diff_channel_after_checkpoint_no_op_when_unchanged() -> None:
assert ch._pending == []
def test_diff_channel_from_checkpoint_chain() -> None:
def test_delta_channel_from_checkpoint_chain() -> None:
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.checkpoint.base import DiffChainValue
from langgraph.checkpoint.base import DeltaChainValue
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.message import add_messages
spec = DiffChannel(add_messages)
chain = DiffChainValue(
spec = DeltaChannel(add_messages)
chain = DeltaChainValue(
base=None,
deltas=[
[HumanMessage(content="hi", id="h1")],
@@ -191,28 +191,28 @@ def test_diff_channel_from_checkpoint_chain() -> None:
assert msgs[2].content == "bye"
def test_diff_channel_from_checkpoint_backwards_compat() -> None:
def test_delta_channel_from_checkpoint_backwards_compat() -> None:
from langchain_core.messages import HumanMessage
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.message import add_messages
# Old BinaryOperatorAggregate checkpoint: plain list
spec = DiffChannel(add_messages)
spec = DeltaChannel(add_messages)
old_value = [HumanMessage(content="old", id="h1")]
ch = spec.from_checkpoint(old_value)
assert ch.get() == old_value
def test_diff_channel_overwrite_resets_chain() -> None:
def test_delta_channel_overwrite_resets_chain() -> None:
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.base import DiffDelta
from langgraph.checkpoint.base import DeltaValue
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.message import add_messages
from langgraph.types import Overwrite
ch = DiffChannel(add_messages).from_checkpoint(MISSING)
ch = DeltaChannel(add_messages).from_checkpoint(MISSING)
ch.after_checkpoint(None)
ch.update([HumanMessage(content="old", id="h1")])
ch.after_checkpoint("v1")
@@ -220,34 +220,34 @@ def test_diff_channel_overwrite_resets_chain() -> None:
# Overwrite should create a root blob (prev_version=None)
ch.update([Overwrite([HumanMessage(content="new", id="h2")])])
d = ch.checkpoint()
assert isinstance(d, DiffDelta)
assert isinstance(d, DeltaValue)
assert d.prev_version is None # chain root
assert len(d.delta) == 1
assert d.delta[0].content == "new"
def test_diff_channel_unsupported_saver_raises() -> None:
from langgraph.checkpoint.base import DiffDelta
def test_delta_channel_unsupported_saver_raises() -> None:
from langgraph.checkpoint.base import DeltaValue
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.message import add_messages
# If a saver returns a raw DiffDelta (unsupported), from_checkpoint raises
spec = DiffChannel(add_messages)
raw_delta = DiffDelta(delta=[], prev_version=None)
with pytest.raises(ValueError, match="DiffChannel received a raw DiffDelta"):
# If a saver returns a raw DeltaValue (unsupported), from_checkpoint raises
spec = DeltaChannel(add_messages)
raw_delta = DeltaValue(delta=[], prev_version=None)
with pytest.raises(ValueError, match="DeltaChannel received a raw DeltaValue"):
spec.from_checkpoint(raw_delta)
def test_diff_channel_remove_message_delta_and_replay() -> None:
def test_delta_channel_remove_message_delta_and_replay() -> None:
"""RemoveMessage stored in a delta must round-trip correctly through the chain."""
from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
from langgraph.checkpoint.base import DiffChainValue, DiffDelta
from langgraph.checkpoint.base import DeltaChainValue, DeltaValue
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.message import add_messages
spec = DiffChannel(add_messages)
spec = DeltaChannel(add_messages)
ch = spec.from_checkpoint(MISSING)
ch.after_checkpoint(None)
@@ -255,7 +255,7 @@ def test_diff_channel_remove_message_delta_and_replay() -> None:
ch.update([HumanMessage(content="hi", id="h1")])
ch.update([AIMessage(content="hello", id="a1")])
d1 = ch.checkpoint()
assert isinstance(d1, DiffDelta)
assert isinstance(d1, DeltaValue)
ch.after_checkpoint("v1")
assert ch.get() == [
HumanMessage(content="hi", id="h1"),
@@ -265,46 +265,46 @@ def test_diff_channel_remove_message_delta_and_replay() -> None:
# Step 2: remove the AI message
ch.update([RemoveMessage(id="a1")])
d2 = ch.checkpoint()
assert isinstance(d2, DiffDelta)
assert isinstance(d2, DeltaValue)
assert d2.prev_version == "v1"
assert any(isinstance(w, RemoveMessage) for w in d2.delta)
ch.after_checkpoint("v2")
assert ch.get() == [HumanMessage(content="hi", id="h1")]
# Replay the full chain from scratch — must reproduce the post-remove state
chain = DiffChainValue(base=None, deltas=[d1.delta, d2.delta])
chain = DeltaChainValue(base=None, deltas=[d1.delta, d2.delta])
ch2 = spec.from_checkpoint(chain)
assert ch2.get() == [HumanMessage(content="hi", id="h1")]
def test_diff_channel_update_by_id_delta_and_replay() -> None:
def test_delta_channel_update_by_id_delta_and_replay() -> None:
"""Updating a message by ID stored in a delta must round-trip correctly."""
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.base import DiffChainValue, DiffDelta
from langgraph.checkpoint.base import DeltaChainValue, DeltaValue
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.message import add_messages
spec = DiffChannel(add_messages)
spec = DeltaChannel(add_messages)
ch = spec.from_checkpoint(MISSING)
ch.after_checkpoint(None)
# Step 1: add a message
ch.update([HumanMessage(content="original", id="h1")])
d1 = ch.checkpoint()
assert isinstance(d1, DiffDelta)
assert isinstance(d1, DeltaValue)
ch.after_checkpoint("v1")
# Step 2: update the same message by ID
ch.update([HumanMessage(content="updated", id="h1")])
d2 = ch.checkpoint()
assert isinstance(d2, DiffDelta)
assert isinstance(d2, DeltaValue)
assert d2.prev_version == "v1"
ch.after_checkpoint("v2")
assert ch.get() == [HumanMessage(content="updated", id="h1")]
# Replay the full chain — must produce the updated message, not the original
chain = DiffChainValue(base=None, deltas=[d1.delta, d2.delta])
chain = DeltaChainValue(base=None, deltas=[d1.delta, d2.delta])
ch2 = spec.from_checkpoint(chain)
assert len(ch2.get()) == 1
assert ch2.get()[0].content == "updated"
libs/langgraph/tests/test_diff_channel_benchmark.py → libs/langgraph/tests/test_delta_channel_benchmark.py
+13 -13
View File
@@ -1,7 +1,7 @@
"""Benchmark: DiffChannel vs BinaryOperatorAggregate storage and time.
"""Benchmark: DeltaChannel vs BinaryOperatorAggregate storage and time.
Run directly: python tests/test_diff_channel_benchmark.py
Run via pytest: pytest tests/test_diff_channel_benchmark.py -s
Run directly: python tests/test_delta_channel_benchmark.py
Run via pytest: pytest tests/test_delta_channel_benchmark.py -s
"""
from __future__ import annotations
@@ -14,7 +14,7 @@ from langchain_core.messages import AIMessage, HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from typing_extensions import TypedDict
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph import END, StateGraph
from langgraph.graph.message import add_messages
@@ -31,12 +31,12 @@ class BinaryState(TypedDict):
class DiffState(TypedDict):
messages: Annotated[list, DiffChannel(add_messages)]
messages: Annotated[list, DeltaChannel(add_messages)]
class DiffRehydrateState(TypedDict):
messages: Annotated[
list, DiffChannel(add_messages, rehydrate_every=REHYDRATE_EVERY)
list, DeltaChannel(add_messages, snapshot_every=REHYDRATE_EVERY)
]
@@ -102,7 +102,7 @@ TURN_COUNTS = [10, 50, 100, 200, 500]
def run_benchmark() -> None:
print()
print(
"DiffChannel vs BinaryOperatorAggregate — checkpoint storage & time benchmark"
"DeltaChannel vs BinaryOperatorAggregate — checkpoint storage & time benchmark"
)
w = 100
print("=" * w)
@@ -133,7 +133,7 @@ def run_benchmark() -> None:
"time_ratio = diff_ms / bin_ms (higher = more overhead without rehydration)"
)
print(
f"rehy = DiffChannel(rehydrate_every={REHYDRATE_EVERY}) — caps chain depth"
f"rehy = DeltaChannel(snapshot_every={REHYDRATE_EVERY}) — caps chain depth"
)
print()
@@ -143,22 +143,22 @@ def run_benchmark() -> None:
# ---------------------------------------------------------------------------
def test_diff_channel_benchmark(capsys: Any) -> None:
"""Storage grows O(N²) for BinaryOperatorAggregate, O(N) for DiffChannel."""
def test_delta_channel_benchmark(capsys: Any) -> None:
"""Storage grows O(N²) for BinaryOperatorAggregate, O(N) for DeltaChannel."""
with capsys.disabled():
run_benchmark()
# Verify DiffChannel uses strictly less storage for 100+ turns.
# Verify DeltaChannel uses strictly less storage for 100+ turns.
for turns in [100, 500]:
_, b_bytes = _run_turns(turns, BinaryState)
_, d_bytes = _run_turns(turns, DiffState)
_, r_bytes = _run_turns(turns, DiffRehydrateState)
assert d_bytes < b_bytes, (
f"Expected DiffChannel to use less storage at {turns} turns, "
f"Expected DeltaChannel to use less storage at {turns} turns, "
f"got diff={d_bytes} binary={b_bytes}"
)
assert r_bytes < b_bytes, (
f"Expected DiffChannel(rehydrate) to use less storage at {turns} turns, "
f"Expected DeltaChannel(rehydrate) to use less storage at {turns} turns, "
f"got rehydrate={r_bytes} binary={b_bytes}"
)
libs/langgraph/tests/test_pregel.py
+15 -15
View File
@@ -9402,17 +9402,17 @@ def test_fork_does_not_apply_pending_writes(
assert result == {"value": 121}
async def test_diff_channel_end_to_end_inmemory() -> None:
"""Full graph run: DiffChannel accumulates correctly across multiple turns."""
async def test_delta_channel_end_to_end_inmemory() -> None:
"""Full graph run: DeltaChannel accumulates correctly across multiple turns."""
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, DiffChannel(add_messages)]
messages: Annotated[list, DeltaChannel(add_messages)]
def respond(state: State) -> dict:
n = len(state["messages"])
@@ -9444,17 +9444,17 @@ async def test_diff_channel_end_to_end_inmemory() -> None:
assert msgs[5].content == "reply-5"
async def test_diff_channel_time_travel() -> None:
async def test_delta_channel_time_travel() -> None:
"""Time-travel back to turn-1 checkpoint and resume; continuation must not include turn-2 deltas."""
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, DiffChannel(add_messages)]
messages: Annotated[list, DeltaChannel(add_messages)]
counter = {"n": 0}
@@ -9502,17 +9502,17 @@ async def test_diff_channel_time_travel() -> None:
assert msgs[2].content == "h3"
async def test_diff_channel_remove_message_end_to_end() -> None:
"""RemoveMessage inside a DiffChannel graph must persist and reload correctly."""
async def test_delta_channel_remove_message_end_to_end() -> None:
"""RemoveMessage inside a DeltaChannel graph must persist and reload correctly."""
from langchain_core.messages import AIMessage, HumanMessage, RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, DiffChannel(add_messages)]
messages: Annotated[list, DeltaChannel(add_messages)]
def respond(state: State) -> dict:
return {"messages": [AIMessage(content="reply", id="ai-1")]}
@@ -9549,17 +9549,17 @@ async def test_diff_channel_remove_message_end_to_end() -> None:
)
async def test_diff_channel_update_by_id_end_to_end() -> None:
"""Updating a message by ID via DiffChannel must persist and reload correctly."""
async def test_delta_channel_update_by_id_end_to_end() -> None:
"""Updating a message by ID via DeltaChannel must persist and reload correctly."""
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.channels.diff import DiffChannel
from langgraph.channels.delta import DeltaChannel
from langgraph.graph import START, StateGraph
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, DiffChannel(add_messages)]
messages: Annotated[list, DeltaChannel(add_messages)]
def update_msg(state: State) -> dict:
# re-send h1 with updated content
libs/langgraph/tests/test_rehydrate_sweep.py
+185
View File
@@ -0,0 +1,185 @@
"""Sweep snapshot_every values to find the storage vs. time-travel tradeoff.
Run directly: python tests/test_rehydrate_sweep.py
Run via pytest: pytest tests/test_rehydrate_sweep.py -s
"""
from __future__ import annotations
import sys
import time
from typing import Annotated, Any
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from typing_extensions import TypedDict
from langgraph.channels.delta import DeltaChannel
from langgraph.graph import END, StateGraph
from langgraph.graph.message import add_messages
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
REHYDRATE_SWEEP = [5, 10, 25, 50, 100, None] # None = no rehydration (pure diff)
TURN_COUNTS = [50, 100, 250, 500]
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_state(snapshot_every: int | None) -> type:
channel = DeltaChannel(add_messages, snapshot_every=snapshot_every)
return TypedDict("S", {"messages": Annotated[list, channel]})
def _make_graph(state_cls: type) -> Any:
def human_node(state: Any) -> dict:
return {}
def ai_node(state: Any) -> dict:
last = state["messages"][-1]
return {"messages": [AIMessage(content=f"reply-to-{last.id}")]}
g = StateGraph(state_cls)
g.add_node("human", human_node)
g.add_node("ai", ai_node)
g.add_edge("human", "ai")
g.add_edge("ai", END)
g.set_entry_point("human")
return g.compile(checkpointer=MemorySaver())
def _total_blob_bytes(saver: MemorySaver) -> int:
total = 0
for (_, _, _, _), (type_tag, blob) in saver.blobs.items():
if blob is not None:
total += len(blob)
return total
def _measure_time_travel_ms(graph: Any, config: dict) -> float:
"""Time how long it takes to get state at the very first checkpoint (worst case)."""
history = list(graph.get_state_history(config))
if not history:
return 0.0
oldest = history[-1]
t0 = time.perf_counter()
graph.get_state(oldest.config)
return (time.perf_counter() - t0) * 1000
def _run(n_turns: int, snapshot_every: int | None) -> tuple[float, int, float]:
"""Returns (write_ms, blob_bytes, time_travel_ms)."""
state_cls = _make_state(snapshot_every)
graph = _make_graph(state_cls)
saver: MemorySaver = graph.checkpointer # type: ignore[assignment]
config = {"configurable": {"thread_id": "sweep"}}
t0 = time.perf_counter()
for i in range(n_turns):
graph.invoke(
{"messages": [HumanMessage(content=f"msg-{i}", id=f"h{i}")]}, config
)
write_ms = (time.perf_counter() - t0) * 1000
blob_bytes = _total_blob_bytes(saver)
tt_ms = _measure_time_travel_ms(graph, config)
return write_ms, blob_bytes, tt_ms
# ---------------------------------------------------------------------------
# ASCII sparkline
# ---------------------------------------------------------------------------
def _sparkline(values: list[float], width: int = 20) -> str:
bars = " ▁▂▃▄▅▆▇█"
lo, hi = min(values), max(values)
span = hi - lo or 1
chars = [bars[round((v - lo) / span * (len(bars) - 1))] for v in values]
return "".join(chars).ljust(width)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def run_sweep() -> None:
label = {v: (str(v) if v is not None else "None(∞)") for v in REHYDRATE_SWEEP}
print()
print("snapshot_every sweep — storage vs time-travel cost")
print("=" * 90)
for turns in TURN_COUNTS:
print(f"\n--- {turns} turns ---")
col_w = 12
header = (
f"{'snapshot_every':>18} "
f"{'blob_bytes':>{col_w}} "
f"{'write_ms':>{col_w}} "
f"{'time_travel_ms':>{col_w}}"
)
print(header)
print("-" * 60)
tt_vals: list[float] = []
byte_vals: list[int] = []
write_vals: list[float] = []
rows: list[tuple] = []
for rv in REHYDRATE_SWEEP:
write_ms, blob_bytes, tt_ms = _run(turns, rv)
rows.append((rv, blob_bytes, write_ms, tt_ms))
byte_vals.append(blob_bytes)
write_vals.append(write_ms)
tt_vals.append(tt_ms)
for rv, blob_bytes, write_ms, tt_ms in rows:
print(
f"{label[rv]:>18} "
f"{blob_bytes:>{col_w},} "
f"{write_ms:>{col_w}.1f} "
f"{tt_ms:>{col_w}.2f}"
)
print()
print(
f" bytes spark: [{_sparkline(byte_vals)}] "
f"lo={min(byte_vals):,} hi={max(byte_vals):,}"
)
print(
f" time-travel spark: [{_sparkline(tt_vals)}] "
f"lo={min(tt_vals):.2f}ms hi={max(tt_vals):.2f}ms"
)
print(
f" write spark: [{_sparkline(write_vals)}] "
f"lo={min(write_vals):.1f}ms hi={max(write_vals):.1f}ms"
)
print()
print("=" * 90)
print(
"snapshot_every=None means pure diff (no snapshots) — "
"lowest storage, highest time-travel cost."
)
print(
"Lower snapshot_every = more frequent full snapshots = "
"faster time-travel, more storage."
)
print()
def test_rehydrate_sweep(capsys: Any) -> None:
with capsys.disabled():
run_sweep()
if __name__ == "__main__":
run_sweep()
sys.exit(0)