From 9a7b1fa12aeb1309ab76dcbc134ec6398e26d2f1 Mon Sep 17 00:00:00 2001 From: Sydney Runkle Date: Tue, 22 Apr 2025 10:15:19 -0700 Subject: [PATCH] initial pass - surfacing interrupts for stream_mode='values' --- libs/langgraph/langgraph/pregel/__init__.py | 15 ++++++--- libs/langgraph/langgraph/pregel/loop.py | 34 ++++++++++----------- libs/langgraph/langgraph/types.py | 12 +++++++- 3 files changed, 39 insertions(+), 22 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index 2b016e587..8c7cd2e28 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -2732,10 +2732,11 @@ class Pregel(PregelProtocol): If stream_mode is not "values", it returns a list of output chunks. """ output_keys = output_keys if output_keys is not None else self.output_channels - if stream_mode == "values": - latest: Union[dict[str, Any], Any] = None - else: - chunks = [] + + latest: Union[dict[str, Any], Any] = None + chunks: list[Union[dict[str, Any], Any]] = [] + interrupts: list[Interrupt] = [] + for chunk in self.stream( input, config, @@ -2748,10 +2749,16 @@ class Pregel(PregelProtocol): **kwargs, ): if stream_mode == "values": + if isinstance(chunk, dict) and (ints := chunk.get(INTERRUPT)) is not None: + interrupts.extend(ints) latest = chunk else: chunks.append(chunk) if stream_mode == "values": + if len(interrupts) > 0: + return { + INTERRUPT: interrupts + } return latest else: return chunks diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index 644fd5a68..01471a1c0 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -910,23 +910,23 @@ class PregelLoop(LoopProtocol): ): return if writes[0][0] == INTERRUPT: - self._emit( - "updates", - lambda: iter( - [ - { - INTERRUPT: tuple( - v - for w in writes - if w[0] == INTERRUPT - for v in ( - w[1] if isinstance(w[1], Sequence) else (w[1],) - ) - ) - } - ] - ), - ) + interrupts = [ + { + INTERRUPT: tuple( + v + for w in writes + if w[0] == INTERRUPT + for v in ( + w[1] if isinstance(w[1], Sequence) else (w[1],) + ) + ) + } + ] + stream_modes = self.stream.modes if self.stream else [] + if "updates" in stream_modes: + self._emit("updates", lambda: iter(interrupts)) + elif "values" in stream_modes: + self._emit("values", lambda: iter(interrupts)) elif writes[0][0] != ERROR: self._emit( "updates", diff --git a/libs/langgraph/langgraph/types.py b/libs/langgraph/langgraph/types.py index 195acd4ff..de3e86ce6 100644 --- a/libs/langgraph/langgraph/types.py +++ b/libs/langgraph/langgraph/types.py @@ -24,6 +24,8 @@ from typing_extensions import Self from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointMetadata from langgraph.utils.fields import get_update_as_tuples +import hashlib +import uuid if TYPE_CHECKING: from langgraph.pregel.protocol import PregelProtocol @@ -144,6 +146,13 @@ class Interrupt: when: Literal["during"] = dataclasses.field(default="during", repr=False) + @property + def interrupt_id(self) -> str: + """Generate a unique ID for the interrupt based on its namespace.""" + identifier = uuid.uuid4().bytes if self.ns is None else ''.join(self.ns).encode() + return hashlib.sha256(identifier).hexdigest() + + class StateUpdate(NamedTuple): values: Optional[dict[str, Any]] as_node: Optional[str] = None @@ -483,11 +492,12 @@ def interrupt(value: Any) -> Any: CONFIG_KEY_SEND, NS_SEP, RESUME, + CONF, ) from langgraph.errors import GraphInterrupt from langgraph.utils.config import get_config - conf = get_config()["configurable"] + conf = get_config()[CONF] # track interrupt index scratchpad: PregelScratchpad = conf[CONFIG_KEY_SCRATCHPAD] idx = scratchpad.interrupt_counter()