diff --git a/libs/langgraph/langgraph/pregel/algo.py b/libs/langgraph/langgraph/pregel/algo.py index 94a5ade90..19cd9d1ff 100644 --- a/libs/langgraph/langgraph/pregel/algo.py +++ b/libs/langgraph/langgraph/pregel/algo.py @@ -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]) diff --git a/libs/langgraph/langgraph/pregel/loop.py b/libs/langgraph/langgraph/pregel/loop.py index c88362df6..3e3849fc7 100644 --- a/libs/langgraph/langgraph/pregel/loop.py +++ b/libs/langgraph/langgraph/pregel/loop.py @@ -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( 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..af3b16dfa 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, @@ -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: