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

refactor(delta-channel): infer typ from Annotated outer type instead of constructor arg

Browse Source 
Remove the positional `typ` parameter from `DeltaChannel.__init__`. The type is
now injected automatically from the `Annotated` outer type in `_is_field_channel`
(matching how `BinaryOperatorAggregate` receives its type). `copy()` and
`from_checkpoint()` propagate `self.typ` explicitly. Test helpers updated to
use `_get_channel` with the proper `Annotated` path.
This commit is contained in:
Sydney Runkle
2026-04-22 14:03:37 -04:00
parent 10da2326a9
commit 06e302bbff
3 changed files with 43 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/langgraph/langgraph/channels/delta.py
+13 -17
View File
@@ -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()
libs/langgraph/langgraph/graph/state.py
+15 -1
View File
@@ -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]
libs/langgraph/tests/test_channels.py
+15 -12
View File
@@ -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"}