diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index e33df210c..9e7aa561d 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -2806,19 +2806,23 @@ class Pregel(PregelProtocol): **kwargs, ): if stream_mode == "values": - if isinstance(chunk, dict): - if (ints := chunk.get(INTERRUPT)) is not None: - interrupts.extend(ints) - latest = chunk + if ( + isinstance(chunk, dict) + and (ints := chunk.get(INTERRUPT)) is not None + ): + interrupts.extend(ints) + else: + latest = chunk else: chunks.append(chunk) if stream_mode == "values": if interrupts: - return { - **latest, - INTERRUPT: interrupts, - } + return ( + {**latest, INTERRUPT: interrupts} + if isinstance(latest, dict) + else {INTERRUPT: interrupts} + ) return latest else: return chunks @@ -2871,19 +2875,23 @@ class Pregel(PregelProtocol): **kwargs, ): if stream_mode == "values": - if isinstance(chunk, dict): - if (ints := chunk.get(INTERRUPT)) is not None: - interrupts.extend(ints) - latest = chunk + if ( + isinstance(chunk, dict) + and (ints := chunk.get(INTERRUPT)) is not None + ): + interrupts.extend(ints) + else: + latest = chunk else: chunks.append(chunk) if stream_mode == "values": if interrupts: - return { - **latest, - INTERRUPT: interrupts, - } + return ( + {**latest, INTERRUPT: interrupts} + if isinstance(latest, dict) + else {INTERRUPT: interrupts} + ) return latest else: return chunks diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index 7b3cc987e..b2a681cb7 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -1,7 +1,5 @@ import dataclasses -import hashlib import sys -import uuid from collections import deque from collections.abc import Hashable, Sequence from typing import ( @@ -21,6 +19,7 @@ from typing import ( from langchain_core.runnables import Runnable, RunnableConfig from typing_extensions import Self +from xxhash import xxh3_128_hexdigest from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointMetadata from langgraph.utils.fields import get_update_as_tuples @@ -146,10 +145,9 @@ class Interrupt: @property def interrupt_id(self) -> str: """Generate a unique ID for the interrupt based on its namespace.""" - identifier = ( - uuid.uuid4().bytes if self.ns is None else "".join(self.ns).encode() - ) - return hashlib.sha256(identifier).hexdigest() + if self.ns is None: + return "placeholder-id" + return xxh3_128_hexdigest("".join(self.ns).encode()) class StateUpdate(NamedTuple): @@ -486,7 +484,6 @@ def interrupt(value: Any) -> Any: GraphInterrupt: On the first invocation within the node, halts execution and surfaces the provided value to the client. """ from langgraph.constants import ( - CONF, CONFIG_KEY_CHECKPOINT_NS, CONFIG_KEY_SCRATCHPAD, CONFIG_KEY_SEND, diff --git a/libs/langgraph/tests/test_large_cases.py b/libs/langgraph/tests/test_large_cases.py index 3c0e7da87..074978876 100644 --- a/libs/langgraph/tests/test_large_cases.py +++ b/libs/langgraph/tests/test_large_cases.py @@ -5666,6 +5666,9 @@ def test_dynamic_interrupt( ) == { "my_key": "value", "market": "DE", + "__interrupt__": [ + Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")]) + ], } assert tool_two_node_count == 1, "interrupts aren't retried" assert len(tracer.runs) == 1 @@ -5712,6 +5715,9 @@ def test_dynamic_interrupt( assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == { "my_key": "value ⛰️", "market": "DE", + "__interrupt__": [ + Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")]) + ], } if "shallow" not in checkpointer_name: @@ -5840,6 +5846,9 @@ def test_copy_checkpoint( ) == { "my_key": "value one", "market": "DE", + "__interrupt__": [ + Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")]) + ], } assert tool_two_node_count == 1, "interrupts aren't retried" assert len(tracer.runs) == 1 @@ -5890,6 +5899,9 @@ def test_copy_checkpoint( assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == { "my_key": "value ⛰️ one", "market": "DE", + "__interrupt__": [ + Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")]) + ], } if "shallow" not in checkpointer_name: assert [c.metadata for c in tool_two.checkpointer.list(thread1)] == [ @@ -6040,6 +6052,13 @@ def test_dynamic_interrupt_subgraph( ) == { "my_key": "value", "market": "DE", + "__interrupt__": [ + Interrupt( + value="Just because...", + resumable=True, + ns=[AnyStr("tool_two:"), AnyStr("do:")], + ) + ], } assert tool_two_node_count == 1, "interrupts aren't retried" assert len(tracer.runs) == 1 @@ -6086,6 +6105,13 @@ def test_dynamic_interrupt_subgraph( assert tool_two.invoke({"my_key": "value ⛰️", "market": "DE"}, thread1) == { "my_key": "value ⛰️", "market": "DE", + "__interrupt__": [ + Interrupt( + value="Just because...", + resumable=True, + ns=[AnyStr("tool_two:"), AnyStr("do:")], + ) + ], } if "shallow" not in checkpointer_name: @@ -7318,15 +7344,15 @@ def test_send_dedupe_on_resume( graph = builder.compile(checkpointer=checkpointer) thread1 = {"configurable": {"thread_id": "1"}} - assert graph.invoke(["0"], thread1, checkpoint_during=checkpoint_during) == [ - "0", - "1", - "3.1", - "2|Command(goto=Send(node='2', arg=3))", - "2|Command(goto=Send(node='flaky', arg=4))", - "3", - "2|3", - ] + assert graph.invoke(["0"], thread1, checkpoint_during=checkpoint_during) == { + "__interrupt__": [ + Interrupt( + value="Bahh", + resumable=False, + ns=None, + ), + ], + } assert builder.nodes["2"].runnable.func.ticks == 3 assert builder.nodes["flaky"].runnable.func.ticks == 1 # check state diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index ce6474ca1..316d44817 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -3703,7 +3703,16 @@ def test_subgraph_checkpoint_true_interrupt( assert graph.invoke( {"foo": "foo"}, config, checkpoint_during=checkpoint_during - ) == {"foo": "hi! foo"} + ) == { + "foo": "hi! foo", + "__interrupt__": [ + Interrupt( + value="Provide baz value", + resumable=True, + ns=[AnyStr("node_2"), AnyStr("subgraph_node_1:")], + ) + ], + } assert graph.get_state(config, subgraphs=True).tasks[0].state.values == { "bar": "hi! foo" } @@ -5444,7 +5453,7 @@ def test_interrupt_task_functional( config = {"configurable": {"thread_id": "1"}} # First run, interrupted at bar - assert not graph.invoke({"a": ""}, config) + graph.invoke({"a": ""}, config) # Resume with an answer res = graph.invoke(Command(resume="bar"), config) assert res == {"a": "foobar"} @@ -7334,10 +7343,27 @@ def test_interrupt_subgraph_reenter_checkpointer_true( ) config = {"configurable": {"thread_id": "1"}} - assert parent.invoke({"foo": "", "counter": 0}, config) == {"foo": "", "counter": 0} + assert parent.invoke({"foo": "", "counter": 0}, config) == { + "foo": "", + "counter": 0, + "__interrupt__": [ + Interrupt( + value="Provide value", + resumable=True, + ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")], + ) + ], + } assert parent.invoke(Command(resume="bar"), config) == { "foo": "subgraph_2", "counter": 1, + "__interrupt__": [ + Interrupt( + value="Provide value", + resumable=True, + ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")], + ) + ], } assert parent.invoke(Command(resume="qux"), config) == { "foo": "subgraph_2|parent", @@ -7362,6 +7388,13 @@ def test_interrupt_subgraph_reenter_checkpointer_true( assert parent.invoke({"foo": "meow", "counter": 0}, config) == { "foo": "meow", "counter": 0, + "__interrupt__": [ + Interrupt( + value="Provide value", + resumable=True, + ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")], + ) + ], } # confirm that we preserve the state values from the previous invocation assert bar_values == [None, "barbaz", "quxbaz"] diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index 1f6b25574..77f530aa4 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -531,6 +531,9 @@ async def test_dynamic_interrupt(checkpointer_name: str) -> None: ) == { "my_key": "value", "market": "DE", + "__interrupt__": [ + Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")]) + ], } assert tool_two_node_count == 1, "interrupts aren't retried" assert len(tracer.runs) == 1 @@ -713,6 +716,13 @@ async def test_dynamic_interrupt_subgraph(checkpointer_name: str) -> None: ) == { "my_key": "value", "market": "DE", + "__interrupt__": [ + Interrupt( + value="Just because...", + resumable=True, + ns=[AnyStr("tool_two:"), AnyStr("do:")], + ) + ], } assert tool_two_node_count == 1, "interrupts aren't retried" assert len(tracer.runs) == 1 @@ -903,6 +913,9 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None: ) == { "my_key": "value one", "market": "DE", + "__interrupt__": [ + Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")]) + ], } assert tool_two_node_count == 1, "interrupts aren't retried" assert len(tracer.runs) == 1 @@ -964,6 +977,13 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None: ) == { "my_key": "value ⛰️ one", "market": "DE", + "__interrupt__": [ + Interrupt( + value="Just because...", + resumable=True, + ns=[AnyStr("tool_two:")], + ) + ], } if "shallow" not in checkpointer_name: @@ -1108,13 +1128,29 @@ async def test_node_not_cancelled_on_other_node_interrupted( # writes from "awhile" are applied to last chunk assert await graph.ainvoke({"hello": "world"}, thread) == { - "hello": "world again" + "hello": "world again", + "__interrupt__": [ + Interrupt( + value="I am bad", + resumable=True, + ns=[AnyStr("bad:")], + ) + ], } assert not inner_task_cancelled assert awhiles == 1 - assert await graph.ainvoke(None, thread, debug=True) == {"hello": "world again"} + assert await graph.ainvoke(None, thread, debug=True) == { + "hello": "world again", + "__interrupt__": [ + Interrupt( + value="I am bad", + resumable=True, + ns=[AnyStr("bad:")], + ) + ], + } assert not inner_task_cancelled assert awhiles == 1 @@ -2795,15 +2831,15 @@ async def test_send_dedupe_on_resume( thread1 = {"configurable": {"thread_id": "1"}} assert await graph.ainvoke( ["0"], thread1, checkpoint_during=checkpoint_during - ) == [ - "0", - "1", - "3.1", - "2|Command(goto=Send(node='2', arg=3))", - "2|Command(goto=Send(node='flaky', arg=4))", - "3", - "2|3", - ] + ) == { + "__interrupt__": [ + Interrupt( + value="Bahh", + resumable=False, + ns=None, + ), + ], + } assert builder.nodes["2"].runnable.func.ticks == 3 assert builder.nodes["flaky"].runnable.func.ticks == 1 # resume execution @@ -5555,7 +5591,16 @@ async def test_subgraph_checkpoint_true_interrupt( assert await graph.ainvoke( {"foo": "foo"}, config, checkpoint_during=checkpoint_during - ) == {"foo": "hi! foo"} + ) == { + "foo": "hi! foo", + "__interrupt__": [ + Interrupt( + value="Provide baz value", + resumable=True, + ns=[AnyStr("node_2"), AnyStr("subgraph_node_1:")], + ) + ], + } assert (await graph.aget_state(config, subgraphs=True)).tasks[ 0 ].state.values == {"bar": "hi! foo"} @@ -8104,6 +8149,13 @@ async def test_interrupt_subgraph_reenter_checkpointer_true( assert await parent.ainvoke({"foo": "", "counter": 0}, config) == { "foo": "", "counter": 0, + "__interrupt__": [ + Interrupt( + value="Provide value", + resumable=True, + ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")], + ) + ], } assert await parent.ainvoke(Command(resume="bar"), config) == { "foo": "subgraph_2", @@ -8132,6 +8184,13 @@ async def test_interrupt_subgraph_reenter_checkpointer_true( assert await parent.ainvoke({"foo": "meow", "counter": 0}, config) == { "foo": "meow", "counter": 0, + "__interrupt__": [ + Interrupt( + value="Provide value", + resumable=True, + ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")], + ) + ], } # confirm that we preserve the state values from the previous invocation assert bar_values == [None, "barbaz", "quxbaz"]