WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-30 11:49:38 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor(channels): DeltaChannel takes typ as first arg, matching BinOpChannel

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Sydney Runkle
2026-04-30 14:49:05 -04:00
co-authored by Claude Sonnet 4.6
parent 41b2d00d46
commit f043ffc529
6 changed files with 59 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/langgraph/langgraph/channels/delta.py
+25 -18
View File
@@ -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:
libs/langgraph/langgraph/graph/state.py
+6 -23
View File
@@ -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]
libs/langgraph/tests/test_channels.py
+20 -16
View File
@@ -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".
libs/langgraph/tests/test_delta_channel_benchmark.py
+2 -2
View File
@@ -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').
libs/langgraph/tests/test_delta_channel_migration.py
+1 -1
View File
@@ -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)
libs/langgraph/tests/test_pregel.py
+5 -5
View File
@@ -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"])