From b799b951382838391b6508024e03aef4fa379b6e Mon Sep 17 00:00:00 2001 From: Sydney Runkle Date: Tue, 21 Apr 2026 07:56:43 -0400 Subject: [PATCH] chore: rename DiffChannel/DiffDelta/DiffChainValue to Delta* across libs Renames the diff-channel types to DeltaChannel, DeltaValue, and DeltaChainValue for consistency with the settled naming convention. --- .../langgraph/checkpoint/postgres/__init__.py | 4 +- .../langgraph/checkpoint/postgres/aio.py | 4 +- .../langgraph/checkpoint/postgres/base.py | 2 +- libs/checkpoint-postgres/tests/test_async.py | 10 +- .../langgraph/checkpoint/base/__init__.py | 8 +- .../langgraph/checkpoint/memory/__init__.py | 8 +- .../langgraph/checkpoint/serde/jsonplus.py | 4 +- libs/checkpoint/tests/test_jsonplus.py | 8 +- libs/checkpoint/tests/test_memory.py | 22 +-- libs/langgraph/langgraph/channels/__init__.py | 4 +- libs/langgraph/langgraph/channels/base.py | 2 +- .../langgraph/channels/{diff.py => delta.py} | 40 ++-- libs/langgraph/tests/test_channels.py | 84 ++++---- ...ark.py => test_delta_channel_benchmark.py} | 26 +-- libs/langgraph/tests/test_pregel.py | 30 +-- libs/langgraph/tests/test_rehydrate_sweep.py | 185 ++++++++++++++++++ 16 files changed, 313 insertions(+), 128 deletions(-) rename libs/langgraph/langgraph/channels/{diff.py => delta.py} (83%) rename libs/langgraph/tests/{test_diff_channel_benchmark.py => test_delta_channel_benchmark.py} (85%) create mode 100644 libs/langgraph/tests/test_rehydrate_sweep.py diff --git a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py index c9b9d3d45..7c3bc3dcf 100644 --- a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py +++ b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py @@ -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 diff --git a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py index 8e07248bf..550c82033 100644 --- a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py +++ b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py @@ -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 diff --git a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py index 82f1f5b59..fe87758e8 100644 --- a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py +++ b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py @@ -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( diff --git a/libs/checkpoint-postgres/tests/test_async.py b/libs/checkpoint-postgres/tests/test_async.py index 099094f2a..4a9f08336 100644 --- a/libs/checkpoint-postgres/tests/test_async.py +++ b/libs/checkpoint-postgres/tests/test_async.py @@ -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"]) diff --git a/libs/checkpoint/langgraph/checkpoint/base/__init__.py b/libs/checkpoint/langgraph/checkpoint/base/__init__.py index 40a73bed5..cbbfe74ed 100644 --- a/libs/checkpoint/langgraph/checkpoint/base/__init__.py +++ b/libs/checkpoint/langgraph/checkpoint/base/__init__.py @@ -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 diff --git a/libs/checkpoint/langgraph/checkpoint/memory/__init__.py b/libs/checkpoint/langgraph/checkpoint/memory/__init__.py index 5fe5e0ca5..4cc98f9db 100644 --- a/libs/checkpoint/langgraph/checkpoint/memory/__init__.py +++ b/libs/checkpoint/langgraph/checkpoint/memory/__init__.py @@ -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 diff --git a/libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py b/libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py index 37647fd51..04c2e7f1d 100644 --- a/libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py +++ b/libs/checkpoint/langgraph/checkpoint/serde/jsonplus.py @@ -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): diff --git a/libs/checkpoint/tests/test_jsonplus.py b/libs/checkpoint/tests/test_jsonplus.py index 1695fd5ac..2360721cc 100644 --- a/libs/checkpoint/tests/test_jsonplus.py +++ b/libs/checkpoint/tests/test_jsonplus.py @@ -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" diff --git a/libs/checkpoint/tests/test_memory.py b/libs/checkpoint/tests/test_memory.py index 3830eb37d..0c2204db1 100644 --- a/libs/checkpoint/tests/test_memory.py +++ b/libs/checkpoint/tests/test_memory.py @@ -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"]] diff --git a/libs/langgraph/langgraph/channels/__init__.py b/libs/langgraph/langgraph/channels/__init__.py index 213fd6778..cc7fcc471 100644 --- a/libs/langgraph/langgraph/channels/__init__.py +++ b/libs/langgraph/langgraph/channels/__init__.py @@ -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 diff --git a/libs/langgraph/langgraph/channels/base.py b/libs/langgraph/langgraph/channels/base.py index 4ee061953..77ded1f9e 100644 --- a/libs/langgraph/langgraph/channels/base.py +++ b/libs/langgraph/langgraph/channels/base.py @@ -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 diff --git a/libs/langgraph/langgraph/channels/diff.py b/libs/langgraph/langgraph/channels/delta.py similarity index 83% rename from libs/langgraph/langgraph/channels/diff.py rename to libs/langgraph/langgraph/channels/delta.py index 09caeb3df..abc3de12b 100644 --- a/libs/langgraph/langgraph/channels/diff.py +++ b/libs/langgraph/langgraph/channels/delta.py @@ -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__ != "" @@ -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 diff --git a/libs/langgraph/tests/test_channels.py b/libs/langgraph/tests/test_channels.py index f73e12930..9d34890fa 100644 --- a/libs/langgraph/tests/test_channels.py +++ b/libs/langgraph/tests/test_channels.py @@ -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" diff --git a/libs/langgraph/tests/test_diff_channel_benchmark.py b/libs/langgraph/tests/test_delta_channel_benchmark.py similarity index 85% rename from libs/langgraph/tests/test_diff_channel_benchmark.py rename to libs/langgraph/tests/test_delta_channel_benchmark.py index c89b29e21..cbdf427ff 100644 --- a/libs/langgraph/tests/test_diff_channel_benchmark.py +++ b/libs/langgraph/tests/test_delta_channel_benchmark.py @@ -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}" ) diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index e975d02ab..9ca522bd8 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -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 diff --git a/libs/langgraph/tests/test_rehydrate_sweep.py b/libs/langgraph/tests/test_rehydrate_sweep.py new file mode 100644 index 000000000..5a72ed9b2 --- /dev/null +++ b/libs/langgraph/tests/test_rehydrate_sweep.py @@ -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)