mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-26 09:32:25 +02:00
fix(langgraph): revert selective interrupt task scheduling (#6252)
Reverts langchain-ai/langgraph#6158
This commit is contained in:
Caspar Broekhuizen
@@ -114,7 +114,6 @@ from langgraph.types import (
|
||||
CachePolicy,
|
||||
Command,
|
||||
Durability,
|
||||
Interrupt,
|
||||
PregelExecutableTask,
|
||||
RetryPolicy,
|
||||
StreamMode,
|
||||
@@ -249,7 +248,6 @@ class PregelLoop:
|
||||
self.retry_policy = retry_policy
|
||||
self.cache_policy = cache_policy
|
||||
self.durability = durability
|
||||
self.skipped_task_ids: set[str] = set()
|
||||
if self.stream is not None and CONFIG_KEY_STREAM in config[CONF]:
|
||||
self.stream = DuplexStream(self.stream, config[CONF][CONFIG_KEY_STREAM])
|
||||
scratchpad: PregelScratchpad | None = config[CONF].get(CONFIG_KEY_SCRATCHPAD)
|
||||
@@ -318,19 +316,14 @@ class PregelLoop:
|
||||
writes_to_save: WritesT = [
|
||||
w[1:] for w in self.checkpoint_pending_writes if w[0] == task_id
|
||||
] + list(writes)
|
||||
self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes)
|
||||
else:
|
||||
writes_to_save = [
|
||||
# aggregate existing interrupts for this task
|
||||
(ch, self._merge_interrupts(task_id, v) if ch == INTERRUPT else v)
|
||||
for ch, v in writes
|
||||
]
|
||||
|
||||
# replace all writes for this task_id in one shot
|
||||
# remove existing writes for this task
|
||||
self.checkpoint_pending_writes = [
|
||||
w for w in self.checkpoint_pending_writes if w[0] != task_id
|
||||
] + [(task_id, c, v) for c, v in writes_to_save]
|
||||
|
||||
]
|
||||
writes_to_save = writes
|
||||
# save writes
|
||||
self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes)
|
||||
if self.durability != "exit" and self.checkpointer_put_writes is not None:
|
||||
config = patch_configurable(
|
||||
self.checkpoint_config,
|
||||
@@ -478,20 +471,6 @@ class PregelLoop:
|
||||
cache_policy=self.cache_policy,
|
||||
)
|
||||
|
||||
resume_map = self.config.get(CONF, {}).get(CONFIG_KEY_RESUME_MAP, {})
|
||||
if resume_map:
|
||||
skipped_interrupt_ids = self._pending_interrupts() - set(resume_map)
|
||||
self.skipped_task_ids = {
|
||||
task_id
|
||||
for task_id, channel, value in self.checkpoint_pending_writes
|
||||
if channel == INTERRUPT
|
||||
# interrupts within a task are uncovered sequentially as resumes are provided,
|
||||
# so we only need to check the last interrupt id
|
||||
and value[-1].id in skipped_interrupt_ids
|
||||
}
|
||||
else:
|
||||
self.skipped_task_ids = set()
|
||||
|
||||
# produce debug output
|
||||
if self._checkpointer_put_after_previous is not None:
|
||||
self._emit(
|
||||
@@ -537,45 +516,9 @@ class PregelLoop:
|
||||
if task.writes:
|
||||
self.output_writes(task.id, task.writes, cached=True)
|
||||
|
||||
if self.skipped_task_ids:
|
||||
# remove tasks with writes that may have been matched from previous loop
|
||||
self.skipped_task_ids = {
|
||||
task_id
|
||||
for task_id in self.skipped_task_ids
|
||||
if not self.tasks[task_id].writes
|
||||
}
|
||||
# output interrupt writes for blocked tasks so they are still visible in the stream
|
||||
for task_id, channel, value in self.checkpoint_pending_writes:
|
||||
if task_id in self.skipped_task_ids and channel == INTERRUPT:
|
||||
# find resume count for this task
|
||||
resumes = next(
|
||||
(
|
||||
v
|
||||
for tid, ch, v in self.checkpoint_pending_writes
|
||||
if tid == task_id and ch == RESUME
|
||||
),
|
||||
None,
|
||||
)
|
||||
resume_count = len(resumes) if resumes is not None else 0
|
||||
# only output unresumed interrupts
|
||||
if resume_count < len(value):
|
||||
self.output_writes(task_id, [(INTERRUPT, value[resume_count:])])
|
||||
|
||||
return True
|
||||
|
||||
def after_tick(self) -> None:
|
||||
if self.skipped_task_ids:
|
||||
# raise early GraphInterrupt for skipped tasks.
|
||||
# since we know len(resumes) != len(interrupts) for these tasks, we
|
||||
# can prevent unnecessary node re-execution by raising preemptively
|
||||
interrupts = []
|
||||
for task_id, channel, value in self.checkpoint_pending_writes:
|
||||
if channel == INTERRUPT and task_id in self.skipped_task_ids:
|
||||
interrupts.extend(value)
|
||||
if interrupts:
|
||||
raise GraphInterrupt(interrupts)
|
||||
|
||||
self.skipped_task_ids.clear()
|
||||
# finish superstep
|
||||
writes = [w for t in self.tasks.values() for w in t.writes]
|
||||
# all tasks have finished
|
||||
@@ -627,53 +570,34 @@ class PregelLoop:
|
||||
|
||||
def _pending_interrupts(self) -> set[str]:
|
||||
"""Return the set of interrupt ids that are pending without corresponding resume values."""
|
||||
# mapping of task ids to (interrupt_id, interrupt_count)
|
||||
pending_interrupts: dict[str, tuple[str, int]] = {}
|
||||
# mapping of task ids to resume count
|
||||
pending_resumes: dict[str, int] = {}
|
||||
# mapping of task ids to interrupt ids
|
||||
pending_interrupts: dict[str, str] = {}
|
||||
|
||||
for task_id, channel, value in self.checkpoint_pending_writes:
|
||||
if channel == INTERRUPT:
|
||||
pending_interrupts[task_id] = (
|
||||
value[0].id,
|
||||
len(value),
|
||||
)
|
||||
elif channel == RESUME:
|
||||
resume_list = value if isinstance(value, list) else [value]
|
||||
pending_resumes[task_id] = len(resume_list)
|
||||
# set of resume task ids
|
||||
pending_resumes: set[str] = set()
|
||||
|
||||
# keep only interrupt ids where resume_count < interrupt_count
|
||||
for task_id, write_type, value in self.checkpoint_pending_writes:
|
||||
if write_type == INTERRUPT:
|
||||
# interrupts is always a list, but there should only be one element
|
||||
pending_interrupts[task_id] = value[0].id
|
||||
elif write_type == RESUME:
|
||||
pending_resumes.add(task_id)
|
||||
|
||||
resumed_interrupt_ids = {
|
||||
pending_interrupts[task_id]
|
||||
for task_id in pending_resumes
|
||||
if task_id in pending_interrupts
|
||||
}
|
||||
|
||||
# Keep only interrupts whose interrupt_id is not resumed
|
||||
hanging_interrupts: set[str] = {
|
||||
interrupt_id
|
||||
for task_id, (interrupt_id, interrupt_count) in pending_interrupts.items()
|
||||
if pending_resumes.get(task_id, 0) < interrupt_count
|
||||
for interrupt_id in pending_interrupts.values()
|
||||
if interrupt_id not in resumed_interrupt_ids
|
||||
}
|
||||
|
||||
return hanging_interrupts
|
||||
|
||||
def _merge_interrupts(
|
||||
self, task_id: str, value: Sequence[Interrupt]
|
||||
) -> Sequence[Interrupt]:
|
||||
"""Normalize interrupt value to list and merge with existing interrupts.
|
||||
|
||||
If the interrupt ID matches existing, append; otherwise replace.
|
||||
|
||||
Returns list of Interrupt objects for this task.
|
||||
"""
|
||||
new = value if isinstance(value, list) else list(value)
|
||||
existing = next(
|
||||
(
|
||||
v
|
||||
for tid, ch, v in self.checkpoint_pending_writes
|
||||
if tid == task_id and ch == INTERRUPT
|
||||
),
|
||||
None,
|
||||
)
|
||||
if existing is None:
|
||||
return new
|
||||
old = existing if isinstance(existing, list) else list(existing)
|
||||
return old + new if old and new and old[0].id == new[0].id else new
|
||||
|
||||
def _first(
|
||||
self, *, input_keys: str | Sequence[str], updated_channels: set[str] | None
|
||||
) -> set[str] | None:
|
||||
@@ -1102,7 +1026,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
|
||||
|
||||
def put_writes(self, task_id: str, writes: WritesT) -> None:
|
||||
"""Put writes for a task, to be read by the next tick."""
|
||||
|
||||
super().put_writes(task_id, writes)
|
||||
if not writes or self.cache is None or not hasattr(self, "tasks"):
|
||||
return
|
||||
|
||||
@@ -2672,11 +2672,7 @@ class Pregel(
|
||||
for task in loop.match_cached_writes():
|
||||
loop.output_writes(task.id, task.writes, cached=True)
|
||||
for _ in runner.tick(
|
||||
[
|
||||
t
|
||||
for t in loop.tasks.values()
|
||||
if not t.writes and t.id not in loop.skipped_task_ids
|
||||
],
|
||||
[t for t in loop.tasks.values() if not t.writes],
|
||||
timeout=self.step_timeout,
|
||||
get_waiter=get_waiter,
|
||||
schedule_task=loop.accept_push,
|
||||
@@ -2995,11 +2991,7 @@ class Pregel(
|
||||
for task in await loop.amatch_cached_writes():
|
||||
loop.output_writes(task.id, task.writes, cached=True)
|
||||
async for _ in runner.atick(
|
||||
[
|
||||
t
|
||||
for t in loop.tasks.values()
|
||||
if not t.writes and t.id not in loop.skipped_task_ids
|
||||
],
|
||||
[t for t in loop.tasks.values() if not t.writes],
|
||||
timeout=self.step_timeout,
|
||||
get_waiter=get_waiter,
|
||||
schedule_task=loop.aaccept_push,
|
||||
|
||||
@@ -1,21 +1,12 @@
|
||||
import operator
|
||||
import sys
|
||||
from typing import Annotated
|
||||
|
||||
import pytest
|
||||
from langgraph.checkpoint.base import BaseCheckpointSaver
|
||||
from typing_extensions import TypedDict
|
||||
|
||||
from langgraph.graph import END, START, StateGraph
|
||||
from langgraph.types import Command, Durability, Send, interrupt
|
||||
from langgraph.types import Durability
|
||||
|
||||
pytestmark = pytest.mark.anyio
|
||||
|
||||
NEEDS_CONTEXTVARS = pytest.mark.skipif(
|
||||
sys.version_info < (3, 11),
|
||||
reason="Python 3.11+ is required for async contextvars support",
|
||||
)
|
||||
|
||||
|
||||
def test_interruption_without_state_updates(
|
||||
sync_checkpointer: BaseCheckpointSaver, durability: Durability
|
||||
@@ -99,486 +90,3 @@ async def test_interruption_without_state_updates_async(
|
||||
assert (await graph.aget_state(thread)).next == ()
|
||||
n_checkpoints = len([c async for c in graph.aget_state_history(thread)])
|
||||
assert n_checkpoints == (5 if durability != "exit" else 3)
|
||||
|
||||
|
||||
def test_interrupt_with_send_payloads(sync_checkpointer: BaseCheckpointSaver) -> None:
|
||||
"""Test interruption in map node with Send payloads and human-in-the-loop resume."""
|
||||
|
||||
# Global counter to track node executions
|
||||
node_counter = {"entry": 0, "map_node": 0}
|
||||
|
||||
class State(TypedDict):
|
||||
items: list[str]
|
||||
processed: Annotated[list[str], operator.add]
|
||||
|
||||
def entry_node(state: State):
|
||||
node_counter["entry"] += 1
|
||||
return {} # No state updates in entry node
|
||||
|
||||
def send_to_map(state: State):
|
||||
return [Send("map_node", {"item": item}) for item in state["items"]]
|
||||
|
||||
def map_node(state: State):
|
||||
node_counter["map_node"] += 1
|
||||
if "dangerous" in state["item"]:
|
||||
value = interrupt({"processing": state["item"]})
|
||||
return {"processed": [f"processed_{value}"]}
|
||||
else:
|
||||
return {"processed": [f"processed_{state['item']}_auto"]}
|
||||
|
||||
builder = StateGraph(State)
|
||||
builder.add_node("entry", entry_node)
|
||||
builder.add_node("map_node", map_node)
|
||||
builder.add_edge(START, "entry")
|
||||
builder.add_conditional_edges("entry", send_to_map, ["map_node"])
|
||||
builder.add_edge("map_node", END)
|
||||
|
||||
graph = builder.compile(checkpointer=sync_checkpointer)
|
||||
|
||||
config = {"configurable": {"thread_id": "test_interrupt_send"}}
|
||||
|
||||
# Run until interrupts
|
||||
result = graph.invoke(
|
||||
{"items": ["item1", "dangerous_item1", "dangerous_item2"]}, config=config
|
||||
)
|
||||
|
||||
# Verify we have interrupts (only one for dangerous_item)
|
||||
interrupts = result.get("__interrupt__", [])
|
||||
assert len(interrupts) == 2
|
||||
assert "dangerous_item" in interrupts[0].value["processing"]
|
||||
|
||||
# Resume with mapping of interrupt IDs to values
|
||||
resume_map = {i.id: f"human_input_{i.value['processing']}" for i in interrupts}
|
||||
|
||||
final_result = graph.invoke(Command(resume=resume_map), config=config)
|
||||
|
||||
# Verify final result contains processed items
|
||||
assert "processed" in final_result
|
||||
processed_items = final_result["processed"]
|
||||
assert len(processed_items) == 3
|
||||
assert "processed_item1_auto" in processed_items # item1 processed automatically
|
||||
assert any(
|
||||
"processed_human_input_dangerous_item1" in item for item in processed_items
|
||||
) # dangerous_item1 processed after interrupt
|
||||
assert any(
|
||||
"processed_human_input_dangerous_item2" in item for item in processed_items
|
||||
) # dangerous_item2 processed after interrupt
|
||||
|
||||
# Verify node execution counts
|
||||
assert node_counter["entry"] == 1 # Entry node runs once
|
||||
# Map node runs 3 times initially (item1 completes, 2 dangerous_items interrupt),
|
||||
# then 2 times on resume
|
||||
assert node_counter["map_node"] == 5
|
||||
|
||||
|
||||
@NEEDS_CONTEXTVARS
|
||||
async def test_interrupt_with_send_payloads_async(
|
||||
async_checkpointer: BaseCheckpointSaver, durability: Durability
|
||||
) -> None:
|
||||
"""Test interruption in map node with Send payloads and human-in-the-loop resume."""
|
||||
|
||||
# Global counter to track node executions
|
||||
node_counter = {"entry": 0, "map_node": 0}
|
||||
|
||||
class State(TypedDict):
|
||||
items: list[str]
|
||||
processed: Annotated[list[str], operator.add]
|
||||
|
||||
def entry_node(state: State):
|
||||
node_counter["entry"] += 1
|
||||
return {} # No state updates in entry node
|
||||
|
||||
def send_to_map(state: State):
|
||||
return [Send("map_node", {"item": item}) for item in state["items"]]
|
||||
|
||||
def map_node(state: State):
|
||||
node_counter["map_node"] += 1
|
||||
if "dangerous" in state["item"]:
|
||||
value = interrupt({"processing": state["item"]})
|
||||
return {"processed": [f"processed_{value}"]}
|
||||
else:
|
||||
return {"processed": [f"processed_{state['item']}_auto"]}
|
||||
|
||||
builder = StateGraph(State)
|
||||
builder.add_node("entry", entry_node)
|
||||
builder.add_node("map_node", map_node)
|
||||
builder.add_edge(START, "entry")
|
||||
builder.add_conditional_edges("entry", send_to_map, ["map_node"])
|
||||
builder.add_edge("map_node", END)
|
||||
|
||||
graph = builder.compile(checkpointer=async_checkpointer)
|
||||
|
||||
config = {"configurable": {"thread_id": "test_interrupt_send"}}
|
||||
|
||||
# Run until interrupts
|
||||
result = await graph.ainvoke(
|
||||
{"items": ["item1", "dangerous_item1", "dangerous_item2"]}, config=config
|
||||
)
|
||||
|
||||
# Verify we have interrupts (only one for dangerous_item)
|
||||
interrupts = result.get("__interrupt__", [])
|
||||
assert len(interrupts) == 2
|
||||
assert "dangerous_item" in interrupts[0].value["processing"]
|
||||
|
||||
# Resume with mapping of interrupt IDs to values
|
||||
resume_map = {i.id: f"human_input_{i.value['processing']}" for i in interrupts}
|
||||
|
||||
final_result = await graph.ainvoke(Command(resume=resume_map), config=config)
|
||||
|
||||
# Verify final result contains processed items
|
||||
assert "processed" in final_result
|
||||
processed_items = final_result["processed"]
|
||||
assert len(processed_items) == 3
|
||||
assert "processed_item1_auto" in processed_items # item1 processed automatically
|
||||
assert any(
|
||||
"processed_human_input_dangerous_item1" in item for item in processed_items
|
||||
) # dangerous_item1 processed after interrupt
|
||||
assert any(
|
||||
"processed_human_input_dangerous_item2" in item for item in processed_items
|
||||
) # dangerous_item2 processed after interrupt
|
||||
|
||||
# Verify node execution counts
|
||||
assert node_counter["entry"] == 1 # Entry node runs once
|
||||
# Map node runs 3 times initially (item1 completes, 2 dangerous_items interrupt),
|
||||
# then 2 times on resume
|
||||
assert node_counter["map_node"] == 5
|
||||
|
||||
|
||||
def test_interrupt_with_send_payloads_sequential_resume(
|
||||
sync_checkpointer: BaseCheckpointSaver,
|
||||
) -> None:
|
||||
"""Test interruption in map node with Send payloads and sequential resume."""
|
||||
|
||||
# Global counter to track node executions
|
||||
node_counter = {"entry": 0, "map_node": 0}
|
||||
|
||||
class State(TypedDict):
|
||||
items: list[str]
|
||||
processed: Annotated[list[str], operator.add]
|
||||
|
||||
def entry_node(state: State):
|
||||
node_counter["entry"] += 1
|
||||
return {} # No state updates in entry node
|
||||
|
||||
def send_to_map(state: State):
|
||||
return [Send("map_node", {"item": item}) for item in state["items"]]
|
||||
|
||||
def map_node(state: State):
|
||||
node_counter["map_node"] += 1
|
||||
if "dangerous" in state["item"]:
|
||||
value = interrupt({"processing": state["item"]})
|
||||
return {"processed": [f"processed_{value}"]}
|
||||
else:
|
||||
return {"processed": [f"processed_{state['item']}_auto"]}
|
||||
|
||||
builder = StateGraph(State)
|
||||
builder.add_node("entry", entry_node)
|
||||
builder.add_node("map_node", map_node)
|
||||
builder.add_edge(START, "entry")
|
||||
builder.add_conditional_edges("entry", send_to_map, ["map_node"])
|
||||
builder.add_edge("map_node", END)
|
||||
|
||||
graph = builder.compile(checkpointer=sync_checkpointer)
|
||||
|
||||
config = {"configurable": {"thread_id": "test_interrupt_send_sequential"}}
|
||||
|
||||
# Run until interrupts
|
||||
result = graph.invoke(
|
||||
{"items": ["item1", "dangerous_item1", "dangerous_item2"]}, config=config
|
||||
)
|
||||
|
||||
# Verify we have interrupts
|
||||
interrupts = result.get("__interrupt__", [])
|
||||
assert len(interrupts) == 2
|
||||
assert "dangerous_item" in interrupts[0].value["processing"]
|
||||
|
||||
# Resume first interrupt only
|
||||
first_interrupt = interrupts[0]
|
||||
first_resume_map = {
|
||||
first_interrupt.id: f"human_input_{first_interrupt.value['processing']}"
|
||||
}
|
||||
|
||||
partial_result = graph.invoke(Command(resume=first_resume_map), config=config)
|
||||
|
||||
# Verify we still have one pending interrupt
|
||||
remaining_interrupts = partial_result.get("__interrupt__", [])
|
||||
assert len(remaining_interrupts) == 1
|
||||
|
||||
# Resume second interrupt
|
||||
second_interrupt = remaining_interrupts[0]
|
||||
second_resume_map = {
|
||||
second_interrupt.id: f"human_input_{second_interrupt.value['processing']}"
|
||||
}
|
||||
|
||||
final_result = graph.invoke(Command(resume=second_resume_map), config=config)
|
||||
|
||||
# Verify final result contains processed items
|
||||
assert "processed" in final_result
|
||||
processed_items = final_result["processed"]
|
||||
assert len(processed_items) == 3
|
||||
assert "processed_item1_auto" in processed_items # item1 processed automatically
|
||||
assert any(
|
||||
"processed_human_input_dangerous_item1" in item for item in processed_items
|
||||
) # dangerous_item1 processed after interrupt
|
||||
assert any(
|
||||
"processed_human_input_dangerous_item2" in item for item in processed_items
|
||||
) # dangerous_item2 processed after interrupt
|
||||
|
||||
# Verify node execution counts
|
||||
assert node_counter["entry"] == 1 # Entry node runs once
|
||||
# Map node runs 3 times initially (item1 completes, 2 dangerous_items interrupt),
|
||||
# then 1 time on first resume, then 1 time on second resume
|
||||
assert node_counter["map_node"] == 5
|
||||
|
||||
|
||||
@NEEDS_CONTEXTVARS
|
||||
async def test_interrupt_with_send_payloads_sequential_resume_async(
|
||||
async_checkpointer: BaseCheckpointSaver,
|
||||
) -> None:
|
||||
"""Test interruption in map node with Send payloads and sequential resume."""
|
||||
|
||||
# Global counter to track node executions
|
||||
node_counter = {"entry": 0, "map_node": 0}
|
||||
|
||||
class State(TypedDict):
|
||||
items: list[str]
|
||||
processed: Annotated[list[str], operator.add]
|
||||
|
||||
def entry_node(state: State):
|
||||
node_counter["entry"] += 1
|
||||
return {} # No state updates in entry node
|
||||
|
||||
def send_to_map(state: State):
|
||||
return [Send("map_node", {"item": item}) for item in state["items"]]
|
||||
|
||||
def map_node(state: State):
|
||||
node_counter["map_node"] += 1
|
||||
if "dangerous" in state["item"]:
|
||||
value = interrupt({"processing": state["item"]})
|
||||
return {"processed": [f"processed_{value}"]}
|
||||
else:
|
||||
return {"processed": [f"processed_{state['item']}_auto"]}
|
||||
|
||||
builder = StateGraph(State)
|
||||
builder.add_node("entry", entry_node)
|
||||
builder.add_node("map_node", map_node)
|
||||
builder.add_edge(START, "entry")
|
||||
builder.add_conditional_edges("entry", send_to_map, ["map_node"])
|
||||
builder.add_edge("map_node", END)
|
||||
|
||||
graph = builder.compile(checkpointer=async_checkpointer)
|
||||
|
||||
config = {"configurable": {"thread_id": "test_interrupt_send_sequential"}}
|
||||
|
||||
# Run until interrupts
|
||||
result = await graph.ainvoke(
|
||||
{"items": ["item1", "dangerous_item1", "dangerous_item2"]}, config=config
|
||||
)
|
||||
|
||||
# Verify we have interrupts
|
||||
interrupts = result.get("__interrupt__", [])
|
||||
assert len(interrupts) == 2
|
||||
assert "dangerous_item" in interrupts[0].value["processing"]
|
||||
|
||||
# Resume first interrupt only
|
||||
first_interrupt = interrupts[0]
|
||||
first_resume_map = {
|
||||
first_interrupt.id: f"human_input_{first_interrupt.value['processing']}"
|
||||
}
|
||||
|
||||
partial_result = await graph.ainvoke(
|
||||
Command(resume=first_resume_map), config=config
|
||||
)
|
||||
|
||||
# Verify we still have one pending interrupt
|
||||
remaining_interrupts = partial_result.get("__interrupt__", [])
|
||||
assert len(remaining_interrupts) == 1
|
||||
|
||||
# Resume second interrupt
|
||||
second_interrupt = remaining_interrupts[0]
|
||||
second_resume_map = {
|
||||
second_interrupt.id: f"human_input_{second_interrupt.value['processing']}"
|
||||
}
|
||||
|
||||
final_result = await graph.ainvoke(Command(resume=second_resume_map), config=config)
|
||||
|
||||
# Verify final result contains processed items
|
||||
assert "processed" in final_result
|
||||
processed_items = final_result["processed"]
|
||||
assert len(processed_items) == 3
|
||||
assert "processed_item1_auto" in processed_items # item1 processed automatically
|
||||
assert any(
|
||||
"processed_human_input_dangerous_item1" in item for item in processed_items
|
||||
) # dangerous_item1 processed after interrupt
|
||||
assert any(
|
||||
"processed_human_input_dangerous_item2" in item for item in processed_items
|
||||
) # dangerous_item2 processed after interrupt
|
||||
|
||||
# Verify node execution counts
|
||||
assert node_counter["entry"] == 1 # Entry node runs once
|
||||
# Map node runs 3 times initially (item1 completes, 2 dangerous_items interrupt),
|
||||
# then 1 time on first resume, then 1 time on second resume
|
||||
assert node_counter["map_node"] == 5
|
||||
|
||||
|
||||
def test_node_with_multiple_interrupts_requires_full_resume(
|
||||
sync_checkpointer: BaseCheckpointSaver,
|
||||
) -> None:
|
||||
"""Test a number of different resume patterns for a node with multiple interrupts,
|
||||
|
||||
Ensures that a node is not re-executed until valid resume values have been provided to all
|
||||
discovered interrupts"""
|
||||
|
||||
node_counter = 0
|
||||
|
||||
class State(TypedDict):
|
||||
input: str
|
||||
|
||||
def double_interrupt_node(state: State):
|
||||
nonlocal node_counter
|
||||
node_counter += 1
|
||||
first = interrupt("first")
|
||||
second = interrupt("second")
|
||||
third = interrupt("third")
|
||||
return {"input": f"{first}-{second}-{third}"}
|
||||
|
||||
builder = StateGraph(State)
|
||||
builder.add_node("double_interrupt", double_interrupt_node)
|
||||
builder.add_edge(START, "double_interrupt")
|
||||
builder.add_edge("double_interrupt", END)
|
||||
|
||||
graph = builder.compile(checkpointer=sync_checkpointer)
|
||||
|
||||
config = {"configurable": {"thread_id": "test_double_interrupt"}}
|
||||
|
||||
result = graph.invoke({"input": "start"}, config=config)
|
||||
|
||||
interrupts = result.get("__interrupt__", [])
|
||||
assert len(interrupts) == 1
|
||||
first_interrupt = interrupts[0]
|
||||
assert node_counter == 1
|
||||
|
||||
# invoke with an interrupt map that matches double_interrupt_node.
|
||||
# this should execute the node
|
||||
partial = graph.invoke(
|
||||
Command(resume={first_interrupt.id: "human_first"}), config=config
|
||||
)
|
||||
remaining_interrupts = partial.get("__interrupt__", [])
|
||||
assert len(remaining_interrupts) == 1
|
||||
assert remaining_interrupts[0].value == "second"
|
||||
assert node_counter == 2
|
||||
|
||||
# invoke with an interrupt map that DOES NOT match double_interrupt_node.
|
||||
# this should not execute the node because the optimization kicks in
|
||||
partial = graph.invoke(
|
||||
Command(resume={"00000000000000000000000000000000": "nothing_burger"}),
|
||||
config=config,
|
||||
)
|
||||
remaining_interrupts = partial.get("__interrupt__", [])
|
||||
assert len(remaining_interrupts) == 1
|
||||
assert remaining_interrupts[0].value == "second"
|
||||
assert node_counter == 2
|
||||
|
||||
# invoke with None resume. this should execute the node
|
||||
partial = graph.invoke(None, config=config)
|
||||
remaining_interrupts = partial.get("__interrupt__", [])
|
||||
assert len(remaining_interrupts) == 1
|
||||
assert remaining_interrupts[0].value == "second"
|
||||
assert node_counter == 3
|
||||
|
||||
# invoke with nonspecific resume. this should execute the node
|
||||
partial = graph.invoke(Command(resume="human_second"), config=config)
|
||||
remaining_interrupts = partial.get("__interrupt__", [])
|
||||
assert len(remaining_interrupts) == 1
|
||||
print("REMAINING INTERRUPTS: ", remaining_interrupts)
|
||||
assert remaining_interrupts[0].value == "third"
|
||||
assert node_counter == 4
|
||||
|
||||
# finally, invoke with an interrupt map that matches double_interrupt_node.
|
||||
# this should execute the node and all interrupts should be resolved
|
||||
final_result = graph.invoke(Command(resume="human_third"), config=config)
|
||||
assert "input" in final_result
|
||||
assert final_result["input"] == "human_first-human_second-human_third"
|
||||
assert node_counter == 5
|
||||
|
||||
|
||||
@NEEDS_CONTEXTVARS
|
||||
async def test_node_with_multiple_interrupts_requires_full_resume_async(
|
||||
async_checkpointer: BaseCheckpointSaver,
|
||||
) -> None:
|
||||
"""Test a number of different resume patterns for a node with multiple interrupts,
|
||||
|
||||
Ensures that a node is not re-executed until valid resume values have been provided to all
|
||||
discovered interrupts"""
|
||||
|
||||
node_counter = 0
|
||||
|
||||
class State(TypedDict):
|
||||
input: str
|
||||
|
||||
def double_interrupt_node(state: State):
|
||||
nonlocal node_counter
|
||||
node_counter += 1
|
||||
first = interrupt("first")
|
||||
second = interrupt("second")
|
||||
third = interrupt("third")
|
||||
return {"input": f"{first}-{second}-{third}"}
|
||||
|
||||
builder = StateGraph(State)
|
||||
builder.add_node("double_interrupt", double_interrupt_node)
|
||||
builder.add_edge(START, "double_interrupt")
|
||||
builder.add_edge("double_interrupt", END)
|
||||
|
||||
graph = builder.compile(checkpointer=async_checkpointer)
|
||||
|
||||
config = {"configurable": {"thread_id": "test_double_interrupt"}}
|
||||
|
||||
result = await graph.ainvoke({"input": "start"}, config=config)
|
||||
|
||||
interrupts = result.get("__interrupt__", [])
|
||||
assert len(interrupts) == 1
|
||||
first_interrupt = interrupts[0]
|
||||
assert node_counter == 1
|
||||
|
||||
# invoke with an interrupt map that matches double_interrupt_node.
|
||||
# this should execute the node
|
||||
partial = await graph.ainvoke(
|
||||
Command(resume={first_interrupt.id: "human_first"}), config=config
|
||||
)
|
||||
remaining_interrupts = partial.get("__interrupt__", [])
|
||||
assert len(remaining_interrupts) == 1
|
||||
assert remaining_interrupts[0].value == "second"
|
||||
assert node_counter == 2
|
||||
|
||||
# invoke with an interrupt map that DOES NOT match double_interrupt_node.
|
||||
# this should not execute the node because the optimization kicks in
|
||||
partial = await graph.ainvoke(
|
||||
Command(resume={"00000000000000000000000000000000": "nothing_burger"}),
|
||||
config=config,
|
||||
)
|
||||
remaining_interrupts = partial.get("__interrupt__", [])
|
||||
assert len(remaining_interrupts) == 1
|
||||
assert remaining_interrupts[0].value == "second"
|
||||
assert node_counter == 2
|
||||
|
||||
# invoke with None resume. this should execute the node
|
||||
partial = await graph.ainvoke(None, config=config)
|
||||
remaining_interrupts = partial.get("__interrupt__", [])
|
||||
assert len(remaining_interrupts) == 1
|
||||
assert remaining_interrupts[0].value == "second"
|
||||
assert node_counter == 3
|
||||
|
||||
# invoke with nonspecific resume. this should execute the node
|
||||
partial = await graph.ainvoke(Command(resume="human_second"), config=config)
|
||||
remaining_interrupts = partial.get("__interrupt__", [])
|
||||
assert len(remaining_interrupts) == 1
|
||||
print("REMAINING INTERRUPTS: ", remaining_interrupts)
|
||||
assert remaining_interrupts[0].value == "third"
|
||||
assert node_counter == 4
|
||||
|
||||
# finally, invoke with an interrupt map that matches double_interrupt_node.
|
||||
# this should execute the node and all interrupts should be resolved
|
||||
final_result = await graph.ainvoke(Command(resume="human_third"), config=config)
|
||||
assert "input" in final_result
|
||||
assert final_result["input"] == "human_first-human_second-human_third"
|
||||
assert node_counter == 5
|
||||
|
||||
Reference in New Issue
Block a user