mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-21 23:22:27 +02:00
fix: restore dict-reducer tests; rename _delta → delta throughout
The dict-reducer tests were incorrectly removed — they test valid DeltaChannel behavior (dict type inference via Annotated) that was already supported by the base branch's _is_field_channel logic in state.py. The only thing needed was fixing the module name. Also updates all remaining channels._delta imports to channels.delta across state.py and test files. Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 4.6
parent
036d29f30e
commit
a455cefd24
@@ -317,6 +317,233 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None:
|
||||
assert len(state.values["messages"]) == 4 # 2 human + 2 AI
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dict-reducer tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
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."""
|
||||
|
||||
def merge_dicts(left: dict, right: dict) -> dict:
|
||||
return {**left, **right}
|
||||
|
||||
ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING)
|
||||
assert ch.is_available()
|
||||
assert ch.get() == {}
|
||||
|
||||
|
||||
def test_delta_channel_dict_reducer_basic_updates() -> None:
|
||||
"""DeltaChannel with a dict reducer accumulates key/value pairs across steps."""
|
||||
|
||||
def merge_dicts(left: dict, right: dict) -> dict:
|
||||
return {**left, **right}
|
||||
|
||||
ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING)
|
||||
|
||||
ch.update([{"a": 1}])
|
||||
d1 = ch.checkpoint()
|
||||
assert d1 is DELTA_SENTINEL
|
||||
|
||||
ch.update([{"b": 2}])
|
||||
d2 = ch.checkpoint()
|
||||
assert d2 is DELTA_SENTINEL
|
||||
|
||||
assert ch.get() == {"a": 1, "b": 2}
|
||||
|
||||
|
||||
def test_delta_channel_dict_reducer_writes_reconstruction() -> None:
|
||||
"""replay_writes on a fresh channel replays through a dict merge reducer."""
|
||||
|
||||
def merge_dicts(left: dict, right: dict) -> dict:
|
||||
return {**left, **right}
|
||||
|
||||
spec = _delta_channel_with_type(merge_dicts, dict)
|
||||
ch = spec.from_checkpoint(DELTA_SENTINEL)
|
||||
ch.replay_writes(
|
||||
[
|
||||
("t0", "files", {"a": 1}),
|
||||
("t1", "files", {"b": 2}),
|
||||
("t2", "files", {"c": 3}),
|
||||
]
|
||||
)
|
||||
assert ch.get() == {"a": 1, "b": 2, "c": 3}
|
||||
|
||||
|
||||
def test_delta_channel_dict_reducer_with_deletions() -> None:
|
||||
"""Dict reducer that treats None values as deletions works end-to-end."""
|
||||
|
||||
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}
|
||||
result = {**left}
|
||||
for k, v in right.items():
|
||||
if v is None:
|
||||
result.pop(k, None)
|
||||
else:
|
||||
result[k] = v
|
||||
return result
|
||||
|
||||
ch = _delta_channel_with_type(merge_files, dict).from_checkpoint(MISSING)
|
||||
ch.update([{"file1.py": "content1", "file2.py": "content2"}])
|
||||
ch.update([{"file1.py": None, "file3.py": "content3"}])
|
||||
assert ch.get() == {"file2.py": "content2", "file3.py": "content3"}
|
||||
|
||||
spec = _delta_channel_with_type(merge_files, dict)
|
||||
ch2 = spec.from_checkpoint(DELTA_SENTINEL)
|
||||
ch2.replay_writes(
|
||||
[
|
||||
("t0", "files", {"file1.py": "content1", "file2.py": "content2"}),
|
||||
("t1", "files", {"file1.py": None, "file3.py": "content3"}),
|
||||
]
|
||||
)
|
||||
assert ch2.get() == {"file2.py": "content2", "file3.py": "content3"}
|
||||
|
||||
|
||||
def test_delta_channel_dict_reducer_overwrite_in_update() -> None:
|
||||
"""Overwrite(dict) in update() must preserve dict shape, not coerce to list."""
|
||||
from langgraph.types import Overwrite
|
||||
|
||||
def merge_dicts(left: dict, right: dict) -> dict:
|
||||
return {**left, **right}
|
||||
|
||||
ch = _delta_channel_with_type(merge_dicts, dict).from_checkpoint(MISSING)
|
||||
ch.update([{"a": 1}])
|
||||
ch.update([Overwrite({"b": 2, "c": 3})])
|
||||
assert ch.get() == {"b": 2, "c": 3}
|
||||
|
||||
|
||||
def test_delta_channel_dict_reducer_overwrite_in_writes_replay() -> None:
|
||||
"""Overwrite(dict) embedded in replayed writes must reconstruct as dict."""
|
||||
from langgraph.types import Overwrite
|
||||
|
||||
def merge_dicts(left: dict, right: dict) -> dict:
|
||||
return {**left, **right}
|
||||
|
||||
spec = _delta_channel_with_type(merge_dicts, dict)
|
||||
ch = spec.from_checkpoint(DELTA_SENTINEL)
|
||||
ch.replay_writes(
|
||||
[
|
||||
("t0", "files", {"a": 1}),
|
||||
("t1", "files", Overwrite({"x": 10, "y": 20})),
|
||||
("t2", "files", {"z": 30}),
|
||||
]
|
||||
)
|
||||
assert ch.get() == {"x": 10, "y": 20, "z": 30}
|
||||
|
||||
|
||||
def test_delta_channel_dict_reducer_with_notrequired_annotation() -> None:
|
||||
"""DeltaChannel infers dict type through `Annotated[NotRequired[dict[...]], ch]`."""
|
||||
from typing import Annotated
|
||||
|
||||
from typing_extensions import NotRequired
|
||||
|
||||
from langgraph.channels.delta import DeltaChannel
|
||||
from langgraph.graph.state import _get_channel
|
||||
|
||||
def merge_dicts(left: dict | None, right: dict) -> dict:
|
||||
if left is None:
|
||||
return dict(right)
|
||||
return {**left, **right}
|
||||
|
||||
annotation = Annotated[NotRequired[dict[str, int]], DeltaChannel(merge_dicts)]
|
||||
ch = _get_channel("files", annotation).from_checkpoint(MISSING)
|
||||
assert ch.get() == {}
|
||||
ch.update([{"a": 1}])
|
||||
ch.update([{"b": 2}])
|
||||
assert ch.get() == {"a": 1, "b": 2}
|
||||
|
||||
|
||||
def test_delta_channel_dict_reducer_end_to_end_filesystem() -> None:
|
||||
"""End-to-end: graph with dict-reducer (filesystem-style) channel wrapped in DeltaChannel."""
|
||||
from typing import Annotated
|
||||
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
from typing_extensions import TypedDict
|
||||
|
||||
from langgraph.channels.delta import DeltaChannel
|
||||
from langgraph.graph import START, StateGraph
|
||||
|
||||
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}
|
||||
result = {**left}
|
||||
for k, v in right.items():
|
||||
if v is None:
|
||||
result.pop(k, None)
|
||||
else:
|
||||
result[k] = v
|
||||
return result
|
||||
|
||||
class State(TypedDict):
|
||||
files: Annotated[dict[str, str], DeltaChannel(merge_files)]
|
||||
|
||||
turn = {"v": 0}
|
||||
|
||||
def write_file(state: State) -> dict:
|
||||
turn["v"] += 1
|
||||
n = turn["v"]
|
||||
return {"files": {f"/doc_{n}.txt": f"content for turn {n}"}}
|
||||
|
||||
builder = StateGraph(State)
|
||||
builder.add_node("write_file", write_file)
|
||||
builder.add_edge(START, "write_file")
|
||||
saver = InMemorySaver()
|
||||
graph = builder.compile(checkpointer=saver)
|
||||
config = {"configurable": {"thread_id": "fs"}}
|
||||
|
||||
for _ in range(3):
|
||||
graph.invoke({"files": {}}, config)
|
||||
|
||||
saved = saver.get_tuple(config)
|
||||
assert saved is not None
|
||||
assert saved.checkpoint["channel_values"]["files"] is DELTA_SENTINEL
|
||||
state = graph.get_state(config)
|
||||
assert state.values["files"] == {
|
||||
"/doc_1.txt": "content for turn 1",
|
||||
"/doc_2.txt": "content for turn 2",
|
||||
"/doc_3.txt": "content for turn 3",
|
||||
}
|
||||
|
||||
def delete_file(state: State) -> dict:
|
||||
return {"files": {"/doc_1.txt": None}}
|
||||
|
||||
builder2 = StateGraph(State)
|
||||
builder2.add_node("write_file", write_file)
|
||||
builder2.add_node("delete_file", delete_file)
|
||||
builder2.add_edge(START, "write_file")
|
||||
builder2.add_edge("write_file", "delete_file")
|
||||
turn["v"] = 0
|
||||
saver2 = InMemorySaver()
|
||||
graph2 = builder2.compile(checkpointer=saver2)
|
||||
config2 = {"configurable": {"thread_id": "fs2"}}
|
||||
graph2.invoke({"files": {}}, config2)
|
||||
state2 = graph2.get_state(config2)
|
||||
assert state2.values["files"] == {}
|
||||
|
||||
|
||||
def test_delta_channel_dict_reducer_backwards_compat() -> None:
|
||||
"""A pre-DeltaChannel dict checkpoint must load as a dict, not be listified."""
|
||||
|
||||
def merge_dicts(left: dict, right: dict) -> dict:
|
||||
return {**left, **right}
|
||||
|
||||
spec = _delta_channel_with_type(merge_dicts, dict)
|
||||
old_value = {"a": 1, "b": 2}
|
||||
ch = spec.from_checkpoint(old_value)
|
||||
assert ch.get() == {"a": 1, "b": 2}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# seed / pre-delta migration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user