From f043ffc529ecf35e48fa6d0a46d20a0212322e3b Mon Sep 17 00:00:00 2001 From: Sydney Runkle Date: Wed, 29 Apr 2026 09:42:45 -0400 Subject: [PATCH] refactor(channels): DeltaChannel takes typ as first arg, matching BinOpChannel MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously DeltaChannel.__init__ hardcoded typ=list and _is_field_channel patched item.typ/item.value after construction. This mirrors BinaryOperatorAggregate: - DeltaChannel(typ, operator, *, snapshot_frequency=None) — typ is now a required first argument; __init__ strips abstract/parameterized types to their concrete counterparts (same logic as BinaryOperatorAggregate) - _is_field_channel reconstructs the channel via its constructor instead of patching typ and value externally - copy() and from_checkpoint() use self.__class__(self.typ, self.operator, ...) — no post-construction attribute hacking needed - _empty() helper removed; self.typ() is always a concrete callable - All call sites updated: DeltaChannel(list, op), DeltaChannel(dict, op), etc. Co-Authored-By: Claude Sonnet 4.6 (1M context) --- libs/langgraph/langgraph/channels/delta.py | 43 +++++++++++-------- libs/langgraph/langgraph/graph/state.py | 29 +++---------- libs/langgraph/tests/test_channels.py | 36 +++++++++------- .../tests/test_delta_channel_benchmark.py | 4 +- .../tests/test_delta_channel_migration.py | 2 +- libs/langgraph/tests/test_pregel.py | 10 ++--- 6 files changed, 59 insertions(+), 65 deletions(-) diff --git a/libs/langgraph/langgraph/channels/delta.py b/libs/langgraph/langgraph/channels/delta.py index 3c73dd010..f109e32c2 100644 --- a/libs/langgraph/langgraph/channels/delta.py +++ b/libs/langgraph/langgraph/channels/delta.py @@ -1,5 +1,6 @@ from __future__ import annotations +import collections.abc import copy as _copy from collections.abc import Callable, Sequence from typing import Any, Generic @@ -10,7 +11,7 @@ from typing_extensions import Self from langgraph._internal._typing import MISSING from langgraph.channels.base import BaseChannel, Value -from langgraph.channels.binop import _get_overwrite, _operators_equal +from langgraph.channels.binop import _get_overwrite, _operators_equal, _strip_extras from langgraph.errors import ( EmptyChannelError, ErrorCode, @@ -21,13 +22,6 @@ from langgraph.errors import ( __all__ = ("DeltaChannel",) -def _empty(typ: Any) -> Any: - try: - return typ() - except Exception: - return [] - - class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]): """Fold-reducer channel with configurable snapshot cadence. @@ -41,6 +35,7 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]): length. Parameters: + typ: The value type (e.g. `list`, `dict`). operator: Binary reducer `(Value, Value) -> Value`. snapshot_frequency: Every Nth pregel step writes a snapshot blob. `None` (default) = pure delta, never snapshot. @@ -51,13 +46,23 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]): def __init__( self, + typ: type[Value], operator: Callable[[Any, Any], Any], *, snapshot_frequency: int | None = None, ) -> None: - super().__init__(list) + super().__init__(typ) self.operator = operator self.snapshot_frequency = snapshot_frequency + # Normalize abstract / parameterized types to their concrete counterparts. + typ = _strip_extras(typ) + if typ in (collections.abc.Sequence, collections.abc.MutableSequence): + typ = list + if typ in (collections.abc.Set, collections.abc.MutableSet): + typ = set + if typ in (collections.abc.Mapping, collections.abc.MutableMapping): + typ = dict + self.typ = typ self.value: Any = MISSING def __eq__(self, other: object) -> bool: @@ -84,9 +89,10 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]): ) def copy(self) -> Self: - new = self.__class__(self.operator, snapshot_frequency=self.snapshot_frequency) - new.typ = self.typ # typ may differ from list when set via Annotated injection - new.key = self.key # key is injected externally by the graph builder + new = self.__class__( + self.typ, self.operator, snapshot_frequency=self.snapshot_frequency + ) + new.key = self.key new.value = self.value if self.value is MISSING else _copy.copy(self.value) return new @@ -96,9 +102,9 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]): return ( _copy.copy(overwrite_value) if overwrite_value is not None - else _empty(self.typ) + else self.typ() ) - base = _empty(self.typ) if value is MISSING else value + base = self.typ() if value is MISSING else value return self.operator(base, write) def from_checkpoint(self, checkpoint: Any) -> Self: @@ -109,11 +115,12 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]): * `_DeltaSnapshot(value)`: restore value directly from snapshot. * plain value (migration from old BinOp blobs): use directly. """ - new = self.__class__(self.operator, snapshot_frequency=self.snapshot_frequency) - new.typ = self.typ # typ may differ from list when set via Annotated injection - new.key = self.key # key is injected externally by the graph builder + new = self.__class__( + self.typ, self.operator, snapshot_frequency=self.snapshot_frequency + ) + new.key = self.key if checkpoint is MISSING or checkpoint is DELTA_SENTINEL: - new.value = _empty(new.typ) + new.value = self.typ() elif isinstance(checkpoint, _DeltaSnapshot): new.value = checkpoint.value else: diff --git a/libs/langgraph/langgraph/graph/state.py b/libs/langgraph/langgraph/graph/state.py index 78cdbb10a..51f887317 100644 --- a/libs/langgraph/langgraph/graph/state.py +++ b/libs/langgraph/langgraph/graph/state.py @@ -1,6 +1,5 @@ from __future__ import annotations -import collections.abc import inspect import logging import typing @@ -48,7 +47,7 @@ from langgraph._internal._pydantic import create_model from langgraph._internal._runnable import coerce_to_runnable from langgraph._internal._typing import EMPTY_SEQ, MISSING, DeprecatedKwargs from langgraph.channels.base import BaseChannel -from langgraph.channels.binop import BinaryOperatorAggregate, _strip_extras +from langgraph.channels.binop import BinaryOperatorAggregate from langgraph.channels.delta import DeltaChannel from langgraph.channels.ephemeral_value import EphemeralValue from langgraph.channels.last_value import LastValue, LastValueAfterFinish @@ -1679,27 +1678,11 @@ def _is_field_channel(typ: type[Any]) -> BaseChannel | None: NotRequired, ): origin = origin.__args__[0] - outer = _strip_extras(origin) - if outer in ( - collections.abc.Sequence, - collections.abc.MutableSequence, - ): - outer = list - elif outer in ( - collections.abc.Mapping, - collections.abc.MutableMapping, - ): - outer = dict - elif outer in ( - collections.abc.Set, - collections.abc.MutableSet, - ): - outer = set - item.typ = outer - try: - item.value = outer() - except Exception: - item.value = [] + item = item.__class__( + origin, + item.operator, + snapshot_frequency=item.snapshot_frequency, + ) return item elif isclass(item) and issubclass(item, BaseChannel): # ex, Annotated[int, EphemeralValue, SomeOtherAnnotation] diff --git a/libs/langgraph/tests/test_channels.py b/libs/langgraph/tests/test_channels.py index ab86a4174..562a45001 100644 --- a/libs/langgraph/tests/test_channels.py +++ b/libs/langgraph/tests/test_channels.py @@ -129,7 +129,7 @@ def test_delta_channel_basic_two_steps() -> None: from langgraph.graph.message import add_messages - ch = DeltaChannel(add_messages).from_checkpoint(MISSING) + ch = DeltaChannel(list, add_messages).from_checkpoint(MISSING) # Step 1: one message added ch.update([HumanMessage(content="hi", id="h1")]) @@ -153,7 +153,7 @@ def test_delta_channel_from_checkpoint_writes_list() -> None: from langgraph.graph.message import add_messages - spec = DeltaChannel(add_messages) + spec = DeltaChannel(list, add_messages) ch = spec.from_checkpoint(DELTA_SENTINEL) ch.replay_writes( [ @@ -175,7 +175,7 @@ def test_delta_channel_from_checkpoint_backwards_compat() -> None: from langgraph.graph.message import add_messages # Old BinaryOperatorAggregate checkpoint: plain list treated as backward compat - spec = DeltaChannel(add_messages) + spec = DeltaChannel(list, add_messages) old_value = [HumanMessage(content="old", id="h1")] ch = spec.from_checkpoint(old_value) assert ch.get() == old_value @@ -188,7 +188,7 @@ def test_delta_channel_overwrite() -> None: from langgraph.graph.message import add_messages from langgraph.types import Overwrite - ch = DeltaChannel(add_messages).from_checkpoint(MISSING) + ch = DeltaChannel(list, add_messages).from_checkpoint(MISSING) ch.update([HumanMessage(content="old", id="h1")]) ch.update([Overwrite([HumanMessage(content="new", id="h2")])]) @@ -205,7 +205,7 @@ def test_delta_channel_remove_message_and_replay() -> None: from langgraph.graph.message import add_messages - spec = DeltaChannel(add_messages) + spec = DeltaChannel(list, add_messages) ch = spec.from_checkpoint(MISSING) # Step 1: add two messages @@ -238,7 +238,7 @@ def test_delta_channel_update_by_id_and_replay() -> None: from langgraph.graph.message import add_messages - spec = DeltaChannel(add_messages) + spec = DeltaChannel(list, add_messages) ch = spec.from_checkpoint(MISSING) # Step 1: add a message @@ -266,7 +266,7 @@ def test_delta_channel_checkpoint_returns_sentinel() -> None: from langgraph.graph.message import add_messages - ch = DeltaChannel(add_messages).from_checkpoint(MISSING) + ch = DeltaChannel(list, add_messages).from_checkpoint(MISSING) assert ch.checkpoint() is DELTA_SENTINEL from langchain_core.messages import HumanMessage @@ -294,7 +294,9 @@ def test_delta_channel_snapshot_step_based() -> None: # snapshot_frequency=5: snapshot every 5 pregel steps class State(TypedDict): - messages: Annotated[list, DeltaChannel(add_messages, snapshot_frequency=5)] + messages: Annotated[ + list, DeltaChannel(list, add_messages, snapshot_frequency=5) + ] other: str def node_a(state: State) -> dict: @@ -350,7 +352,9 @@ def test_delta_channel_snapshot_fires_even_when_not_written() -> None: from langgraph.graph.message import add_messages class State(TypedDict): - messages: Annotated[list, DeltaChannel(add_messages, snapshot_frequency=3)] + messages: Annotated[ + list, DeltaChannel(list, add_messages, snapshot_frequency=3) + ] tick: int def writer(state: State) -> dict: @@ -405,7 +409,7 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None: from langgraph.graph.message import add_messages class State(TypedDict): - messages: Annotated[list, DeltaChannel(add_messages)] + messages: Annotated[list, DeltaChannel(list, add_messages)] n = {"v": 0} @@ -447,7 +451,7 @@ def _delta_channel_with_type(operator, typ): from langgraph.channels.delta import DeltaChannel from langgraph.graph.state import _get_channel - return _get_channel("_test", Annotated[typ, DeltaChannel(operator)]) + return _get_channel("_test", Annotated[typ, DeltaChannel(typ, operator)]) def test_delta_channel_dict_reducer_fresh_channel() -> None: @@ -574,7 +578,7 @@ def test_delta_channel_dict_reducer_with_notrequired_annotation() -> None: return dict(right) return {**left, **right} - annotation = Annotated[NotRequired[dict[str, int]], DeltaChannel(merge_dicts)] + annotation = Annotated[NotRequired[dict[str, int]], DeltaChannel(dict, merge_dicts)] ch = _get_channel("files", annotation).from_checkpoint(MISSING) assert ch.get() == {} ch.update([{"a": 1}]) @@ -604,7 +608,7 @@ def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None: return result class State(TypedDict): - files: Annotated[dict[str, str], DeltaChannel(merge_files)] + files: Annotated[dict[str, str], DeltaChannel(dict, merge_files)] turn = {"v": 0} @@ -674,7 +678,7 @@ def test_delta_channel_from_checkpoint_honors_seed() -> None: a pre-DeltaChannel blob it passes it as `seed` so replay reconstructs the post-migration state correctly rather than replaying from empty. """ - spec = DeltaChannel(add_messages) + spec = DeltaChannel(list, add_messages) seed = [HumanMessage(content="pre-delta", id="p1")] ch = spec.from_checkpoint(seed) ch.replay_writes( @@ -690,7 +694,7 @@ def test_delta_channel_from_checkpoint_honors_seed() -> None: def test_delta_channel_from_checkpoint_seed_without_writes() -> None: """Reconstruction at a pre-delta ancestor with no newer deltas returns just the seed — the saver's terminator fired immediately.""" - spec = DeltaChannel(add_messages) + spec = DeltaChannel(list, add_messages) seed = [HumanMessage(content="only-snap", id="s1")] ch = spec.from_checkpoint(seed) ch.replay_writes([]) @@ -707,7 +711,7 @@ def test_delta_channel_from_checkpoint_seed_none_is_distinct_from_sentinel() -> def replace(left, right): return right - spec = DeltaChannel(replace) + spec = DeltaChannel(list, replace) ch = spec.from_checkpoint(None) ch.replay_writes([("t0", "x", "after")]) # Reducer replaces; seed=None → first write produces "after". diff --git a/libs/langgraph/tests/test_delta_channel_benchmark.py b/libs/langgraph/tests/test_delta_channel_benchmark.py index 7e42c7da2..41bc68c25 100644 --- a/libs/langgraph/tests/test_delta_channel_benchmark.py +++ b/libs/langgraph/tests/test_delta_channel_benchmark.py @@ -114,12 +114,12 @@ class BinaryState(TypedDict): class DeltaState(TypedDict): - messages: Annotated[list, DeltaChannel(add_messages)] + messages: Annotated[list, DeltaChannel(list, add_messages)] def _make_delta_state(snapshot_frequency: int | float) -> type: """Create a TypedDict with DeltaChannel at the given snapshot_frequency.""" - channel = DeltaChannel(add_messages, snapshot_frequency=snapshot_frequency) + channel = DeltaChannel(list, add_messages, snapshot_frequency=snapshot_frequency) # Use the functional TypedDict form so the Annotated type is stored as an # already-evaluated object rather than a forward-reference string (which # would fail when get_type_hints tries to resolve 'snapshot_frequency'). diff --git a/libs/langgraph/tests/test_delta_channel_migration.py b/libs/langgraph/tests/test_delta_channel_migration.py index 72286522f..e74a713da 100644 --- a/libs/langgraph/tests/test_delta_channel_migration.py +++ b/libs/langgraph/tests/test_delta_channel_migration.py @@ -85,7 +85,7 @@ def _binop_graph(checkpointer: Any) -> Any: def _delta_graph(checkpointer: Any) -> Any: class DeltaState(TypedDict): - items: Annotated[list, DeltaChannel(operator.add)] + items: Annotated[list, DeltaChannel(list, operator.add)] return ( StateGraph(DeltaState) diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index 52057abe5..5ccb347ab 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -9412,7 +9412,7 @@ async def test_delta_channel_end_to_end_inmemory() -> None: from langgraph.graph.message import add_messages class State(TypedDict): - messages: Annotated[list, DeltaChannel(add_messages)] + messages: Annotated[list, DeltaChannel(list, add_messages)] def respond(state: State) -> dict: n = len(state["messages"]) @@ -9453,7 +9453,7 @@ async def test_delta_channel_time_travel() -> None: from langgraph.graph.message import add_messages class State(TypedDict): - messages: Annotated[list, DeltaChannel(add_messages)] + messages: Annotated[list, DeltaChannel(list, add_messages)] counter = {"n": 0} @@ -9510,7 +9510,7 @@ async def test_delta_channel_remove_message_end_to_end() -> None: from langgraph.graph.message import add_messages class State(TypedDict): - messages: Annotated[list, DeltaChannel(add_messages)] + messages: Annotated[list, DeltaChannel(list, add_messages)] def respond(state: State) -> dict: return {"messages": [AIMessage(content="reply", id="ai-1")]} @@ -9556,7 +9556,7 @@ async def test_delta_channel_update_by_id_end_to_end() -> None: from langgraph.graph.message import add_messages class State(TypedDict): - messages: Annotated[list, DeltaChannel(add_messages)] + messages: Annotated[list, DeltaChannel(list, add_messages)] def update_msg(state: State) -> dict: # re-send h1 with updated content @@ -9604,7 +9604,7 @@ async def test_delta_channel_write_flushed_before_put() -> None: from langgraph.graph.message import add_messages class State(TypedDict): - messages: Annotated[list, DeltaChannel(add_messages)] + messages: Annotated[list, DeltaChannel(list, add_messages)] def respond(state: State) -> dict: i = len(state["messages"])