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

fix double interrupt raise with task path bool flag

Browse Source
This commit is contained in:
Sydney Runkle
2025-04-23 14:16:35 -07:00
parent 6bcab08f55
commit f42fc971e3
5 changed files with 52 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/langgraph/langgraph/pregel/algo.py
+10 -8
View File
@@ -542,11 +542,12 @@ def prepare_single_task(
str(task_path[2]),
)
task_checkpoint_ns = f"{checkpoint_ns}:{task_id}"
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 +576,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 +599,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 +638,12 @@ def prepare_single_task(
logger.warning(f"Ignoring invalid PUSH task path {task_path}")
return
task_checkpoint_ns = f"{checkpoint_ns}:{task_id}"
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 +680,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 +710,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
+6
View File
@@ -903,12 +903,18 @@ class PregelLoop(LoopProtocol):
def _output_writes(
self, task_id: str, writes: Sequence[tuple[str, Any]], *, cached: bool = False
) -> None:
print(f"output writes {task_id}, {writes}")
if task := self.tasks.get(task_id):
if task.config is not None and TAG_HIDDEN in task.config.get(
"tags", EMPTY_SEQ
):
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
+10 -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,