diff --git a/libs/langgraph/langgraph/graph/state.py b/libs/langgraph/langgraph/graph/state.py index deb0e9e94..a608a2a16 100644 --- a/libs/langgraph/langgraph/graph/state.py +++ b/libs/langgraph/langgraph/graph/state.py @@ -5,7 +5,7 @@ import logging import typing import warnings from collections import defaultdict -from collections.abc import Awaitable, Callable, Hashable, Sequence +from collections.abc import Awaitable, Callable, Hashable, Mapping, Sequence from dataclasses import dataclass, is_dataclass from datetime import timedelta from functools import partial @@ -27,7 +27,7 @@ from typing import ( from langchain_core.runnables import Runnable, RunnableConfig from langgraph.cache.base import BaseCache -from langgraph.checkpoint.base import Checkpoint +from langgraph.checkpoint.base import BaseCheckpointSaver, Checkpoint from langgraph.store.base import BaseStore from pydantic import BaseModel, TypeAdapter from typing_extensions import NotRequired, Required, Self, Unpack, is_typeddict @@ -120,6 +120,35 @@ def _warn_invalid_state_schema(schema: type[Any] | Any) -> None: ) +def _warn_if_checkpointer_lacks_delta_support( + channels: Mapping[str, Any], checkpointer: Checkpointer +) -> None: + """Warn when a `DeltaChannel` graph has a checkpointer that can't reconstruct it. + + `DeltaChannel` stores only a sentinel in checkpoint blobs and rebuilds + state by replaying ancestor writes via `BaseCheckpointSaver`'s delta + history API (`get_delta_channel_history`, added in + `langgraph-checkpoint>=4.1.0`). A saver from an older `langgraph-checkpoint` + lacks that method, so reconstruction fails at runtime. Warn at compile + time so users know to upgrade. + """ + if not isinstance(checkpointer, BaseCheckpointSaver): + return + if not any(isinstance(c, DeltaChannel) for c in channels.values()): + return + if hasattr(checkpointer, "get_delta_channel_history"): + return + warnings.warn( + f"The configured checkpointer ({type(checkpointer).__name__}) does not " + "support `DeltaChannel` state reconstruction, which requires the " + "`get_delta_channel_history` API added in `langgraph-checkpoint>=4.1.0`. " + "State reconstruction will fail at runtime. Upgrade `langgraph-checkpoint` " + "(e.g. `pip install -U langgraph-checkpoint`) to enable `DeltaChannel` support.", + UserWarning, + stacklevel=4, + ) + + def _get_node_name(node: StateNode[Any, ContextT]) -> str: try: return getattr(node, "__name__", node.__class__.__name__) @@ -1216,6 +1245,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]): CompiledStateGraph: The compiled `StateGraph`. """ checkpointer = ensure_valid_checkpointer(checkpointer) + _warn_if_checkpointer_lacks_delta_support(self.channels, checkpointer) serde_allowlist: set[tuple[str, ...]] | None = None if _serde.STRICT_MSGPACK_ENABLED: diff --git a/libs/langgraph/tests/test_delta_channel_checkpointer_support.py b/libs/langgraph/tests/test_delta_channel_checkpointer_support.py new file mode 100644 index 000000000..7e516d425 --- /dev/null +++ b/libs/langgraph/tests/test_delta_channel_checkpointer_support.py @@ -0,0 +1,84 @@ +"""Compile-time warning when a checkpointer can't reconstruct `DeltaChannel`. + +`DeltaChannel` reconstructs state via `BaseCheckpointSaver`'s delta history +API (`get_delta_channel_history`, added in `langgraph-checkpoint>=4.1.0`). A +saver from an older package lacks that method, so reconstruction fails at +runtime. `StateGraph.compile` warns at compile time so users upgrade. +""" + +from __future__ import annotations + +import warnings +from typing import Annotated + +import pytest +from langgraph.checkpoint.base import BaseCheckpointSaver +from langgraph.checkpoint.memory import InMemorySaver +from typing_extensions import TypedDict + +from langgraph.channels.delta import DeltaChannel +from langgraph.constants import START +from langgraph.graph import StateGraph +from langgraph.graph.message import _messages_delta_reducer + + +class _DeltaState(TypedDict): + messages: Annotated[ + list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=50) + ] + + +class _PlainState(TypedDict): + values: Annotated[list, lambda a, b: (a or []) + (b or [])] + + +def _delta_graph() -> StateGraph: + return ( + StateGraph(_DeltaState) + .add_node("n", lambda s: {"messages": []}) + .add_edge(START, "n") + ) + + +@pytest.fixture +def saver_without_delta_api(monkeypatch: pytest.MonkeyPatch) -> InMemorySaver: + """A saver whose class tree lacks the delta history API (old package).""" + for cls in (InMemorySaver, BaseCheckpointSaver): + for name in ("get_delta_channel_history", "aget_delta_channel_history"): + if name in cls.__dict__: + monkeypatch.delattr(cls, name) + saver = InMemorySaver() + assert not hasattr(saver, "get_delta_channel_history") + return saver + + +def test_delta_graph_warns_for_unsupported_checkpointer( + saver_without_delta_api: InMemorySaver, +) -> None: + with pytest.warns(UserWarning, match="does not support `DeltaChannel`"): + _delta_graph().compile(checkpointer=saver_without_delta_api) + + +def test_delta_graph_no_warning_for_supported_checkpointer() -> None: + with warnings.catch_warnings(): + warnings.simplefilter("error") + _delta_graph().compile(checkpointer=InMemorySaver()) + + +def test_delta_graph_no_warning_without_checkpointer() -> None: + with warnings.catch_warnings(): + warnings.simplefilter("error") + _delta_graph().compile() + + +def test_non_delta_graph_no_warning_with_unsupported_checkpointer( + saver_without_delta_api: InMemorySaver, +) -> None: + graph = ( + StateGraph(_PlainState) + .add_node("n", lambda s: {"values": []}) + .add_edge(START, "n") + ) + with warnings.catch_warnings(): + warnings.simplefilter("error") + graph.compile(checkpointer=saver_without_delta_api)