WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-08 18:57:52 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix double interrupt return caused by dynamic tasks (#4389)

Browse Source 
Fix issue found in
https://github.com/langchain-ai/langgraph/pull/4374#discussion_r2056701206
This commit is contained in:
Sydney Runkle
2025-04-23 15:56:47 -07:00
committed by GitHub
parent 6bcab08f55 67075bfb7f
commit 0ff913c50b
5 changed files with 116 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/langgraph/langgraph/pregel/algo.py
+14 -8
View File
@@ -542,11 +542,14 @@ def prepare_single_task(
str(task_path[2]),
)
task_checkpoint_ns = f"{checkpoint_ns}:{task_id}"
# we append True to the task path to indicate that a call is being
# made, so we should not return interrupts from this task (responsibility lies with the parent)
task_path = (*task_path[:3], True)
metadata = {
"langgraph_step": step,
"langgraph_node": name,
"langgraph_triggers": triggers,
"langgraph_path": task_path[:3],
"langgraph_path": task_path,
"langgraph_checkpoint_ns": task_checkpoint_ns,
}
if task_id_checksum is not None:
@@ -575,7 +578,7 @@ def prepare_single_task(
local_read,
channels,
managed,
PregelTaskWrites(task_path[:3], name, writes, triggers),
PregelTaskWrites(task_path, name, writes, triggers),
),
CONFIG_KEY_STORE: (store or configurable.get(CONFIG_KEY_STORE)),
CONFIG_KEY_CHECKPOINTER: (
@@ -598,10 +601,10 @@ def prepare_single_task(
call.retry,
None,
task_id,
task_path[:3],
task_path,
)
else:
return PregelTask(task_id, name, task_path[:3])
return PregelTask(task_id, name, task_path)
elif task_path[0] == PUSH:
if len(task_path) == 2:
# SEND tasks, executed in superstep n+1
@@ -637,11 +640,14 @@ def prepare_single_task(
logger.warning(f"Ignoring invalid PUSH task path {task_path}")
return
task_checkpoint_ns = f"{checkpoint_ns}:{task_id}"
# we append False to the task path to indicate that a call is not being made
# so we should return interrupts from this task
task_path = (*task_path[:3], False)
metadata = {
"langgraph_step": step,
"langgraph_node": packet.node,
"langgraph_triggers": triggers,
"langgraph_path": task_path[:3],
"langgraph_path": task_path,
"langgraph_checkpoint_ns": task_checkpoint_ns,
}
if task_id_checksum is not None:
@@ -678,7 +684,7 @@ def prepare_single_task(
channels,
managed,
PregelTaskWrites(
task_path[:3], packet.node, writes, triggers
task_path, packet.node, writes, triggers
),
),
CONFIG_KEY_STORE: (
@@ -708,12 +714,12 @@ def prepare_single_task(
proc.retry_policy,
None,
task_id,
task_path[:3],
task_path,
writers=proc.flat_writers,
subgraphs=proc.subgraphs,
)
else:
return PregelTask(task_id, packet.node, task_path[:3])
return PregelTask(task_id, packet.node, task_path)
elif task_path[0] == PULL:
# (PULL, node name)
name = cast(str, task_path[1])
libs/langgraph/langgraph/pregel/loop.py
+5
View File
@@ -909,6 +909,11 @@ class PregelLoop(LoopProtocol):
):
return
if writes[0][0] == INTERRUPT:
# in loop.py we append a bool to the PUSH task paths to indicate
# whether or not a call was present (that was popped). If so,
# we don't emit the interrupt as it'll be emitted by the parent
if task.path[0] == PUSH and task.path[-1] is True:
return
self._emit(
"updates",
lambda: iter(
libs/langgraph/tests/test_large_cases.py
+18 -18
View File
@@ -3034,7 +3034,7 @@ def test_state_graph_packets(
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config={
"configurable": {
@@ -3098,7 +3098,7 @@ def test_state_graph_packets(
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config={
"configurable": {
@@ -3209,8 +3209,8 @@ def test_state_graph_packets(
]
},
tasks=(
PregelTask(AnyStr(), "tools", (PUSH, 0)),
PregelTask(AnyStr(), "tools", (PUSH, 1)),
PregelTask(AnyStr(), "tools", (PUSH, 0, False)),
PregelTask(AnyStr(), "tools", (PUSH, 1, False)),
),
next=("tools", "tools"),
config={
@@ -3367,7 +3367,7 @@ def test_state_graph_packets(
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config={
"configurable": {
@@ -3431,7 +3431,7 @@ def test_state_graph_packets(
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config=app_w_interrupt.checkpointer.get_tuple(config).config,
created_at=AnyStr(),
@@ -3536,8 +3536,8 @@ def test_state_graph_packets(
]
},
tasks=(
PregelTask(AnyStr(), "tools", (PUSH, 0)),
PregelTask(AnyStr(), "tools", (PUSH, 1)),
PregelTask(AnyStr(), "tools", (PUSH, 0, False)),
PregelTask(AnyStr(), "tools", (PUSH, 1, False)),
),
next=("tools", "tools"),
config={
@@ -5916,7 +5916,7 @@ def test_copy_checkpoint(
PregelTask(
id=AnyStr(),
name="tool_one",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
result={"my_key": " one"},
),
PregelTask(
@@ -5970,7 +5970,7 @@ def test_copy_checkpoint(
PregelTask(
id=AnyStr(),
name="tool_one",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
),
PregelTask(
AnyStr(),
@@ -7485,7 +7485,7 @@ def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="2",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -7494,7 +7494,7 @@ def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="flaky",
path=("__pregel_push", 1),
path=("__pregel_push", 1, False),
error=None,
interrupts=(Interrupt(value="Bahh", resumable=False, ns=None),),
state=None,
@@ -7540,7 +7540,7 @@ def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="2",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -7549,7 +7549,7 @@ def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="2",
path=("__pregel_push", 1),
path=("__pregel_push", 1, False),
error=None,
interrupts=(),
state=None,
@@ -9543,7 +9543,7 @@ def test_send_react_interrupt(
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -9703,7 +9703,7 @@ def test_send_react_interrupt(
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -9794,7 +9794,7 @@ def test_send_react_interrupt(
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -10013,7 +10013,7 @@ def test_send_react_interrupt_control(
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
libs/langgraph/tests/test_large_cases_async.py
+8 -8
View File
@@ -2757,7 +2757,7 @@ async def test_state_graph_packets(checkpointer_name: str) -> None:
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config={
"configurable": {
@@ -2822,7 +2822,7 @@ async def test_state_graph_packets(checkpointer_name: str) -> None:
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config=tup.config,
created_at=tup.checkpoint["ts"],
@@ -2929,8 +2929,8 @@ async def test_state_graph_packets(checkpointer_name: str) -> None:
]
},
tasks=(
PregelTask(AnyStr(), "tools", (PUSH, 0)),
PregelTask(AnyStr(), "tools", (PUSH, 1)),
PregelTask(AnyStr(), "tools", (PUSH, 0, False)),
PregelTask(AnyStr(), "tools", (PUSH, 1, False)),
),
next=("tools", "tools"),
config=tup.config,
@@ -3074,7 +3074,7 @@ async def test_state_graph_packets(checkpointer_name: str) -> None:
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config=tup.config,
created_at=tup.checkpoint["ts"],
@@ -3135,7 +3135,7 @@ async def test_state_graph_packets(checkpointer_name: str) -> None:
),
]
},
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0)),),
tasks=(PregelTask(AnyStr(), "tools", (PUSH, 0, False)),),
next=("tools",),
config=tup.config,
created_at=tup.checkpoint["ts"],
@@ -3242,8 +3242,8 @@ async def test_state_graph_packets(checkpointer_name: str) -> None:
]
},
tasks=(
PregelTask(AnyStr(), "tools", (PUSH, 0)),
PregelTask(AnyStr(), "tools", (PUSH, 1)),
PregelTask(AnyStr(), "tools", (PUSH, 0, False)),
PregelTask(AnyStr(), "tools", (PUSH, 1, False)),
),
next=("tools", "tools"),
config=tup.config,
libs/langgraph/tests/test_pregel_async.py
+71 -10
View File
@@ -992,7 +992,7 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None:
PregelTask(
AnyStr(),
name="tool_one",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -1044,7 +1044,7 @@ async def test_copy_checkpoint(checkpointer_name: str) -> None:
PregelTask(
AnyStr(),
"tool_one",
(PUSH, 0),
(PUSH, 0, False),
result=None,
),
PregelTask(
@@ -2952,7 +2952,7 @@ async def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="2",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -2961,7 +2961,7 @@ async def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="flaky",
path=("__pregel_push", 1),
path=("__pregel_push", 1, False),
error=None,
interrupts=(Interrupt(value="Bahh", resumable=False, ns=None),),
state=None,
@@ -3007,7 +3007,7 @@ async def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="2",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -3016,7 +3016,7 @@ async def test_send_dedupe_on_resume(
PregelTask(
id=AnyStr(),
name="2",
path=("__pregel_push", 1),
path=("__pregel_push", 1, False),
error=None,
interrupts=(),
state=None,
@@ -3295,7 +3295,7 @@ async def test_send_react_interrupt(checkpointer_name: str) -> None:
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -3453,7 +3453,7 @@ async def test_send_react_interrupt(checkpointer_name: str) -> None:
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -3544,7 +3544,7 @@ async def test_send_react_interrupt(checkpointer_name: str) -> None:
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -3761,7 +3761,7 @@ async def test_send_react_interrupt_control(
PregelTask(
id=AnyStr(),
name="foo",
path=("__pregel_push", 0),
path=("__pregel_push", 0, False),
error=None,
interrupts=(),
state=None,
@@ -8190,6 +8190,67 @@ async def test_handles_multiple_interrupts_from_tasks() -> None:
assert result[1] == "Added Will!"
@NEEDS_CONTEXTVARS
async def test_interrupts_in_tasks_surfaced_once() -> None:
@task
async def add_participant(name: str) -> str:
feedback = interrupt(f"Hey do you want to add {name}?")
if feedback is False:
return f"The user changed their mind and doesn't want to add {name}!"
if feedback is True:
return f"Added {name}!"
raise ValueError("Invalid feedback")
@entrypoint(checkpointer=MemorySaver())
async def program(_state: Any) -> list[str]:
first = await add_participant("James")
second = await add_participant("Will")
return [first, second]
config = {"configurable": {"thread_id": "1"}}
interrupts = [
e
async for e in program.astream("this is ignored", config=config)
if "__interrupt__" in e
]
assert len(interrupts) == 1
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 = [
e
async for e in program.astream(Command(resume=True), config=config)
if "__interrupt__" in e
]
assert len(interrupts) == 1
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)
assert result is not None
assert len(result) == 2
assert result[0] == "Added James!"
assert result[1] == "Added Will!"
async def test_pregel_loop_refcount():
gc.collect()
try: