diff --git a/libs/langgraph/langgraph/channels/delta.py b/libs/langgraph/langgraph/channels/delta.py index d8f2a67f9..07618668b 100644 --- a/libs/langgraph/langgraph/channels/delta.py +++ b/libs/langgraph/langgraph/channels/delta.py @@ -1,6 +1,5 @@ from __future__ import annotations -import collections.abc from collections.abc import Callable, Sequence from typing import Any, Generic @@ -9,7 +8,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, _strip_extras +from langgraph.channels.binop import _get_overwrite from langgraph.errors import EmptyChannelError __all__ = ("DeltaChannel",) @@ -39,6 +38,8 @@ class DeltaChannel(Generic[Value], BaseChannel[list[Value], Value, DeltaValue]): messages: Annotated[list[AnyMessage], DeltaChannel(add_messages)] # Cap reconstruction depth (recommended for SQLite / MongoDB savers): messages: Annotated[list[AnyMessage], DeltaChannel(add_messages, snapshot_every=50)] + # Dict-type reducer (type inferred from the Annotated outer type): + files: Annotated[dict, DeltaChannel(merge_files)] """ __slots__ = ( @@ -54,23 +55,13 @@ class DeltaChannel(Generic[Value], BaseChannel[list[Value], Value, DeltaValue]): def __init__( self, operator: Callable[[list[Value], Any], list[Value]], - typ: type = list, *, snapshot_every: int | None = None, ) -> None: - typ = _strip_extras(typ) - if typ in ( - collections.abc.Sequence, - collections.abc.MutableSequence, - ): - typ = list - super().__init__(typ) + super().__init__(list) self.operator = operator self.snapshot_every = snapshot_every - try: - self.value: list[Value] = typ() - except Exception: - self.value = [] + self.value: list[Value] = [] self._pending: list[Any] = [] self._base_version: str | None = None self._overwritten: bool = False @@ -97,7 +88,8 @@ class DeltaChannel(Generic[Value], BaseChannel[list[Value], Value, DeltaValue]): return self.typ | list[self.typ] # type: ignore[name-defined] def copy(self) -> Self: - new = DeltaChannel(self.operator, self.typ, snapshot_every=self.snapshot_every) + new = DeltaChannel(self.operator, snapshot_every=self.snapshot_every) + new.typ = self.typ new.key = self.key new.value = self.value if self.value is MISSING else self.value.copy() new._pending = self._pending[:] @@ -107,10 +99,14 @@ class DeltaChannel(Generic[Value], BaseChannel[list[Value], Value, DeltaValue]): return new def from_checkpoint(self, checkpoint: Any) -> Self: - new = DeltaChannel(self.operator, self.typ, snapshot_every=self.snapshot_every) + new = DeltaChannel(self.operator, snapshot_every=self.snapshot_every) + new.typ = self.typ new.key = self.key if checkpoint is MISSING: - pass + try: + new.value = new.typ() + except Exception: + new.value = [] elif isinstance(checkpoint, DeltaChainValue): accumulated: list[Value] = ( checkpoint.base if checkpoint.base is not None else new.typ() diff --git a/libs/langgraph/langgraph/graph/state.py b/libs/langgraph/langgraph/graph/state.py index 9f8dd26c7..d251973fa 100644 --- a/libs/langgraph/langgraph/graph/state.py +++ b/libs/langgraph/langgraph/graph/state.py @@ -1,5 +1,6 @@ from __future__ import annotations +import collections.abc import inspect import logging import typing @@ -47,7 +48,8 @@ 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 +from langgraph.channels.binop import BinaryOperatorAggregate, _strip_extras +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 ( @@ -1668,6 +1670,18 @@ def _is_field_channel(typ: type[Any]) -> BaseChannel | None: # Search through all annotated medata to find channel annotations for item in meta: if isinstance(item, BaseChannel): + if isinstance(item, DeltaChannel) and hasattr(typ, "__origin__"): + outer = _strip_extras(typ.__origin__) + if outer in ( + collections.abc.Sequence, + collections.abc.MutableSequence, + ): + outer = list + item.typ = outer + try: + item.value = outer() + except Exception: + item.value = [] 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 e01d3acb0..74f9d5bf8 100644 --- a/libs/langgraph/tests/test_channels.py +++ b/libs/langgraph/tests/test_channels.py @@ -427,14 +427,23 @@ def test_delta_channel_inmemory_saver_assembles_chain() -> None: assert len(state.values["messages"]) == 4 # 2 human + 2 AI +def _delta_channel_with_type(operator, typ): + """Build a DeltaChannel with an explicit type via the Annotated injection path.""" + from typing import Annotated + + from langgraph.channels.delta import DeltaChannel + from langgraph.graph.state import _get_channel + + return _get_channel("_test", Annotated[typ, DeltaChannel(operator)]) + + def test_delta_channel_dict_reducer_fresh_channel() -> None: """DeltaChannel with a dict reducer starts as empty dict on MISSING checkpoint.""" - from langgraph.channels.delta import DeltaChannel def merge_dicts(left: dict, right: dict) -> dict: return {**left, **right} - ch = DeltaChannel(merge_dicts, dict).from_checkpoint(MISSING) + ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING) # Should be available (not raise EmptyChannelError) and start empty assert ch.is_available() assert ch.get() == {} @@ -444,12 +453,10 @@ def test_delta_channel_dict_reducer_basic_updates() -> None: """DeltaChannel with a dict reducer accumulates key/value pairs across steps.""" from langgraph.checkpoint.base import DeltaValue - from langgraph.channels.delta import DeltaChannel - def merge_dicts(left: dict, right: dict) -> dict: return {**left, **right} - ch = DeltaChannel(merge_dicts, dict).from_checkpoint(MISSING) + ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING) ch.after_checkpoint(None) ch.update([{"a": 1}]) @@ -470,12 +477,10 @@ def test_delta_channel_dict_reducer_chain_reconstruction() -> None: """DeltaChainValue replays correctly through a dict merge reducer.""" from langgraph.checkpoint.base import DeltaChainValue - from langgraph.channels.delta import DeltaChannel - def merge_dicts(left: dict, right: dict) -> dict: return {**left, **right} - spec = DeltaChannel(merge_dicts, dict) + spec = _delta_channel_with_type(merge_dicts, dict) chain = DeltaChainValue( base={"a": 1}, deltas=[[{"b": 2}], [{"c": 3}]], @@ -489,8 +494,6 @@ def test_delta_channel_dict_reducer_with_deletions() -> None: """Dict reducer that treats None values as deletions works end-to-end (deepagents pattern).""" from langgraph.checkpoint.base import DeltaChainValue - from langgraph.channels.delta import DeltaChannel - def merge_files(left: dict | None, right: dict) -> dict: if left is None: return {k: v for k, v in right.items() if v is not None} @@ -502,7 +505,7 @@ def test_delta_channel_dict_reducer_with_deletions() -> None: result[k] = v return result - ch = DeltaChannel(merge_files, dict).from_checkpoint(MISSING) + ch = _delta_channel_with_type(merge_files, dict).from_checkpoint(MISSING) ch.after_checkpoint(None) ch.update([{"file1.py": "content1", "file2.py": "content2"}]) @@ -522,6 +525,6 @@ def test_delta_channel_dict_reducer_with_deletions() -> None: [{"file1.py": None, "file3.py": "content3"}], ], ) - spec = DeltaChannel(merge_files, dict) + spec = _delta_channel_with_type(merge_files, dict) ch2 = spec.from_checkpoint(chain) assert ch2.get() == {"file2.py": "content2", "file3.py": "content3"}