WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-08-17 21:25:46 +02:00
Code Issues Packages Projects Releases Wiki Activity

change[langgraph]: clean up Interrupt interface for v1 (#5405)

Browse Source
This commit is contained in:
Sydney Runkle
2025-07-09 14:03:08 -04:00
committed by GitHub
parent e5947bcd30
commit d1710e2eac
17 changed files with 320 additions and 370 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/docs/cloud/how-tos/add-human-in-the-loop.md
+2 -6
View File
@@ -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': '...',
# > }
# > ]
docs/docs/concepts/functional_api.md
+30 -26
View File
@@ -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:
docs/docs/how-tos/human_in_the_loop/add-human-in-the-loop.md
+23 -36
View File
@@ -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'}}
```
libs/langgraph/langgraph/errors.py
+24 -3
View File
@@ -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):
libs/langgraph/langgraph/pregel/remote.py
+4 -4
View File
@@ -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"]
libs/langgraph/langgraph/types.py
+49 -22
View File
@@ -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],
),
)
)
libs/langgraph/tests/test_checkpoint_migration.py
+9 -13
View File
@@ -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(),
),
)
},
libs/langgraph/tests/test_deprecation.py
+20 -1
View File
@@ -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")
libs/langgraph/tests/test_interrupt_migration.py
+50
View File
@@ -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"
libs/langgraph/tests/test_large_cases.py
+19 -40
View File
@@ -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"],
libs/langgraph/tests/test_pregel.py
+27 -80
View File
@@ -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
libs/langgraph/tests/test_pregel_async.py
+44 -108
View File
@@ -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)
libs/langgraph/tests/test_remote_graph.py
+5 -12
View File
@@ -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(),
)
]
libs/prebuilt/langgraph/prebuilt/tool_node.py
+4 -3
View File
@@ -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:
libs/prebuilt/tests/test_react_agent.py
+1 -3
View File
@@ -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(),
),
)
libs/prebuilt/tests/test_tool_node.py
+6 -6
View File
@@ -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": [
libs/sdk-py/langgraph_sdk/schema.py
+3 -7
View File
@@ -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):