From 569055539449409e95c4760c71f7c77361f6ed3f Mon Sep 17 00:00:00 2001 From: vbarda Date: Tue, 8 Apr 2025 12:51:36 -0400 Subject: [PATCH 1/4] langgraph: raise GraphInterrupt only if used as a subgraph --- libs/langgraph/langgraph/pregel/remote.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/remote.py b/libs/langgraph/langgraph/pregel/remote.py index f58d0a4d9..07d44cb66 100644 --- a/libs/langgraph/langgraph/pregel/remote.py +++ b/libs/langgraph/langgraph/pregel/remote.py @@ -654,9 +654,10 @@ class RemoteGraph(PregelProtocol): # raise interrupt or errors if chunk.event.startswith("updates"): if isinstance(chunk.data, dict) and INTERRUPT in chunk.data: - raise GraphInterrupt( - [Interrupt(**i) for i in chunk.data[INTERRUPT]] - ) + if caller_ns: + raise GraphInterrupt( + [Interrupt(**i) for i in chunk.data[INTERRUPT]] + ) elif chunk.event.startswith("error"): raise RemoteException(chunk.data) # filter for what was actually requested @@ -748,9 +749,10 @@ class RemoteGraph(PregelProtocol): # raise interrupt or errors if chunk.event.startswith("updates"): if isinstance(chunk.data, dict) and INTERRUPT in chunk.data: - raise GraphInterrupt( - [Interrupt(**i) for i in chunk.data[INTERRUPT]] - ) + if caller_ns: + raise GraphInterrupt( + [Interrupt(**i) for i in chunk.data[INTERRUPT]] + ) elif chunk.event.startswith("error"): raise RemoteException(chunk.data) # filter for what was actually requested From 8f32fc4819a3b1b2e93bf1490fa80c57d3cd7678 Mon Sep 17 00:00:00 2001 From: vbarda Date: Tue, 8 Apr 2025 13:01:28 -0400 Subject: [PATCH 2/4] update tests --- libs/langgraph/tests/test_remote_graph.py | 180 ++++++++++++---------- 1 file changed, 101 insertions(+), 79 deletions(-) diff --git a/libs/langgraph/tests/test_remote_graph.py b/libs/langgraph/tests/test_remote_graph.py index 4aea1fc32..878f5e6f1 100644 --- a/libs/langgraph/tests/test_remote_graph.py +++ b/libs/langgraph/tests/test_remote_graph.py @@ -437,15 +437,17 @@ def test_stream(): sync_client=mock_sync_client, ) - # stream modes doesn't include 'updates' - stream_parts = [] + # test raising graph interrupt if invoked as a subgraph with pytest.raises(GraphInterrupt) as exc: for stream_part in remote_pregel.stream( {"input": "data"}, - config={"configurable": {"thread_id": "thread_1"}}, + # pretend we invoked this as a subgraph + config={ + "configurable": {"thread_id": "thread_1", "checkpoint_ns": "some_ns"} + }, stream_mode="values", ): - stream_parts.append(stream_part) + pass assert exc.value.args[0] == [ Interrupt( @@ -456,6 +458,15 @@ def test_stream(): ) ] + # stream modes doesn't include 'updates' + stream_parts = [] + for stream_part in remote_pregel.stream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode="values", + ): + stream_parts.append(stream_part) + assert stream_parts == [ {"chunk": "data1"}, {"chunk": "data2"}, @@ -470,62 +481,62 @@ def test_stream(): # default stream_mode is updates stream_parts = [] - with pytest.raises(GraphInterrupt): - for stream_part in remote_pregel.stream( - {"input": "data"}, - config={"configurable": {"thread_id": "thread_1"}}, - ): - stream_parts.append(stream_part) + for stream_part in remote_pregel.stream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + ): + stream_parts.append(stream_part) assert stream_parts == [ {"chunk": "data3"}, {"chunk": "data4"}, + {"__interrupt__": ()}, ] # list stream_mode includes mode names stream_parts = [] - with pytest.raises(GraphInterrupt): - for stream_part in remote_pregel.stream( - {"input": "data"}, - config={"configurable": {"thread_id": "thread_1"}}, - stream_mode=["updates"], - ): - stream_parts.append(stream_part) + for stream_part in remote_pregel.stream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode=["updates"], + ): + stream_parts.append(stream_part) assert stream_parts == [ ("updates", {"chunk": "data3"}), ("updates", {"chunk": "data4"}), + ("updates", {"__interrupt__": ()}), ] # subgraphs + list modes stream_parts = [] - with pytest.raises(GraphInterrupt): - for stream_part in remote_pregel.stream( - {"input": "data"}, - config={"configurable": {"thread_id": "thread_1"}}, - stream_mode=["updates"], - subgraphs=True, - ): - stream_parts.append(stream_part) + for stream_part in remote_pregel.stream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode=["updates"], + subgraphs=True, + ): + stream_parts.append(stream_part) assert stream_parts == [ ((), "updates", {"chunk": "data3"}), ((), "updates", {"chunk": "data4"}), + ((), "updates", {"__interrupt__": ()}), ] # subgraphs + single mode stream_parts = [] - with pytest.raises(GraphInterrupt): - for stream_part in remote_pregel.stream( - {"input": "data"}, - config={"configurable": {"thread_id": "thread_1"}}, - subgraphs=True, - ): - stream_parts.append(stream_part) + for stream_part in remote_pregel.stream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + subgraphs=True, + ): + stream_parts.append(stream_part) assert stream_parts == [ ((), {"chunk": "data3"}), ((), {"chunk": "data4"}), + ((), {"__interrupt__": ()}), ] @@ -561,15 +572,17 @@ async def test_astream(): client=mock_async_client, ) - # stream modes doesn't include 'updates' - stream_parts = [] + # test raising graph interrupt if invoked as a subgraph with pytest.raises(GraphInterrupt) as exc: async for stream_part in remote_pregel.astream( {"input": "data"}, - config={"configurable": {"thread_id": "thread_1"}}, + # pretend we invoked this as a subgraph + config={ + "configurable": {"thread_id": "thread_1", "checkpoint_ns": "some_ns"} + }, stream_mode="values", ): - stream_parts.append(stream_part) + pass assert exc.value.args[0] == [ Interrupt( @@ -580,6 +593,15 @@ async def test_astream(): ) ] + # stream modes doesn't include 'updates' + stream_parts = [] + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode="values", + ): + stream_parts.append(stream_part) + assert stream_parts == [ {"chunk": "data1"}, {"chunk": "data2"}, @@ -596,62 +618,62 @@ async def test_astream(): # default stream_mode is updates stream_parts = [] - with pytest.raises(GraphInterrupt): - async for stream_part in remote_pregel.astream( - {"input": "data"}, - config={"configurable": {"thread_id": "thread_1"}}, - ): - stream_parts.append(stream_part) + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + ): + stream_parts.append(stream_part) assert stream_parts == [ {"chunk": "data3"}, {"chunk": "data4"}, + {"__interrupt__": ()}, ] # list stream_mode includes mode names stream_parts = [] - with pytest.raises(GraphInterrupt): - async for stream_part in remote_pregel.astream( - {"input": "data"}, - config={"configurable": {"thread_id": "thread_1"}}, - stream_mode=["updates"], - ): - stream_parts.append(stream_part) + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode=["updates"], + ): + stream_parts.append(stream_part) assert stream_parts == [ ("updates", {"chunk": "data3"}), ("updates", {"chunk": "data4"}), + ("updates", {"__interrupt__": ()}), ] # subgraphs + list modes stream_parts = [] - with pytest.raises(GraphInterrupt): - async for stream_part in remote_pregel.astream( - {"input": "data"}, - config={"configurable": {"thread_id": "thread_1"}}, - stream_mode=["updates"], - subgraphs=True, - ): - stream_parts.append(stream_part) + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode=["updates"], + subgraphs=True, + ): + stream_parts.append(stream_part) assert stream_parts == [ ((), "updates", {"chunk": "data3"}), ((), "updates", {"chunk": "data4"}), + ((), "updates", {"__interrupt__": ()}), ] # subgraphs + single mode stream_parts = [] - with pytest.raises(GraphInterrupt): - async for stream_part in remote_pregel.astream( - {"input": "data"}, - config={"configurable": {"thread_id": "thread_1"}}, - subgraphs=True, - ): - stream_parts.append(stream_part) + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + subgraphs=True, + ): + stream_parts.append(stream_part) assert stream_parts == [ ((), {"chunk": "data3"}), ((), {"chunk": "data4"}), + ((), {"__interrupt__": ()}), ] async_iter = MagicMock() @@ -664,33 +686,33 @@ async def test_astream(): # subgraphs + list modes stream_parts = [] - with pytest.raises(GraphInterrupt): - async for stream_part in remote_pregel.astream( - {"input": "data"}, - config={"configurable": {"thread_id": "thread_1"}}, - stream_mode=["updates"], - subgraphs=True, - ): - stream_parts.append(stream_part) + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode=["updates"], + subgraphs=True, + ): + stream_parts.append(stream_part) assert stream_parts == [ (("my", "subgraph"), "updates", {"chunk": "data3"}), (("hello", "subgraph"), "updates", {"chunk": "data4"}), + (("bye", "subgraph"), "updates", {"__interrupt__": ()}), ] # subgraphs + single mode stream_parts = [] - with pytest.raises(GraphInterrupt): - async for stream_part in remote_pregel.astream( - {"input": "data"}, - config={"configurable": {"thread_id": "thread_1"}}, - subgraphs=True, - ): - stream_parts.append(stream_part) + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + subgraphs=True, + ): + stream_parts.append(stream_part) assert stream_parts == [ (("my", "subgraph"), {"chunk": "data3"}), (("hello", "subgraph"), {"chunk": "data4"}), + (("bye", "subgraph"), {"__interrupt__": ()}), ] From cff349e22ed2a70bcb8bf513577e290cd6a19c94 Mon Sep 17 00:00:00 2001 From: vbarda Date: Tue, 8 Apr 2025 13:34:47 -0400 Subject: [PATCH 3/4] add warning --- libs/langgraph/langgraph/pregel/remote.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/libs/langgraph/langgraph/pregel/remote.py b/libs/langgraph/langgraph/pregel/remote.py index 07d44cb66..b5c38a046 100644 --- a/libs/langgraph/langgraph/pregel/remote.py +++ b/libs/langgraph/langgraph/pregel/remote.py @@ -1,3 +1,4 @@ +import warnings from dataclasses import asdict from typing import ( Any, @@ -620,6 +621,13 @@ class RemoteGraph(PregelProtocol): stream_mode, config ) if isinstance(input, Command): + if input.resume and "thread_id" not in sanitized_config["configurable"]: + warnings.warn( + "Trying to resume interrupted graph without `thread_id` in the config. " + "This will likely lead to downstream errors. Please provide a valid config in the following form: " + "{'configurable': {'thread_id': '...'}}." + ) + command: Optional[CommandSDK] = cast(CommandSDK, asdict(input)) input = None else: @@ -715,6 +723,13 @@ class RemoteGraph(PregelProtocol): stream_mode, config ) if isinstance(input, Command): + if input.resume and "thread_id" not in sanitized_config["configurable"]: + warnings.warn( + "Trying to resume interrupted graph without `thread_id` in the config. " + "This will likely lead to downstream errors. Please provide a valid config in the following form: " + "{'configurable': {'thread_id': '...'}}." + ) + command: Optional[CommandSDK] = cast(CommandSDK, asdict(input)) input = None else: From 41fb5ec77cfe55a1bb11f3a896f538c6ff97cae4 Mon Sep 17 00:00:00 2001 From: vbarda Date: Tue, 8 Apr 2025 13:38:49 -0400 Subject: [PATCH 4/4] Revert "add warning" This reverts commit cff349e22ed2a70bcb8bf513577e290cd6a19c94. --- libs/langgraph/langgraph/pregel/remote.py | 15 --------------- 1 file changed, 15 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/remote.py b/libs/langgraph/langgraph/pregel/remote.py index b5c38a046..07d44cb66 100644 --- a/libs/langgraph/langgraph/pregel/remote.py +++ b/libs/langgraph/langgraph/pregel/remote.py @@ -1,4 +1,3 @@ -import warnings from dataclasses import asdict from typing import ( Any, @@ -621,13 +620,6 @@ class RemoteGraph(PregelProtocol): stream_mode, config ) if isinstance(input, Command): - if input.resume and "thread_id" not in sanitized_config["configurable"]: - warnings.warn( - "Trying to resume interrupted graph without `thread_id` in the config. " - "This will likely lead to downstream errors. Please provide a valid config in the following form: " - "{'configurable': {'thread_id': '...'}}." - ) - command: Optional[CommandSDK] = cast(CommandSDK, asdict(input)) input = None else: @@ -723,13 +715,6 @@ class RemoteGraph(PregelProtocol): stream_mode, config ) if isinstance(input, Command): - if input.resume and "thread_id" not in sanitized_config["configurable"]: - warnings.warn( - "Trying to resume interrupted graph without `thread_id` in the config. " - "This will likely lead to downstream errors. Please provide a valid config in the following form: " - "{'configurable': {'thread_id': '...'}}." - ) - command: Optional[CommandSDK] = cast(CommandSDK, asdict(input)) input = None else: