diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index 94a5ade90..f931c3eb1 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -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]) diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index c88362df6..4a2a69f77 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -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( diff --git a/libs/langgraph/tests/test_large_cases.py b/libs/langgraph/tests/test_large_cases.py index 3c0e7da87..eb3406e1c 100644 --- a/libs/langgraph/tests/test_large_cases.py +++ b/libs/langgraph/tests/test_large_cases.py @@ -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, diff --git a/libs/langgraph/tests/test_large_cases_async.py b/libs/langgraph/tests/test_large_cases_async.py index 6dbf19fb2..3062682f7 100644 --- a/libs/langgraph/tests/test_large_cases_async.py +++ b/libs/langgraph/tests/test_large_cases_async.py @@ -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, diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index 1f6b25574..802845e46 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/libs/langgraph/tests/test_pregel_async.py @@ -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,