Add resumeable/ns properties to Interrupt

This commit is contained in:
Nuno Campos
2024-11-13 12:51:55 -08:00
parent 16bfa80b58
commit 7fe6f88876
4 changed files with 67 additions and 10 deletions
+1 -1
View File
@@ -70,7 +70,7 @@ class NodeInterrupt(GraphInterrupt):
"""Raised by a node to interrupt execution."""
def __init__(self, value: Any) -> None:
super().__init__([Interrupt(value)])
super().__init__([Interrupt(value=value)])
class GraphDelegate(Exception):
+25 -4
View File
@@ -1,5 +1,6 @@
from collections import deque
from dataclasses import dataclass
import sys
from typing import (
TYPE_CHECKING,
Any,
@@ -47,6 +48,11 @@ StreamWriter = Callable[[Any], None]
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_KWARGS = {"kw_only": True, "slots": True}
else:
_DC_KWARGS = {}
def default_retry_on(exc: Exception) -> bool:
import httpx
@@ -104,9 +110,11 @@ class CachePolicy(NamedTuple):
pass
@dataclass
@dataclass(**_DC_KWARGS)
class Interrupt:
value: Any
resumable: bool = False
ns: Optional[str] = None
when: Literal["during"] = "during"
@@ -318,12 +326,25 @@ class LoopProtocol:
def interrupt(value: Any) -> Any:
from langgraph.constants import CONFIG_KEY_RESUME_VALUE, MISSING
from langgraph.errors import NodeInterrupt
from langgraph.constants import (
CONFIG_KEY_CHECKPOINT_NS,
CONFIG_KEY_RESUME_VALUE,
MISSING,
NS_SEP,
)
from langgraph.errors import GraphInterrupt
from langgraph.utils.config import get_configurable
conf = get_configurable()
if (resume := conf.get(CONFIG_KEY_RESUME_VALUE, MISSING)) and resume is not MISSING:
return resume
else:
raise NodeInterrupt(value)
raise GraphInterrupt(
(
Interrupt(
value=value,
resumable=True,
ns=cast(str, conf[CONFIG_KEY_CHECKPOINT_NS]).split(NS_SEP),
),
)
)
+16 -2
View File
@@ -8409,7 +8409,15 @@ def test_dynamic_interrupt(
assert [
c for c in tool_two.stream({"my_key": "value ⛰️", "market": "DE"}, thread2)
] == [
{"__interrupt__": [Interrupt(value="Just because...", when="during")]},
{
"__interrupt__": (
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:")],
),
)
},
]
# resume with answer
assert [c for c in tool_two.stream(Command(resume=" my answer"), thread2)] == [
@@ -8447,7 +8455,13 @@ def test_dynamic_interrupt(
AnyStr(),
"tool_two",
(PULL, "tool_two"),
interrupts=(Interrupt("Just because..."),),
interrupts=(
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:")],
),
),
),
),
config=tool_two.checkpointer.get_tuple(thread1).config,
+25 -3
View File
@@ -318,7 +318,15 @@ async def test_dynamic_interrupt(checkpointer_name: str) -> None:
{"my_key": "value ⛰️", "market": "DE"}, thread2
)
] == [
{"__interrupt__": [Interrupt(value="Just because...", when="during")]},
{
"__interrupt__": (
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:")],
),
)
},
]
# resume with answer
assert [
@@ -336,7 +344,15 @@ async def test_dynamic_interrupt(checkpointer_name: str) -> None:
{"my_key": "value ⛰️", "market": "DE"}, thread1
)
] == [
{"__interrupt__": [Interrupt(value="Just because...", when="during")]},
{
"__interrupt__": (
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:")],
),
)
},
]
assert [c.metadata async for c in tool_two.checkpointer.alist(thread1)] == [
{
@@ -363,7 +379,13 @@ async def test_dynamic_interrupt(checkpointer_name: str) -> None:
AnyStr(),
"tool_two",
(PULL, "tool_two"),
interrupts=(Interrupt("Just because..."),),
interrupts=(
Interrupt(
value="Just because...",
resumable=True,
ns=[AnyStr("tool_two:")],
),
),
),
),
config=tup.config,