diff --git a/docs/docs/cloud/how-tos/add-human-in-the-loop.md b/docs/docs/cloud/how-tos/add-human-in-the-loop.md index ecdd2004c..7077d01b4 100644 --- a/docs/docs/cloud/how-tos/add-human-in-the-loop.md +++ b/docs/docs/cloud/how-tos/add-human-in-the-loop.md @@ -30,9 +30,7 @@ To review, edit, and approve tool calls in an agent or workflow, use LangGraph's # > [ # > { # > 'value': {'text_to_revise': 'original text'}, - # > 'resumable': True, - # > 'ns': ['human_node:fc722478-2f21-0578-c572-d9fc4dd07c3b'], - # > 'when': 'during' + # > 'id': '...', # > } # > ] @@ -203,9 +201,7 @@ To review, edit, and approve tool calls in an agent or workflow, use LangGraph's # > [ # > { # > 'value': {'text_to_revise': 'original text'}, - # > 'resumable': True, - # > 'ns': ['human_node:fc722478-2f21-0578-c572-d9fc4dd07c3b'], - # > 'when': 'during' + # > 'id': '...', # > } # > ] diff --git a/docs/docs/concepts/functional_api.md b/docs/docs/concepts/functional_api.md index a0c910139..c81511f71 100644 --- a/docs/docs/concepts/functional_api.md +++ b/docs/docs/concepts/functional_api.md @@ -79,51 +79,55 @@ def workflow(topic: str) -> dict: ```python import time import uuid - from langgraph.func import entrypoint, task from langgraph.types import interrupt from langgraph.checkpoint.memory import MemorySaver + @task def write_essay(topic: str) -> str: """Write an essay about the given topic.""" - time.sleep(1) # This is a placeholder for a long-running task. + time.sleep(1) # This is a placeholder for a long-running task. return f"An essay about topic: {topic}" + @entrypoint(checkpointer=MemorySaver()) def workflow(topic: str) -> dict: """A simple workflow that writes an essay and asks for a review.""" essay = write_essay("cat").result() - is_approved = interrupt({ - # Any json-serializable payload provided to interrupt as argument. - # It will be surfaced on the client side as an Interrupt when streaming data - # from the workflow. - "essay": essay, # The essay we want reviewed. - # We can add any additional information that we need. - # For example, introduce a key called "action" with some instructions. - "action": "Please approve/reject the essay", - }) - + is_approved = interrupt( + { + # Any json-serializable payload provided to interrupt as argument. + # It will be surfaced on the client side as an Interrupt when streaming data + # from the workflow. + "essay": essay, # The essay we want reviewed. + # We can add any additional information that we need. + # For example, introduce a key called "action" with some instructions. + "action": "Please approve/reject the essay", + } + ) return { - "essay": essay, # The essay that was generated - "is_approved": is_approved, # Response from HIL + "essay": essay, # The essay that was generated + "is_approved": is_approved, # Response from HIL } + thread_id = str(uuid.uuid4()) - - config = { - "configurable": { - "thread_id": thread_id - } - } - + config = {"configurable": {"thread_id": thread_id}} for item in workflow.stream("cat", config): print(item) - ``` - - ```pycon - {'write_essay': 'An essay about topic: cat'} - {'__interrupt__': (Interrupt(value={'essay': 'An essay about topic: cat', 'action': 'Please approve/reject the essay'}, resumable=True, ns=['workflow:f7b8508b-21c0-8b4c-5958-4e8de74d2684'], when='during'),)} + # > {'write_essay': 'An essay about topic: cat'} + # > { + # > '__interrupt__': ( + # > Interrupt( + # > value={ + # > 'essay': 'An essay about topic: cat', + # > 'action': 'Please approve/reject the essay' + # > }, + # > id='b9b2b9d788f482663ced6dc755c9e981' + # > ), + # > ) + # > } ``` An essay has been written and is ready for review. Once the review is provided, we can resume the workflow: diff --git a/docs/docs/how-tos/human_in_the_loop/add-human-in-the-loop.md b/docs/docs/how-tos/human_in_the_loop/add-human-in-the-loop.md index 01166b008..b1d9c7aea 100644 --- a/docs/docs/how-tos/human_in_the_loop/add-human-in-the-loop.md +++ b/docs/docs/how-tos/human_in_the_loop/add-human-in-the-loop.md @@ -46,13 +46,7 @@ graph = graph_builder.compile(checkpointer=checkpointer) # (4)! config = {"configurable": {"thread_id": "some_id"}} result = graph.invoke({"some_text": "original text"}, config=config) # (5)! print(result['__interrupt__']) # (6)! -# > [ -# > Interrupt( -# > value={'text_to_revise': 'original text'}, -# > resumable=True, -# > ns=['human_node:6ce9e64f-edef-fe5d-f7dc-511fa9526960'] -# > ) -# > ] +# > [Interrupt(value={'text_to_revise': 'original text'}, id='a0d9dd40440ac7be2720dc5c20858627')] # highlight-next-line print(graph.invoke(Command(resume="Edited text"), config=config)) # (7)! @@ -72,25 +66,27 @@ print(graph.invoke(Command(resume="Edited text"), config=config)) # (7)! ```python from typing import TypedDict import uuid - from langgraph.checkpoint.memory import InMemorySaver from langgraph.constants import START from langgraph.graph import StateGraph + # highlight-next-line from langgraph.types import interrupt, Command + class State(TypedDict): some_text: str + def human_node(state: State): # highlight-next-line - value = interrupt( # (1)! + value = interrupt( # (1)! { - "text_to_revise": state["some_text"] # (2)! + "text_to_revise": state["some_text"] # (2)! } ) return { - "some_text": value # (3)! + "some_text": value # (3)! } @@ -98,25 +94,15 @@ print(graph.invoke(Command(resume="Edited text"), config=config)) # (7)! graph_builder = StateGraph(State) graph_builder.add_node("human_node", human_node) graph_builder.add_edge(START, "human_node") - - checkpointer = InMemorySaver() # (4)! - + checkpointer = InMemorySaver() # (4)! graph = graph_builder.compile(checkpointer=checkpointer) - # Pass a thread ID to the graph to run it. config = {"configurable": {"thread_id": uuid.uuid4()}} - # Run the graph until the interrupt is hit. - result = graph.invoke({"some_text": "original text"}, config=config) # (5)! + result = graph.invoke({"some_text": "original text"}, config=config) # (5)! - print(result['__interrupt__']) # (6)! - # > [ - # > Interrupt( - # > value={'text_to_revise': 'original text'}, - # > resumable=True, - # > ns=['human_node:6ce9e64f-edef-fe5d-f7dc-511fa9526960'] - # > ) - # > ] + print(result["__interrupt__"]) # (6)! + # > [Interrupt(value={'text_to_revise': 'original text'}, id='6d7c4048049254c83195429a3659661d')] # highlight-next-line print(graph.invoke(Command(resume="Edited text"), config=config)) # (7)! @@ -164,7 +150,7 @@ For example, once your graph has been interrupted (multiple times, theoretically ```python resume_map = { - i.interrupt_id: f"human input for prompt {i.value}" + i.id: f"human input for prompt {i.value}" for i in parent.get_state(thread_config).interrupts } @@ -385,14 +371,15 @@ graph.invoke( # Output interrupt payload print(result["__interrupt__"]) # Example output: - # Interrupt( - # value={ - # 'task': 'Please review and edit the generated summary if necessary.', - # 'generated_summary': 'The cat sat on the mat and looked at the stars.' - # }, - # resumable=True, - # ... - # ) + # > [ + # > Interrupt( + # > value={ + # > 'task': 'Please review and edit the generated summary if necessary.', + # > 'generated_summary': 'The cat sat on the mat and looked at the stars.' + # > }, + # > id='...' + # > ) + # > ] # Resume the graph with human-edited input edited_summary = "The cat lay on the rug, gazing peacefully at the night sky." @@ -873,7 +860,7 @@ def node_in_parent_graph(state: State): Entered `parent_node` a total of 1 times Entered `node_in_subgraph` a total of 1 times Entered human_node in sub-graph a total of 1 times - {'__interrupt__': (Interrupt(value='what is your name?', resumable=True, ns=['parent_node:4c3a0248-21f0-1287-eacf-3002bc304db4', 'human_node:2fe86d52-6f70-2a3f-6b2f-b1eededd6348'], when='during'),)} + {'__interrupt__': (Interrupt(value='what is your name?', id='...'),)} --- Resuming --- Entered `parent_node` a total of 2 times Entered human_node in sub-graph a total of 2 times @@ -949,7 +936,7 @@ To avoid issues, refrain from dynamically changing the node's structure between ``` ```pycon - {'__interrupt__': (Interrupt(value='what is your name?', resumable=True, ns=['human_node:3a007ef9-c30d-c357-1ec1-86a1a70d8fba'], when='during'),)} + {'__interrupt__': (Interrupt(value='what is your name?', id='...'),)} Name: N/A. Age: John {'human_node': {'age': 'John', 'name': 'N/A'}} ``` diff --git a/libs/langgraph/langgraph/errors.py b/libs/langgraph/langgraph/errors.py index e1121c1bf..841abbaf7 100644 --- a/libs/langgraph/langgraph/errors.py +++ b/libs/langgraph/langgraph/errors.py @@ -1,10 +1,16 @@ +from __future__ import annotations + from collections.abc import Sequence from enum import Enum from typing import Any +from warnings import warn + +from typing_extensions import deprecated # EmptyChannelError is re-exported from langgraph.channels.base from langgraph.checkpoint.base import EmptyChannelError # noqa: F401 from langgraph.types import Command, Interrupt +from langgraph.warnings import LangGraphDeprecatedSinceV10 __all__ = ( "EmptyChannelError", @@ -83,11 +89,26 @@ class GraphInterrupt(GraphBubbleUp): super().__init__(interrupts) +@deprecated( + "NodeInterrupt is deprecated. Please use `langgraph.types.interrupt` instead.", + stacklevel=2, +) class NodeInterrupt(GraphInterrupt): - """Raised by a node to interrupt execution.""" + """Raised by a node to interrupt execution. - def __init__(self, value: Any) -> None: - super().__init__([Interrupt(value=value)]) + Deprecated in V1.0.0 in favor of [`interrupt`][langgraph.types.interrupt]. + """ + + def __init__(self, value: Any, id: str | None = None) -> None: + warn( + "NodeInterrupt is deprecated. Please use `langgraph.types.interrupt` instead.", + LangGraphDeprecatedSinceV10, + stacklevel=2, + ) + if id is None: + super().__init__([Interrupt(value=value)]) + else: + super().__init__([Interrupt(value=value, id=id)]) class ParentCommand(GraphBubbleUp): diff --git a/libs/langgraph/langgraph/pregel/remote.py b/libs/langgraph/langgraph/pregel/remote.py index 145dd548f..c5e0e8dcd 100644 --- a/libs/langgraph/langgraph/pregel/remote.py +++ b/libs/langgraph/langgraph/pregel/remote.py @@ -260,9 +260,9 @@ class RemoteGraph(PregelProtocol): def _create_state_snapshot(self, state: ThreadState) -> StateSnapshot: tasks: list[PregelTask] = [] for task in state["tasks"]: - interrupts = [] - for interrupt in task["interrupts"]: - interrupts.append(Interrupt(**interrupt)) + interrupts = tuple( + Interrupt(**interrupt) for interrupt in task["interrupts"] + ) tasks.append( PregelTask( @@ -270,7 +270,7 @@ class RemoteGraph(PregelProtocol): name=task["name"], path=tuple(), error=Exception(task["error"]) if task["error"] else None, - interrupts=tuple(interrupts), + interrupts=interrupts, state=( self._create_state_snapshot(task["state"]) if task["state"] diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index 80ac19176..11016c536 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -14,16 +14,20 @@ from typing import ( NamedTuple, TypeVar, Union, - cast, + final, ) +from warnings import warn from langchain_core.runnables import Runnable, RunnableConfig +from typing_extensions import Unpack, deprecated from xxhash import xxh3_128_hexdigest from langgraph._internal._cache import default_cache_key from langgraph._internal._fields import get_cached_annotated_keys, get_update_as_tuples from langgraph._internal._retry import default_retry_on +from langgraph._internal._typing import UNSET, DeprecatedKwargs from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointMetadata +from langgraph.warnings import LangGraphDeprecatedSinceV10 if TYPE_CHECKING: from langgraph.pregel.protocol import PregelProtocol @@ -86,8 +90,10 @@ Always injected into nodes if requested as a keyword argument, but it's a no-op when not using stream_mode="custom".""" if sys.version_info >= (3, 10): + _DC_SLOTS = {"slots": True} _DC_KWARGS = {"kw_only": True, "slots": True, "frozen": True} else: + _DC_SLOTS = {} _DC_KWARGS = {"frozen": True} @@ -128,7 +134,11 @@ class CachePolicy(Generic[KeyFuncT]): """Time to live for the cache entry in seconds. If None, the entry never expires.""" -@dataclasses.dataclass(**_DC_KWARGS) +_DEFAULT_INTERRUPT_ID = "placeholder-id" + + +@final +@dataclasses.dataclass(init=False, **_DC_SLOTS) class Interrupt: """Information about an interrupt that occurred in a node. @@ -136,16 +146,41 @@ class Interrupt: """ value: Any - resumable: bool = False - ns: Sequence[str] | None = None - when: Literal["during"] = dataclasses.field(default="during", repr=False) + id: str + + def __init__( + self, + value: Any, + id: str = _DEFAULT_INTERRUPT_ID, + **deprecated_kwargs: Unpack[DeprecatedKwargs], + ) -> None: + self.value = value + + if ( + (ns := deprecated_kwargs.get("ns", UNSET)) is not UNSET + and (id == _DEFAULT_INTERRUPT_ID) + and (isinstance(ns, Sequence)) + ): + self.id = xxh3_128_hexdigest("|".join(ns).encode()) + else: + self.id = id + + @classmethod + def from_ns(cls, value: Any, ns: str) -> Interrupt: + return cls(value=value, id=xxh3_128_hexdigest(ns.encode())) @property + @deprecated( + "`interrupt_id` is deprecated. Use `id` instead.", + stacklevel=2, + ) def interrupt_id(self) -> str: - """Generate a unique ID for the interrupt based on its namespace.""" - if self.ns is None: - return "placeholder-id" - return xxh3_128_hexdigest("|".join(self.ns).encode()) + warn( + "`interrupt_id` is deprecated. Use `id` instead.", + LangGraphDeprecatedSinceV10, + stacklevel=2, + ) + return self.id class StateUpdate(NamedTuple): @@ -419,22 +454,16 @@ def interrupt(value: Any) -> Any: for chunk in graph.stream({\"foo\": \"abc\"}, config): print(chunk) - ``` - ```pycon - {'__interrupt__': (Interrupt(value='what is your age?', resumable=True, ns=['node:62e598fa-8653-9d6d-2046-a70203020e37'], when='during'),)} - ``` + # > {'__interrupt__': (Interrupt(value='what is your age?', id='45fda8478b2ef754419799e10992af06'),)} - ```python command = Command(resume=\"some input from a human!!!\") for chunk in graph.stream(Command(resume=\"some input from a human!!!\"), config): print(chunk) - ``` - ```pycon - Received an input from the interrupt: some input from a human!!! - {'node': {'human_value': 'some input from a human!!!'}} + # > Received an input from the interrupt: some input from a human!!! + # > {'node': {'human_value': 'some input from a human!!!'}} ``` Args: @@ -451,7 +480,6 @@ def interrupt(value: Any) -> Any: CONFIG_KEY_CHECKPOINT_NS, CONFIG_KEY_SCRATCHPAD, CONFIG_KEY_SEND, - NS_SEP, RESUME, ) from langgraph.errors import GraphInterrupt @@ -474,10 +502,9 @@ def interrupt(value: Any) -> Any: # no resume value found raise GraphInterrupt( ( - Interrupt( + Interrupt.from_ns( value=value, - resumable=True, - ns=cast(str, conf[CONFIG_KEY_CHECKPOINT_NS]).split(NS_SEP), + ns=conf[CONFIG_KEY_CHECKPOINT_NS], ), ) ) diff --git a/libs/langgraph/tests/test_checkpoint_migration.py b/libs/langgraph/tests/test_checkpoint_migration.py index d1dd65e61..9fa6d7421 100644 --- a/libs/langgraph/tests/test_checkpoint_migration.py +++ b/libs/langgraph/tests/test_checkpoint_migration.py @@ -92,8 +92,7 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]: else ( Interrupt( value="", - resumable=True, - ns=[AnyStr("qa:")], + id=AnyStr(), ), ), state=None, @@ -107,8 +106,7 @@ def get_expected_history(*, exc_task_results: int = 0) -> list[StateSnapshot]: else ( Interrupt( value="", - resumable=True, - ns=[AnyStr("qa:")], + id=AnyStr(), ), ), ), @@ -412,8 +410,8 @@ SAVED_CHECKPOINTS = { [ Interrupt( value="", - resumable=True, - ns=["qa:2430f303-da9f-2e3e-738c-2e8ea28e8973"], + resumable=True, # type: ignore[arg-type] + ns=["qa:2430f303-da9f-2e3e-738c-2e8ea28e8973"], # type: ignore[arg-type] ) ], ), @@ -786,8 +784,8 @@ SAVED_CHECKPOINTS = { [ Interrupt( value="", - resumable=True, - ns=["qa:4ee8637e-0a95-285e-75bc-4da721c0beab"], + resumable=True, # type: ignore[arg-type] + ns=["qa:4ee8637e-0a95-285e-75bc-4da721c0beab"], # type: ignore[arg-type] ) ], ), @@ -1173,7 +1171,7 @@ SAVED_CHECKPOINTS = { Interrupt( value="", resumable=True, - ns=["qa:369e94b1-77d1-d67a-ab59-23d1ba20ee73"], + ns=["qa:369e94b1-77d1-d67a-ab59-23d1ba20ee73"], # type: ignore[arg-type] ) ], ), @@ -1525,8 +1523,7 @@ def test_latest_checkpoint_state_graph( "__interrupt__": ( Interrupt( value="", - resumable=True, - ns=[AnyStr("qa:")], + id=AnyStr(), ), ) }, @@ -1570,8 +1567,7 @@ async def test_latest_checkpoint_state_graph_async( "__interrupt__": ( Interrupt( value="", - resumable=True, - ns=[AnyStr("qa:")], + id=AnyStr(), ), ) }, diff --git a/libs/langgraph/tests/test_deprecation.py b/libs/langgraph/tests/test_deprecation.py index 0edb00b6f..bd592f922 100644 --- a/libs/langgraph/tests/test_deprecation.py +++ b/libs/langgraph/tests/test_deprecation.py @@ -1,9 +1,10 @@ import pytest from typing_extensions import TypedDict +from langgraph.errors import NodeInterrupt from langgraph.func import entrypoint, task from langgraph.graph import StateGraph -from langgraph.types import RetryPolicy +from langgraph.types import Interrupt, RetryPolicy from langgraph.warnings import LangGraphDeprecatedSinceV05, LangGraphDeprecatedSinceV10 @@ -88,3 +89,21 @@ def test_pregel_deprecation() -> None: match="Importing from langgraph.pregel.types is deprecated. Please use 'from langgraph.types import ...' instead.", ): from langgraph.pregel.types import StateSnapshot # noqa: F401 + + +def test_interrupt_attributes_deprecation() -> None: + interrupt = Interrupt(value="question", id="abc") + + with pytest.warns( + LangGraphDeprecatedSinceV10, + match="`interrupt_id` is deprecated. Use `id` instead.", + ): + interrupt.interrupt_id + + +def test_node_interrupt_deprecation() -> None: + with pytest.warns( + LangGraphDeprecatedSinceV10, + match="NodeInterrupt is deprecated. Please use `langgraph.types.interrupt` instead.", + ): + NodeInterrupt(value="test") diff --git a/libs/langgraph/tests/test_interrupt_migration.py b/libs/langgraph/tests/test_interrupt_migration.py new file mode 100644 index 000000000..8149e0f85 --- /dev/null +++ b/libs/langgraph/tests/test_interrupt_migration.py @@ -0,0 +1,50 @@ +import warnings + +import pytest + +from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer +from langgraph.types import Interrupt +from langgraph.warnings import LangGraphDeprecatedSinceV10 + + +@pytest.mark.filterwarnings("ignore:LangGraphDeprecatedSinceV10") +def test_interrupt_legacy_ns() -> None: + with warnings.catch_warnings(): + warnings.filterwarnings("ignore", category=LangGraphDeprecatedSinceV10) + + old_interrupt = Interrupt( + value="abc", resumable=True, when="during", ns=["a:b", "c:d"] + ) + + new_interrupt = Interrupt.from_ns(value="abc", ns="a:b|c:d") + assert new_interrupt.value == old_interrupt.value + assert new_interrupt.id == old_interrupt.id + + +serializer = JsonPlusSerializer() + + +def test_serialization_roundtrip() -> None: + """Test that the legacy interrupt (pre v1) can be reserialized as the modern interrupt without id corruption.""" + + # generated with: + # JsonPlusSerializer().dumps(Interrupt(value="legacy_test", ns=["legacy_test"], resumable=True, when="during")) + legacy_interrupt_bytes = b'{"lc": 2, "type": "constructor", "id": ["langgraph", "types", "Interrupt"], "kwargs": {"value": "legacy_test", "resumable": true, "ns": ["legacy_test"], "when": "during"}}' + legacy_interrupt_id = "f1fa625689ec006a5b32b76863e22a6c" + + interrupt = serializer.loads(legacy_interrupt_bytes) + assert interrupt.id == legacy_interrupt_id + assert interrupt.value == "legacy_test" + + +def test_serialization_roundtrip_complex_ns() -> None: + """Test that the legacy interrupt (pre v1), with a more complex ns can be reserialized as the modern interrupt without id corruption.""" + + # generated with: + # JsonPlusSerializer().dumps(Interrupt(value="legacy_test", ns=["legacy:test", "with:complex", "name:space"], resumable=True, when="during")) + legacy_interrupt_bytes = b'{"lc": 2, "type": "constructor", "id": ["langgraph", "types", "Interrupt"], "kwargs": {"value": "legacy_test", "resumable": true, "ns": ["legacy:test", "with:complex", "name:space"], "when": "during"}}' + legacy_interrupt_id = "e69356a9ee3630ee7f4f597f2693000c" + + interrupt = serializer.loads(legacy_interrupt_bytes) + assert interrupt.id == legacy_interrupt_id + assert interrupt.value == "legacy_test" diff --git a/libs/langgraph/tests/test_large_cases.py b/libs/langgraph/tests/test_large_cases.py index e642d07da..419686255 100644 --- a/libs/langgraph/tests/test_large_cases.py +++ b/libs/langgraph/tests/test_large_cases.py @@ -16,7 +16,6 @@ from langgraph.channels.untracked_value import UntrackedValue from langgraph.checkpoint.base import BaseCheckpointSaver from langgraph.checkpoint.memory import InMemorySaver from langgraph.constants import END, PULL, PUSH, START -from langgraph.errors import NodeInterrupt from langgraph.graph import StateGraph from langgraph.graph.message import MessageGraph, MessagesState, add_messages from langgraph.prebuilt.chat_agent_executor import create_react_agent @@ -4173,9 +4172,7 @@ def test_dynamic_interrupt(sync_checkpointer: BaseCheckpointSaver) -> None: ) == { "my_key": "value", "market": "DE", - "__interrupt__": [ - Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")]) - ], + "__interrupt__": [Interrupt(value="Just because...", id=AnyStr())], } assert tool_two_node_count == 1, "interrupts aren't retried" assert len(tracer.runs) == 1 @@ -4205,8 +4202,7 @@ def test_dynamic_interrupt(sync_checkpointer: BaseCheckpointSaver) -> None: "__interrupt__": ( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ) }, @@ -4224,9 +4220,7 @@ def test_dynamic_interrupt(sync_checkpointer: BaseCheckpointSaver) -> None: ) == { "my_key": "value ⛰️", "market": "DE", - "__interrupt__": [ - Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")]) - ], + "__interrupt__": [Interrupt(value="Just because...", id=AnyStr())], } assert [c.metadata for c in tool_two.checkpointer.list(thread1)] == [ @@ -4248,8 +4242,7 @@ def test_dynamic_interrupt(sync_checkpointer: BaseCheckpointSaver) -> None: interrupts=( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ), ), @@ -4271,8 +4264,7 @@ def test_dynamic_interrupt(sync_checkpointer: BaseCheckpointSaver) -> None: interrupts=( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ), ) @@ -4336,9 +4328,7 @@ def test_partial_pending_checkpoint(sync_checkpointer: BaseCheckpointSaver) -> N ) == { "my_key": "value one", "market": "DE", - "__interrupt__": [ - Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")]) - ], + "__interrupt__": [Interrupt(value="Just because...", id=AnyStr())], } assert tool_two_node_count == 1, "interrupts aren't retried" assert len(tracer.runs) == 1 @@ -4371,8 +4361,7 @@ def test_partial_pending_checkpoint(sync_checkpointer: BaseCheckpointSaver) -> N "__interrupt__": ( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ) }, @@ -4391,9 +4380,7 @@ def test_partial_pending_checkpoint(sync_checkpointer: BaseCheckpointSaver) -> N ) == { "my_key": "value ⛰️ one", "market": "DE", - "__interrupt__": [ - Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")]) - ], + "__interrupt__": [Interrupt(value="Just because...", id=AnyStr())], } assert [c.metadata for c in tool_two.checkpointer.list(thread1)] == [ { @@ -4420,8 +4407,7 @@ def test_partial_pending_checkpoint(sync_checkpointer: BaseCheckpointSaver) -> N interrupts=( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ), ), @@ -4443,8 +4429,7 @@ def test_partial_pending_checkpoint(sync_checkpointer: BaseCheckpointSaver) -> N interrupts=( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ), ) @@ -4513,8 +4498,7 @@ def test_dynamic_interrupt_subgraph(sync_checkpointer: BaseCheckpointSaver) -> N "__interrupt__": [ Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:"), AnyStr("do:")], + id=AnyStr(), ) ], } @@ -4546,8 +4530,7 @@ def test_dynamic_interrupt_subgraph(sync_checkpointer: BaseCheckpointSaver) -> N "__interrupt__": ( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:"), AnyStr("do:")], + id=AnyStr(), ), ) }, @@ -4568,8 +4551,7 @@ def test_dynamic_interrupt_subgraph(sync_checkpointer: BaseCheckpointSaver) -> N "__interrupt__": [ Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:"), AnyStr("do:")], + id=AnyStr(), ) ], } @@ -4598,8 +4580,7 @@ def test_dynamic_interrupt_subgraph(sync_checkpointer: BaseCheckpointSaver) -> N interrupts=( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:"), AnyStr("do:")], + id=AnyStr(), ), ), state={ @@ -4627,8 +4608,7 @@ def test_dynamic_interrupt_subgraph(sync_checkpointer: BaseCheckpointSaver) -> N interrupts=( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:"), AnyStr("do:")], + id=AnyStr(), ), ), ) @@ -4672,7 +4652,7 @@ def test_send_dedupe_on_resume( def __call__(self, state): self.ticks += 1 if self.ticks == 1: - raise NodeInterrupt("Bahh") + interrupt("Bahh") return ["|".join(("flaky", str(state)))] class Node: @@ -4722,8 +4702,7 @@ def test_send_dedupe_on_resume( "__interrupt__": [ Interrupt( value="Bahh", - resumable=False, - ns=None, + id=AnyStr(), ), ], } @@ -4884,7 +4863,7 @@ def test_send_dedupe_on_resume( name="flaky", path=("__pregel_push", 1, False), error=None, - interrupts=(Interrupt(value="Bahh", resumable=False, ns=None),), + interrupts=(Interrupt(value="Bahh", id=AnyStr()),), state=None, result=["flaky|4"] if checkpoint_during else None, ), @@ -4898,7 +4877,7 @@ def test_send_dedupe_on_resume( result=["3"], ), ), - interrupts=(Interrupt(value="Bahh", resumable=False, ns=None),), + interrupts=(Interrupt(value="Bahh", id=AnyStr()),), ), StateSnapshot( values=["0", "1"], diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index c82bec7ba..f89c727ae 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -1280,9 +1280,7 @@ def test_imp_task( "__interrupt__": ( Interrupt( value="question", - resumable=True, - ns=[AnyStr("graph:")], - when="during", + id=AnyStr(), ), ) }, @@ -1349,9 +1347,7 @@ def test_imp_nested( "__interrupt__": ( Interrupt( value="question", - resumable=True, - ns=[AnyStr("graph:")], - when="during", + id=AnyStr(), ), ) }, @@ -3371,8 +3367,7 @@ def test_subgraph_checkpoint_true_interrupt( "__interrupt__": [ Interrupt( value="Provide baz value", - resumable=True, - ns=[AnyStr("node_2"), AnyStr("subgraph_node_1:")], + id=AnyStr(), ) ], } @@ -4900,9 +4895,7 @@ def test_interrupt_multiple(sync_checkpointer: BaseCheckpointSaver): "__interrupt__": ( Interrupt( value={"value": 1}, - resumable=True, - ns=[AnyStr("node:")], - when="during", + id=AnyStr(), ), ) } @@ -4918,9 +4911,7 @@ def test_interrupt_multiple(sync_checkpointer: BaseCheckpointSaver): "__interrupt__": ( Interrupt( value={"value": 2}, - resumable=True, - ns=[AnyStr("node:")], - when="during", + id=AnyStr(), ), ) } @@ -4968,9 +4959,7 @@ def test_interrupt_loop(sync_checkpointer: BaseCheckpointSaver): "__interrupt__": ( Interrupt( value="How old are you?", - resumable=True, - ns=[AnyStr("node:")], - when="during", + id=AnyStr(), ), ) } @@ -4987,9 +4976,7 @@ def test_interrupt_loop(sync_checkpointer: BaseCheckpointSaver): "__interrupt__": ( Interrupt( value="invalid response", - resumable=True, - ns=[AnyStr("node:")], - when="during", + id=AnyStr(), ), ) } @@ -5006,9 +4993,7 @@ def test_interrupt_loop(sync_checkpointer: BaseCheckpointSaver): "__interrupt__": ( Interrupt( value="invalid response", - resumable=True, - ns=[AnyStr("node:")], - when="during", + id=AnyStr(), ), ) } @@ -5044,8 +5029,7 @@ def test_interrupt_functional( "__interrupt__": [ Interrupt( value="Provide value for bar:", - resumable=True, - ns=[AnyStr("graph:")], + id=AnyStr(), ) ] } @@ -5078,8 +5062,7 @@ def test_interrupt_task_functional( "__interrupt__": [ Interrupt( value="Provide value for bar:", - resumable=True, - ns=[AnyStr("graph:"), AnyStr("bar:")], + id=AnyStr(), ), ] } @@ -5102,8 +5085,7 @@ def test_interrupt_task_functional( "__interrupt__": [ Interrupt( value="Provide value for bar:", - resumable=True, - ns=[AnyStr("graph:"), AnyStr("bar:")], + id=AnyStr(), ), ] } @@ -5628,12 +5610,8 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver): "id": AnyStr(), "interrupts": [ { - "ns": [ - AnyStr(), - ], - "resumable": True, + "id": AnyStr(), "value": "test", - "when": "during", }, ], "name": "graph", @@ -5676,12 +5654,8 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver): "id": AnyStr(), "interrupts": ( { - "ns": [ - AnyStr(), - ], - "resumable": True, + "id": AnyStr(), "value": "test", - "when": "during", }, ), "name": "graph", @@ -5901,9 +5875,7 @@ def test_double_interrupt_subgraph(sync_checkpointer: BaseCheckpointSaver) -> No "__interrupt__": ( Interrupt( value="interrupt node 1", - resumable=True, - ns=[AnyStr("node_1:")], - when="during", + id=AnyStr(), ), ) }, @@ -5917,9 +5889,7 @@ def test_double_interrupt_subgraph(sync_checkpointer: BaseCheckpointSaver) -> No "__interrupt__": ( Interrupt( value="interrupt node 2", - resumable=True, - ns=[AnyStr("node_2:")], - when="during", + id=AnyStr(), ), ) }, @@ -5950,9 +5920,7 @@ def test_double_interrupt_subgraph(sync_checkpointer: BaseCheckpointSaver) -> No "__interrupt__": ( Interrupt( value="interrupt node 1", - resumable=True, - ns=[AnyStr("invoke_sub_agent:"), AnyStr("node_1:")], - when="during", + id=AnyStr(), ), ) }, @@ -5964,9 +5932,7 @@ def test_double_interrupt_subgraph(sync_checkpointer: BaseCheckpointSaver) -> No "__interrupt__": ( Interrupt( value="interrupt node 2", - resumable=True, - ns=[AnyStr("invoke_sub_agent:"), AnyStr("node_2:")], - when="during", + id=AnyStr(), ), ) } @@ -6045,7 +6011,7 @@ def test_multi_resume(sync_checkpointer: BaseCheckpointSaver) -> None: assert interrupt_values == set(prompts) resume_map: dict[str, str] = { - i.interrupt_id: f"human input for prompt {i.value}" + i.id: f"human input for prompt {i.value}" for i in parent_graph.get_state(thread_config).interrupts } @@ -7123,8 +7089,7 @@ def test_interrupt_subgraph_reenter_checkpointer_true( "__interrupt__": [ Interrupt( value="Provide value", - resumable=True, - ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")], + id=AnyStr(), ) ], } @@ -7134,8 +7099,7 @@ def test_interrupt_subgraph_reenter_checkpointer_true( "__interrupt__": [ Interrupt( value="Provide value", - resumable=True, - ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")], + id=AnyStr(), ) ], } @@ -7165,8 +7129,7 @@ def test_interrupt_subgraph_reenter_checkpointer_true( "__interrupt__": [ Interrupt( value="Provide value", - resumable=True, - ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")], + id=AnyStr(), ) ], } @@ -7312,7 +7275,7 @@ def test_parallel_interrupts(sync_checkpointer: BaseCheckpointSaver) -> None: # assume that it breaks here, because it is an interrupt # get human input and resume - if any(i.resumable for i in current_interrupts): + if len(current_interrupts) > 0: current_input = Command(resume=f"Resume #{invokes}") # not more human input required, must be completed @@ -7329,11 +7292,7 @@ def test_parallel_interrupts(sync_checkpointer: BaseCheckpointSaver) -> None: "__interrupt__": ( Interrupt( value="a", - resumable=True, - ns=[ - AnyStr("child_graph:"), - AnyStr("get_human_input:"), - ], + id=AnyStr(), ), ) }, @@ -7341,11 +7300,7 @@ def test_parallel_interrupts(sync_checkpointer: BaseCheckpointSaver) -> None: "__interrupt__": ( Interrupt( value="b", - resumable=True, - ns=[ - AnyStr("child_graph:"), - AnyStr("get_human_input:"), - ], + id=AnyStr(), ), ) }, @@ -7356,11 +7311,7 @@ def test_parallel_interrupts(sync_checkpointer: BaseCheckpointSaver) -> None: "__interrupt__": ( Interrupt( value="a", - resumable=True, - ns=[ - AnyStr("child_graph:"), - AnyStr("get_human_input:"), - ], + id=AnyStr(), ), ) }, @@ -7371,11 +7322,7 @@ def test_parallel_interrupts(sync_checkpointer: BaseCheckpointSaver) -> None: "__interrupt__": ( Interrupt( value="b", - resumable=True, - ns=[ - AnyStr("child_graph:"), - AnyStr("get_human_input:"), - ], + id=AnyStr(), ), ) }, @@ -7489,7 +7436,7 @@ def test_parallel_interrupts_double(sync_checkpointer: BaseCheckpointSaver) -> N # assume that it breaks here, because it is an interrupt # get human input and resume - if any(i.resumable for i in current_interrupts): + if len(current_interrupts) > 0: current_input = Command(resume=f"Resume #{invokes}") # not more human input required, must be completed diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index dc7a9c3d0..b2e99af23 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -45,7 +45,6 @@ from langgraph.constants import CONFIG_KEY_NODE_FINISHED, ERROR, PULL, START from langgraph.errors import ( GraphRecursionError, InvalidUpdateError, - NodeInterrupt, ParentCommand, ) from langgraph.func import entrypoint, task @@ -592,9 +591,7 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non ) == { "my_key": "value", "market": "DE", - "__interrupt__": [ - Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")]) - ], + "__interrupt__": [Interrupt(value="Just because...", id=AnyStr())], } assert tool_two_node_count == 1, "interrupts aren't retried" assert len(tracer.runs) == 1 @@ -625,8 +622,7 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non "__interrupt__": ( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ) }, @@ -651,8 +647,7 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non "__interrupt__": ( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ) }, @@ -676,8 +671,7 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non interrupts=( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ), ), @@ -693,8 +687,7 @@ async def test_dynamic_interrupt(async_checkpointer: BaseCheckpointSaver) -> Non interrupts=( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ), ) @@ -762,8 +755,7 @@ async def test_dynamic_interrupt_subgraph( "__interrupt__": [ Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:"), AnyStr("do:")], + id=AnyStr(), ) ], } @@ -796,8 +788,7 @@ async def test_dynamic_interrupt_subgraph( "__interrupt__": ( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:"), AnyStr("do:")], + id=AnyStr(), ), ) }, @@ -823,8 +814,7 @@ async def test_dynamic_interrupt_subgraph( "__interrupt__": ( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:"), AnyStr("do:")], + id=AnyStr(), ), ) }, @@ -848,8 +838,7 @@ async def test_dynamic_interrupt_subgraph( interrupts=( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:"), AnyStr("do:")], + id=AnyStr(), ), ), state={ @@ -871,8 +860,7 @@ async def test_dynamic_interrupt_subgraph( interrupts=( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:"), AnyStr("do:")], + id=AnyStr(), ), ), ) @@ -938,9 +926,7 @@ async def test_partial_pending_checkpoint( ) == { "my_key": "value one", "market": "DE", - "__interrupt__": [ - Interrupt(value="Just because...", resumable=True, ns=[AnyStr("tool_two:")]) - ], + "__interrupt__": [Interrupt(value="Just because...", id=AnyStr())], } assert tool_two_node_count == 1, "interrupts aren't retried" assert len(tracer.runs) == 1 @@ -971,8 +957,7 @@ async def test_partial_pending_checkpoint( "__interrupt__": ( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ) }, @@ -1002,8 +987,7 @@ async def test_partial_pending_checkpoint( "__interrupt__": [ Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ) ], } @@ -1037,8 +1021,7 @@ async def test_partial_pending_checkpoint( interrupts=( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ), ), @@ -1054,8 +1037,7 @@ async def test_partial_pending_checkpoint( interrupts=( Interrupt( value="Just because...", - resumable=True, - ns=[AnyStr("tool_two:")], + id=AnyStr(), ), ), ) @@ -1119,8 +1101,7 @@ async def test_node_not_cancelled_on_other_node_interrupted( "__interrupt__": [ Interrupt( value="I am bad", - resumable=True, - ns=[AnyStr("bad:")], + id=AnyStr(), ) ], } @@ -1133,8 +1114,7 @@ async def test_node_not_cancelled_on_other_node_interrupted( "__interrupt__": [ Interrupt( value="I am bad", - resumable=True, - ns=[AnyStr("bad:")], + id=AnyStr(), ) ], } @@ -2292,9 +2272,7 @@ async def test_imp_task( "__interrupt__": ( Interrupt( value="question", - resumable=True, - ns=[AnyStr("graph:")], - when="during", + id=AnyStr(), ), ) }, @@ -2373,9 +2351,7 @@ async def test_imp_nested( "__interrupt__": ( Interrupt( value="question", - resumable=True, - ns=[AnyStr("graph:")], - when="during", + id=AnyStr(), ), ) }, @@ -2428,9 +2404,7 @@ async def test_imp_task_cancel( "__interrupt__": ( Interrupt( value="question", - resumable=True, - ns=[AnyStr("graph:")], - when="during", + id=AnyStr(), ), ) }, @@ -2522,6 +2496,10 @@ async def test_imp_stream_order( ] +@pytest.mark.skipif( + sys.version_info < (3, 11), + reason="Requires Python 3.11 or higher for context management", +) async def test_send_dedupe_on_resume( async_checkpointer: BaseCheckpointSaver, checkpoint_during: bool ) -> None: @@ -2531,7 +2509,7 @@ async def test_send_dedupe_on_resume( def __call__(self, state): self.ticks += 1 if self.ticks == 1: - raise NodeInterrupt("Bahh") + interrupt("Bahh") return ["|".join(("flaky", str(state)))] class Node: @@ -2578,8 +2556,7 @@ async def test_send_dedupe_on_resume( "__interrupt__": [ Interrupt( value="Bahh", - resumable=False, - ns=None, + id=AnyStr(), ), ], } @@ -2730,7 +2707,7 @@ async def test_send_dedupe_on_resume( name="flaky", path=("__pregel_push", 1, False), error=None, - interrupts=(Interrupt(value="Bahh", resumable=False, ns=None),), + interrupts=(Interrupt(value="Bahh", id=AnyStr()),), state=None, result=["flaky|4"] if checkpoint_during else None, ), @@ -2744,7 +2721,7 @@ async def test_send_dedupe_on_resume( result=["3"], ), ), - interrupts=(Interrupt(value="Bahh", resumable=False, ns=None),), + interrupts=(Interrupt(value="Bahh", id=AnyStr()),), ), StateSnapshot( values=["0", "1"], @@ -5143,8 +5120,7 @@ async def test_subgraph_checkpoint_true_interrupt( "__interrupt__": [ Interrupt( value="Provide baz value", - resumable=True, - ns=[AnyStr("node_2"), AnyStr("subgraph_node_1:")], + id=AnyStr(), ) ], } @@ -6220,9 +6196,7 @@ async def test_interrupt_multiple(async_checkpointer: BaseCheckpointSaver): "__interrupt__": ( Interrupt( value={"value": 1}, - resumable=True, - ns=[AnyStr("node:")], - when="during", + id=AnyStr(), ), ) } @@ -6240,9 +6214,7 @@ async def test_interrupt_multiple(async_checkpointer: BaseCheckpointSaver): "__interrupt__": ( Interrupt( value={"value": 2}, - resumable=True, - ns=[AnyStr("node:")], - when="during", + id=AnyStr(), ), ) } @@ -6290,9 +6262,7 @@ async def test_interrupt_loop(async_checkpointer: BaseCheckpointSaver) -> None: "__interrupt__": ( Interrupt( value="How old are you?", - resumable=True, - ns=[AnyStr("node:")], - when="during", + id=AnyStr(), ), ) } @@ -6309,9 +6279,7 @@ async def test_interrupt_loop(async_checkpointer: BaseCheckpointSaver) -> None: "__interrupt__": ( Interrupt( value="invalid response", - resumable=True, - ns=[AnyStr("node:")], - when="during", + id=AnyStr(), ), ) } @@ -6328,9 +6296,7 @@ async def test_interrupt_loop(async_checkpointer: BaseCheckpointSaver) -> None: "__interrupt__": ( Interrupt( value="invalid response", - resumable=True, - ns=[AnyStr("node:")], - when="during", + id=AnyStr(), ), ) } @@ -6392,8 +6358,7 @@ async def test_interrupt_task_functional( "__interrupt__": [ Interrupt( value="Provide value for bar:", - resumable=True, - ns=[AnyStr("graph:"), AnyStr("bar:")], + id=AnyStr(), ), ] } @@ -6914,9 +6879,7 @@ async def test_double_interrupt_subgraph( "__interrupt__": ( Interrupt( value="interrupt node 1", - resumable=True, - ns=[AnyStr("node_1:")], - when="during", + id=AnyStr(), ), ) }, @@ -6930,9 +6893,7 @@ async def test_double_interrupt_subgraph( "__interrupt__": ( Interrupt( value="interrupt node 2", - resumable=True, - ns=[AnyStr("node_2:")], - when="during", + id=AnyStr(), ), ) }, @@ -6964,9 +6925,7 @@ async def test_double_interrupt_subgraph( "__interrupt__": ( Interrupt( value="interrupt node 1", - resumable=True, - ns=[AnyStr("invoke_sub_agent:"), AnyStr("node_1:")], - when="during", + id=AnyStr(), ), ) }, @@ -6978,9 +6937,7 @@ async def test_double_interrupt_subgraph( "__interrupt__": ( Interrupt( value="interrupt node 2", - resumable=True, - ns=[AnyStr("invoke_sub_agent:"), AnyStr("node_2:")], - when="during", + id=AnyStr(), ), ) } @@ -7843,8 +7800,7 @@ async def test_interrupt_subgraph_reenter_checkpointer_true( "__interrupt__": [ Interrupt( value="Provide value", - resumable=True, - ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")], + id=AnyStr(), ) ], } @@ -7854,8 +7810,7 @@ async def test_interrupt_subgraph_reenter_checkpointer_true( "__interrupt__": [ Interrupt( value="Provide value", - resumable=True, - ns=[AnyStr("call_subgraph"), AnyStr("subnode_2")], + id=AnyStr(), ) ], } @@ -7885,8 +7840,7 @@ async def test_interrupt_subgraph_reenter_checkpointer_true( "__interrupt__": [ Interrupt( value="Provide value", - resumable=True, - ns=[AnyStr("call_subgraph"), AnyStr("subnode_2:")], + id=AnyStr(), ) ], } @@ -7923,8 +7877,7 @@ async def test_handles_multiple_interrupts_from_tasks( "__interrupt__": [ Interrupt( value="Hey do you want to add James?", - resumable=True, - ns=[AnyStr("program:"), AnyStr("add_participant:")], + id=AnyStr(), ), ] } @@ -7932,10 +7885,6 @@ async def test_handles_multiple_interrupts_from_tasks( state = await program.aget_state(config=config) assert len(state.tasks[0].interrupts) == 1 task_interrupt = state.tasks[0].interrupts[0] - assert task_interrupt.resumable is True - assert len(task_interrupt.ns) == 2 - assert task_interrupt.ns[0].startswith("program:") - assert task_interrupt.ns[1].startswith("add_participant:") assert task_interrupt.value == "Hey do you want to add James?" result = await program.ainvoke(Command(resume=True), config=config) @@ -7943,8 +7892,7 @@ async def test_handles_multiple_interrupts_from_tasks( "__interrupt__": [ Interrupt( value="Hey do you want to add Will?", - resumable=True, - ns=[AnyStr("program:"), AnyStr("add_participant:")], + id=AnyStr(), ), ] } @@ -7952,10 +7900,6 @@ async def test_handles_multiple_interrupts_from_tasks( state = await program.aget_state(config=config) assert len(state.tasks[0].interrupts) == 1 task_interrupt = state.tasks[0].interrupts[0] - assert task_interrupt.resumable is True - assert len(task_interrupt.ns) == 2 - assert task_interrupt.ns[0].startswith("program:") - assert task_interrupt.ns[1].startswith("add_participant:") assert task_interrupt.value == "Hey do you want to add Will?" result = await program.ainvoke(Command(resume=True), config=config) @@ -7999,10 +7943,6 @@ async def test_interrupts_in_tasks_surfaced_once( state = await program.aget_state(config=config) assert len(state.tasks[0].interrupts) == 1 task_interrupt = state.tasks[0].interrupts[0] - assert task_interrupt.resumable is True - assert len(task_interrupt.ns) == 2 - assert task_interrupt.ns[0].startswith("program:") - assert task_interrupt.ns[1].startswith("add_participant:") assert task_interrupt.value == "Hey do you want to add James?" interrupts = [ @@ -8015,10 +7955,6 @@ async def test_interrupts_in_tasks_surfaced_once( state = await program.aget_state(config=config) assert len(state.tasks[0].interrupts) == 1 task_interrupt = state.tasks[0].interrupts[0] - assert task_interrupt.resumable is True - assert len(task_interrupt.ns) == 2 - assert task_interrupt.ns[0].startswith("program:") - assert task_interrupt.ns[1].startswith("add_participant:") assert task_interrupt.value == "Hey do you want to add Will?" result = await program.ainvoke(Command(resume=True), config=config) diff --git a/libs/langgraph/tests/test_remote_graph.py b/libs/langgraph/tests/test_remote_graph.py index 262109673..76aa1c592 100644 --- a/libs/langgraph/tests/test_remote_graph.py +++ b/libs/langgraph/tests/test_remote_graph.py @@ -16,6 +16,7 @@ from langgraph.graph import StateGraph, add_messages from langgraph.pregel import Pregel from langgraph.pregel.remote import RemoteGraph from langgraph.types import Interrupt, StateSnapshot +from tests.any_str import AnyStr from tests.conftest import NO_DOCKER from tests.example_app.example_graph import app @@ -459,9 +460,7 @@ def test_stream(): "__interrupt__": [ { "value": {"question": "Does this look good?"}, - "resumable": True, - "ns": ["some_ns"], - "when": "during", + "id": AnyStr(), } ] }, @@ -489,9 +488,7 @@ def test_stream(): assert exc.value.args[0] == [ Interrupt( value={"question": "Does this look good?"}, - resumable=True, - ns=["some_ns"], - when="during", + id=AnyStr(), ) ] @@ -632,9 +629,7 @@ async def test_astream(): "__interrupt__": [ { "value": {"question": "Does this look good?"}, - "resumable": True, - "ns": ["some_ns"], - "when": "during", + "id": AnyStr(), } ] }, @@ -663,9 +658,7 @@ async def test_astream(): assert exc.value.args[0] == [ Interrupt( value={"question": "Does this look good?"}, - resumable=True, - ns=["some_ns"], - when="during", + id=AnyStr(), ) ] diff --git a/libs/prebuilt/langgraph/prebuilt/tool_node.py b/libs/prebuilt/langgraph/prebuilt/tool_node.py index 3de19db62..7ec65d44d 100644 --- a/libs/prebuilt/langgraph/prebuilt/tool_node.py +++ b/libs/prebuilt/langgraph/prebuilt/tool_node.py @@ -320,9 +320,10 @@ class ToolNode(RunnableCallable): response = self.tools_by_name[call["name"]].invoke(input, config) # GraphInterrupt is a special exception that will always be raised. - # It can be triggered in the following scenarios: - # (1) a NodeInterrupt is raised inside a tool - # (2) a NodeInterrupt is raised inside a graph node for a graph called as a tool + # It can be triggered in the following scenarios, + # Where GraphInterrupt(GraphBubbleUp) is raised from an `interrupt` invocation most commonly: + # (1) a GraphInterrupt is raised inside a tool + # (2) a GraphInterrupt is raised inside a graph node for a graph called as a tool # (3) a GraphInterrupt is raised when a subgraph is interrupted inside a graph called as a tool # (2 and 3 can happen in a "supervisor w/ tools" multi-agent architecture) except GraphBubbleUp as e: diff --git a/libs/prebuilt/tests/test_react_agent.py b/libs/prebuilt/tests/test_react_agent.py index d704616db..2aac3cbf6 100644 --- a/libs/prebuilt/tests/test_react_agent.py +++ b/libs/prebuilt/tests/test_react_agent.py @@ -1298,9 +1298,7 @@ def test_tool_node_node_interrupt( assert task.interrupts == ( Interrupt( value="provide value for foo", - when="during", - resumable=True, - ns=[AnyStr("tools:")], + id=AnyStr(), ), ) diff --git a/libs/prebuilt/tests/test_tool_node.py b/libs/prebuilt/tests/test_tool_node.py index b94094a23..ec7c91fdd 100644 --- a/libs/prebuilt/tests/test_tool_node.py +++ b/libs/prebuilt/tests/test_tool_node.py @@ -14,7 +14,7 @@ from langchain_core.tools import tool as dec_tool from pydantic import BaseModel, ValidationError from pydantic.v1 import ValidationError as ValidationErrorV1 -from langgraph.errors import NodeInterrupt +from langgraph.errors import GraphBubbleUp, GraphInterrupt from langgraph.prebuilt import ToolNode from langgraph.prebuilt.tool_node import TOOL_CALL_ERROR_TEMPLATE from langgraph.types import Command, Send @@ -462,16 +462,16 @@ def test_tool_node_incorrect_tool_name(): def test_tool_node_node_interrupt(): - def tool_interrupt(some_val: int) -> str: + def tool_interrupt(some_val: int) -> None: """Tool docstring.""" - raise NodeInterrupt("foo") + raise GraphBubbleUp("foo") - def handle(e: NodeInterrupt): + def handle(e: GraphInterrupt): return "handled" - for handle_tool_errors in (True, (NodeInterrupt,), "handled", handle, False): + for handle_tool_errors in (True, (GraphBubbleUp,), "handled", handle, False): node = ToolNode([tool_interrupt], handle_tool_errors=handle_tool_errors) - with pytest.raises(NodeInterrupt) as exc_info: + with pytest.raises(GraphBubbleUp) as exc_info: node.invoke( { "messages": [ diff --git a/libs/sdk-py/langgraph_sdk/schema.py b/libs/sdk-py/langgraph_sdk/schema.py index 9ad1c5b8b..13b525bca 100644 --- a/libs/sdk-py/langgraph_sdk/schema.py +++ b/libs/sdk-py/langgraph_sdk/schema.py @@ -222,17 +222,13 @@ class Assistant(AssistantBase): """The last time the assistant was updated.""" -class Interrupt(TypedDict, total=False): +class Interrupt(TypedDict): """Represents an interruption in the execution flow.""" value: Any """The value associated with the interrupt.""" - when: Literal["during"] - """When the interrupt occurred.""" - resumable: bool - """Whether the interrupt can be resumed.""" - ns: list[str] | None - """Optional namespace for the interrupt.""" + id: str + """The ID of the interrupt. Can be used to resume the interrupt.""" class Thread(TypedDict):