From 2284a54b641de9d31378d7c3a53b5be2ce7dabd6 Mon Sep 17 00:00:00 2001 From: Caspar Broekhuizen Date: Thu, 20 Nov 2025 14:23:09 -0800 Subject: [PATCH] fix: interrupt stream mode values (#6475) This PR improves the consistency of interrupt streaming. - when streaming with stream_mode values, the stream chunk now contains the entire state alongside the interrupt: ```python class State(TypedDict): robot_input: str # at this point in time robot_input is already set to "beep boop i am a robot" app.stream(..., stream_mode="values") # before {"__interrupt__": (Interrupt(value="interrupt",))}} # after {"robot_input": "beep boop i am a robot", "__interrupt__": (Interrupt(value="interrupt"))} ``` - when streaming with stream_mode=["values", "updates"], interrupts are surfaced in both an update stream chunk and the value stream chunk, when previously we keep interrupt in values only if we request values mode only ```python class State(TypedDict): robot_input: str # at this point in time robot_input is already set to "beep boop i am a robot" app.stream(..., stream_mode=["values", "updates"]) # before (interrupt would only emit on update chunk, there would be no values chunk) ("updates", {"__interrupt__": (Interrupt(value="interrupt",))}}) # after ("updates", {"__interrupt__": (Interrupt(value="interrupt",))}}) ("values", {"robot_input": "beep boop i am a robot", "__interrupt__": (Interrupt(value="interrupt"))}) ``` For housekeeping: this PR improves on this revert: https://github.com/langchain-ai/langgraph/pull/6141 --- libs/langgraph/langgraph/pregel/_loop.py | 12 ++++- libs/langgraph/tests/test_pregel.py | 44 ++++++++++++++--- libs/langgraph/tests/test_pregel_async.py | 58 +++++++++++++++++++++++ libs/prebuilt/tests/test_react_agent.py | 5 +- 4 files changed, 109 insertions(+), 10 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/_loop.py b/libs/langgraph/langgraph/pregel/_loop.py index 6876b6c7f..79201d0d6 100644 --- a/libs/langgraph/langgraph/pregel/_loop.py +++ b/libs/langgraph/langgraph/pregel/_loop.py @@ -766,6 +766,7 @@ class PregelLoop: ] self.checkpoint["channel_values"][TASKS] = sanitized_tasks # bail if no checkpointer + if do_checkpoint and self._checkpointer_put_after_previous is not None: self.prev_checkpoint_config = ( self.checkpoint_config @@ -935,8 +936,15 @@ class PregelLoop: stream_modes = self.stream.modes if self.stream else [] if "updates" in stream_modes: self._emit("updates", lambda: iter(interrupts)) - elif "values" in stream_modes: - self._emit("values", lambda: iter(interrupts)) + if "values" in stream_modes: + current_values = read_channels(self.channels, self.output_keys) + # self.output_keys is a sequence, stream chunk contains entire state and interrupts + if isinstance(current_values, dict): + current_values[INTERRUPT] = interrupts[0][INTERRUPT] + self._emit("values", lambda: iter([current_values])) + # self.output_keys is a string, stream chunk contains only interrupts + else: + self._emit("values", lambda: iter(interrupts)) elif writes[0][0] != ERROR: self._emit( "updates", diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index 73e403f81..5ede65afc 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -8422,23 +8422,55 @@ def test_null_resume_disallowed_with_multiple_interrupts( } -def test_interrupt_stream_mode_values(): - """Test that interrupts are surfaced when steam_mode='values'""" +def test_interrupt_stream_mode_values(sync_checkpointer: BaseCheckpointSaver): + """Test that interrupts are surfaced on 'values' stream mode""" class State(TypedDict): + robot_input: str human_input: str + def robot_input_node(state: State) -> State: + return {"robot_input": "beep boop i am a robot"} + def human_input_node(state: State) -> Command: human_input = interrupt("interrupt") return Command(update={"human_input": human_input}) builder = StateGraph(State) + builder.add_node(robot_input_node) builder.add_node(human_input_node) - builder.add_edge(START, "human_input_node") - app = builder.compile() + builder.add_edge(START, "robot_input_node") + builder.add_edge("robot_input_node", "human_input_node") + app = builder.compile(checkpointer=sync_checkpointer) + config = {"configurable": {"thread_id": str(uuid.uuid4())}} - result = [*app.stream(State(), stream_mode="values")] - assert "__interrupt__" in result[-1] + result = [*app.stream(State(), config, stream_mode=["updates", "values"])] + assert len(result) == 4 + assert result == [ + ("updates", {"robot_input_node": {"robot_input": "beep boop i am a robot"}}), + ("values", {"robot_input": "beep boop i am a robot"}), + ("updates", {"__interrupt__": (Interrupt(value="interrupt", id=AnyStr()),)}), + ( + "values", + { + "robot_input": "beep boop i am a robot", + "__interrupt__": (Interrupt(value="interrupt", id=AnyStr()),), + }, + ), + ] + resume_result = [ + *app.stream( + Command(resume="i am a human"), config, stream_mode=["updates", "values"] + ) + ] + assert resume_result == [ + ("values", {"robot_input": "beep boop i am a robot"}), + ("updates", {"human_input_node": {"human_input": "i am a human"}}), + ( + "values", + {"robot_input": "beep boop i am a robot", "human_input": "i am a human"}, + ), + ] def test_supersteps_populate_task_results( diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index ce16b94fd..e57c09b3d 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -9196,6 +9196,64 @@ async def test_astream_waiter_cleanup_on_cancel( assert all(t.done() for t in recorded_tasks) +@NEEDS_CONTEXTVARS +async def test_interrupt_stream_mode_values(async_checkpointer: BaseCheckpointSaver): + """Test that interrupts are surfaced on 'values' stream mode""" + + class State(TypedDict): + robot_input: str + human_input: str + + def robot_input_node(state: State) -> State: + return {"robot_input": "beep boop i am a robot"} + + def human_input_node(state: State) -> Command: + human_input = interrupt("interrupt") + return Command(update={"human_input": human_input}) + + builder = StateGraph(State) + builder.add_node(robot_input_node) + builder.add_node(human_input_node) + builder.add_edge(START, "robot_input_node") + builder.add_edge("robot_input_node", "human_input_node") + app = builder.compile(checkpointer=async_checkpointer) + config = {"configurable": {"thread_id": str(uuid.uuid4())}} + + result = [ + (mode, e) + async for mode, e in app.astream( + State(), config, stream_mode=["updates", "values"] + ) + ] + assert len(result) == 4 + assert result == [ + ("updates", {"robot_input_node": {"robot_input": "beep boop i am a robot"}}), + ("values", {"robot_input": "beep boop i am a robot"}), + ("updates", {"__interrupt__": (Interrupt(value="interrupt", id=AnyStr()),)}), + ( + "values", + { + "robot_input": "beep boop i am a robot", + "__interrupt__": (Interrupt(value="interrupt", id=AnyStr()),), + }, + ), + ] + resume_result = [ + (mode, e) + async for mode, e in app.astream( + Command(resume="i am a human"), config, stream_mode=["updates", "values"] + ) + ] + assert resume_result == [ + ("values", {"robot_input": "beep boop i am a robot"}), + ("updates", {"human_input_node": {"human_input": "i am a human"}}), + ( + "values", + {"robot_input": "beep boop i am a robot", "human_input": "i am a human"}, + ), + ] + + async def test_supersteps_populate_task_results( async_checkpointer: BaseCheckpointSaver, ) -> None: diff --git a/libs/prebuilt/tests/test_react_agent.py b/libs/prebuilt/tests/test_react_agent.py index eaf42106a..4dd265d4b 100644 --- a/libs/prebuilt/tests/test_react_agent.py +++ b/libs/prebuilt/tests/test_react_agent.py @@ -638,8 +638,9 @@ def test_react_agent_parallel_tool_calls( for event in agent.stream( {"messages": [("user", query)]}, config, stream_mode="values" ): - if messages := event.get("messages"): - message_types.append([m.type for m in messages]) + if "__interrupt__" not in event: + if messages := event.get("messages"): + message_types.append([m.type for m in messages]) if version == "v1": assert message_types == [