From afdfbd55dac7cb8ec1927fe685b30ac4dc4eacc4 Mon Sep 17 00:00:00 2001 From: Sydney Runkle <54324534+sydney-runkle@users.noreply.github.com> Date: Fri, 1 May 2026 13:56:01 -0400 Subject: [PATCH] fix(langgraph): coerce dict/str writes in _messages_delta_reducer (#7680) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem `_messages_delta_reducer` assumed writes always contain pre-typed `BaseMessage` objects (as noted in its docstring). In practice, HTTP-driven graphs always receive message input as JSON dicts — the same way `add_messages` receives them. Using `DeltaChannel(_messages_delta_reducer)` with any HTTP input would crash with: ``` AttributeError: 'dict' object has no attribute 'id' ``` This makes `_messages_delta_reducer` unusable for the primary motivating use-case (replacing `add_messages` in production LLM graphs). ## Fix Mirror the coercion contract of `add_messages`: - Regular message dicts (`{"role": "human", "content": "..."}`) → `convert_to_messages` - `RemoveMessage` dicts (`{"type": "remove", "id": "..."}`) → `RemoveMessage` directly (langchain_core's `convert_to_messages` doesn't support this format) - `BaseMessage` objects → pass through unchanged - Lists/sequences → element-wise coercion of the above The fix is a small `_coerce_one` + `_to_msgs` helper pair that replaces the previous `[w] if isinstance(w, BaseMessage) else w` generator. ## Tests Added `test_delta_channel_dict_coercion` in `test_channels.py` covering: - dict append via `{"role": "human", "content": ..., "id": ...}` - dict update-in-place (same ID) - `{"type": "remove", "id": ...}` tombstoning All 23 existing delta-channel tests still pass. Release Notes: None --------- Co-authored-by: Claude Opus 4.7 (1M context) --- libs/langgraph/langgraph/graph/message.py | 39 ++++++++++++----- libs/langgraph/tests/test_channels.py | 52 +++++++++++++++++++++++ 2 files changed, 81 insertions(+), 10 deletions(-) diff --git a/libs/langgraph/langgraph/graph/message.py b/libs/langgraph/langgraph/graph/message.py index 3018d21da..64f375ad2 100644 --- a/libs/langgraph/langgraph/graph/message.py +++ b/libs/langgraph/langgraph/graph/message.py @@ -250,14 +250,16 @@ def _messages_delta_reducer( """**Experimental.** Batch reducer for use with `DeltaChannel`. Processes all writes in one pass — dedup by ID, `RemoveMessage` - tombstoning — without calling `add_messages`. Assumes writes contain - already-typed `BaseMessage` objects (no raw-dict coercion). + tombstoning — without calling `add_messages`. This reducer is batching-invariant, as required by `DeltaChannel`: `reducer(reducer(state, xs), ys) == reducer(state, xs + ys)`. - Use `add_messages` as the reducer for `BinaryOperatorAggregate` or - anywhere raw message dicts / strings need to be coerced first. + Raw dict / string / tuple inputs are coerced to typed `BaseMessage` + objects so that HTTP-driven graphs work without a separate coercion + step. This is not full `add_messages` parity — `REMOVE_ALL_MESSAGES`, + unknown-id `RemoveMessage` errors, missing-id UUID assignment, and + `BaseMessageChunk` conversion are not handled here. Example:: @@ -268,13 +270,30 @@ def _messages_delta_reducer( class State(TypedDict): messages: Annotated[list, DeltaChannel(_messages_delta_reducer)] """ - from itertools import chain - index: dict[str, int] = {m.id: i for i, m in enumerate(state) if m.id is not None} - result: list[AnyMessage | None] = list(state) - for msg in chain.from_iterable( - [w] if isinstance(w, BaseMessage) else w for w in writes - ): + # Each write is either a list of message-likes or a single message-like + # (BaseMessage / dict / str / tuple). Only lists flatten; everything + # else is one message. + flat: list[Any] = [] + for w in writes: + if isinstance(w, list): + flat.extend(w) + else: + flat.append(w) + # Steady state: the reducer's own output is already typed, so skip + # `convert_to_messages` on state when the first element is a BaseMessage. + # Only raw input (initial dicts, deserialized blobs) hits the slow path. + if state and isinstance(state[0], BaseMessage): + state_msgs = state + else: + state_msgs = cast("list[AnyMessage]", convert_to_messages(state)) + msgs = cast("list[AnyMessage]", convert_to_messages(flat)) + + index: dict[str, int] = { + m.id: i for i, m in enumerate(state_msgs) if m.id is not None + } + result: list[AnyMessage | None] = list(state_msgs) + for msg in msgs: mid = msg.id if mid is None: result.append(msg) diff --git a/libs/langgraph/tests/test_channels.py b/libs/langgraph/tests/test_channels.py index fd2e35e33..6c15f3307 100644 --- a/libs/langgraph/tests/test_channels.py +++ b/libs/langgraph/tests/test_channels.py @@ -233,6 +233,58 @@ def test_delta_channel_update_by_id_and_replay() -> None: assert ch2.get()[0].content == "updated" +def test_delta_channel_dict_coercion() -> None: + """_messages_delta_reducer coerces dict writes to BaseMessage objects. + + HTTP-driven input always arrives as JSON dicts. The reducer must coerce + them (same contract as add_messages) so graphs work without a separate + coercion step. + """ + ch = DeltaChannel(_messages_delta_reducer, list).from_checkpoint(MISSING) + + # dict input — simulates what arrives from the HTTP API + ch.update([{"role": "human", "content": "hello", "id": "h1"}]) + assert len(ch.get()) == 1 + assert isinstance(ch.get()[0], HumanMessage) + assert ch.get()[0].content == "hello" + assert ch.get()[0].id == "h1" + + # update by ID via dict + ch.update([{"role": "ai", "content": "world", "id": "h1"}]) + assert len(ch.get()) == 1 + assert ch.get()[0].content == "world" + + # remove via RemoveMessage instance (same contract as add_messages) + ch.update([RemoveMessage(id="h1")]) + assert ch.get() == [] + + +def test_messages_delta_reducer_coerces_state() -> None: + """State (left side) is coerced when raw — supports raw initial input + and deserialized blobs. The steady-state path (state already typed) + short-circuits and skips coercion. + """ + state = [{"role": "human", "content": "hello", "id": "h1"}] + writes = [[{"role": "ai", "content": "world", "id": "h1"}]] + result = _messages_delta_reducer(state, writes) # type: ignore[arg-type] + assert len(result) == 1 + assert isinstance(result[0], AIMessage) + assert result[0].content == "world" + assert result[0].id == "h1" + + +def test_messages_delta_reducer_tuple_write_is_one_message() -> None: + """A top-level tuple write is one message-like, not a sequence to flatten. + + `("user", "hi")` is a valid `MessageLikeRepresentation`; flattening it + would produce two HumanMessages ("user", "hi") instead of one. + """ + result = _messages_delta_reducer([], [("user", "hi")]) # type: ignore[arg-type] + assert len(result) == 1 + assert isinstance(result[0], HumanMessage) + assert result[0].content == "hi" + + def test_delta_channel_checkpoint_returns_sentinel() -> None: """checkpoint() always returns DELTA_SENTINEL regardless of state.""" ch = DeltaChannel(_messages_delta_reducer, list).from_checkpoint(MISSING)