Feature: handling concurrent interrupts

[VMinB12]

FYI, at my company we are thinking about agentic architectures with many concurrent agents, each of which needs to be able to ask for and receive human input. In this kind of setup, we need to be able to surface interrupts to the user without impacting the other agents from continuing their work.

Let's illustrate that setup with a simple example, e.g. a graph like this:

                                                     +-----------+                                                  
                                                  ***| __start__ |*****                                             
                                          ********   +-----------+*    ********                                     
                                 *********          ***            ***         *********                            
                         ********                  *                  ***               ********                    
                    *****                        **                      **                     *****               
+----------------------+           +----------------------+           +------------+                +------------+  
| branch_1_A_Interrupt |           | branch_2_A_Interrupt |           | branch_3_A |                | branch_4_A |  
+----------------------+           +----------------------+           +------------+                +------------+  
            *                                  *                             *                             *        
            *                                  *                             *                             *        
            *                                  *                             *                             *        
     +------------+                     +------------+           +----------------------+           +------------+  
     | branch_1_B |*****                | branch_2_B |           | branch_3_B_Interrupt |       ****| branch_4_B |  
     +------------+     *********       +------------+           +----------------------+*******    +------------+  
                                 ********           ***            ***        *********                             
                                         *********     *        ***  *********                                      
                                                  ***** **    *******                                               
                                                      +---------+                                                   
                                                      | __end__ |                                                   
                                                      +---------+                                                   

So there are 4 branches, each with 2 sequential nodes.

• Branch 1 and 2 both have an interrupt in the fourth node.
• Branch 3 has an interrupt in the second node.
• Branch 4 has no interrupts.

My understanding is that the current LangGraph implementation will surface interrupts one at a time**, sequentially**. When 2 parallel nodes both call interrupt, then whichever one triggers first will be the one to be surfaced first.

This behaviour is not ideal (for my use case at least), because we would prefer to surface all interrupts at once to the user concurrently, rather than one at a time sequentially.
There's 2 separate improvements that can be made imo:

1. All interrupts within a parallel computation are surfaced together. After more testing, I realised that this is already supported.
2. Ideally, while we are waiting for the human to respond to the interrupts from branch 1 and 2, branch 3 and 4 would already continue to the next nodes (branch 3 B and branch 4 B).

For step 2., I think this would imply going beyond the superstep approach that underlies the current graph execution, so I'm not sure how feasible this is. It could offer quite a boost in performance in scenarios where for example Branch 1 A and Branch 3 B are slow, but Branch 3 A is fast.
If step 1 and step 2 are combined, we could even surface all 3 interrupts to the user immediately when they are triggered. That would be really nice!

Could you comment on if any of these steps are on the roadmap?

Here's example code for the example graph:

import operator
from functools import partial
from typing import Annotated, TypedDict
from uuid import uuid4

from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.state import CompiledStateGraph
from langgraph.types import Command, StreamMode, interrupt


class State(TypedDict):
    integer: Annotated[int, operator.add]


def add_one(state: State, message: str) -> State:
    print("Message:", message)
    return {"integer": 1}


def add_one_with_interrupt(state: State, message: str) -> State:
    human_input = interrupt("Interruption!")
    assert human_input == "continue"
    return add_one(state, message)


builder = (
    StateGraph(State)
    .add_node(
        "branch_1_A_Interrupt",
        partial(add_one_with_interrupt, message="Branch 1 A with interrupt"),
    )
    .add_node("branch_1_B", partial(add_one, message="Branch 1 B"))
    .add_node(
        "branch_2_A_Interrupt",
        partial(add_one_with_interrupt, message="Branch 2 A with interrupt"),
    )
    .add_node("branch_2_B", partial(add_one, message="Branch 2 B"))
    .add_node("branch_3_A", partial(add_one, message="Branch 3 A"))
    .add_node(
        "branch_3_B_Interrupt",
        partial(add_one_with_interrupt, message="Branch 3 B with interrupt"),
    )
    .add_node("branch_4_A", partial(add_one, message="Branch 4 A"))
    .add_node("branch_4_B", partial(add_one, message="Branch 4 B"))
    .add_edge(START, "branch_1_A_Interrupt")
    .add_edge(START, "branch_2_A_Interrupt")
    .add_edge(START, "branch_3_A")
    .add_edge(START, "branch_4_A")
    .add_edge("branch_1_A_Interrupt", "branch_1_B")
    .add_edge("branch_2_A_Interrupt", "branch_2_B")
    .add_edge("branch_3_A", "branch_3_B_Interrupt")
    .add_edge("branch_4_A", "branch_4_B")
    .add_edge("branch_1_B", END)
    .add_edge("branch_2_B", END)
    .add_edge("branch_3_B_Interrupt", END)
    .add_edge("branch_4_B", END)
)
app = builder.compile(checkpointer=InMemorySaver())

print(app.get_graph().draw_ascii())


def stream(
    app: CompiledStateGraph,
    inputs: dict,
    config: RunnableConfig | None = None,
    stream_mode: StreamMode | None = None,
):
    _config = config or {"configurable": {"thread_id": str(uuid4())}}
    for chunk in app.stream(
        inputs,
        _config,
        stream_mode=stream_mode,
    ):
        print(chunk)
        if "__interrupt__" in chunk:
            print("\nInterrupted! Continuing...\n")
            stream(
                app,
                inputs=Command(resume="continue"),
                config=_config,
                stream_mode=stream_mode,
            )


stream(app, {"integers": [0, 1]})

Edit 06/04/2025: Edited the message for clarity. Old phrasing is slashed through and new phrasing is bold.

[VMinB12]

A follow up question to this: How does the new functional API play into this? Could this already today surface multiple interrupts at once?

[mgrogger-i11]

Following this - I'm curious how this would effectively be handled by your UI. Is your proposal that all interrupts are surfaced at once and then a user sends back a single response addressing all interrupts? Or is it that all interrupts at a single step are captured at once, surfaced to the user individually, user responds to each interrupt, then all interrupts are passed back in a single "Command" input to the graph to resume?

[VMinB12]

To expand on that:
Let's say we want to address a single interrupt while there are multiple interrupts that triggered. How would we provide a Command to a particular interrupt? Right now the docs just say to send Command(resume=value_from_human), which contains no information about which interrupt to resolve.

[mgrogger-i11]

So, I think it's helpful to separate out some of these behaviors because I actually think you're hitting on a few different scenarios here. One is directly supported today, two aren't, but could be implemented with the right design. IMO the latter point may work better than your original suggestion, but open to your thoughts!

Scenario 1: Selecting the Right Node

In this scenario, I'm going to assume that all interruptions are triggered at the same level in the graph, but surfaced separately to the user. Under Scenario 1, you might find that generalizing the human_node in this example might be perfectly sufficient.

Scenario 2: Surfacing a Single Interrupt

For simplicity, I'll assume in this scenario that all interrupts occur at the same level in the graph, and surfaced together to the user. In reality, the solution for this scenario and the next are the same, save a small tweak.

In this scenario, we will assume the linked example represents the travel_advisor and hotel_advisor as siblings in a hierarchical/supervisor workflow. In that case, you can modify the human_node to combine the request from both the agents, surface that to the user, and then send the user input, along with each agent's interrupt request, to a router agent. The router agent would then be responsible for partitioning the relevant parts of the user's response to the appropriate _advisor agent.

Scenario 3: 1 interrupt to rule them all

Under scenario 3, I'll assume that interrupts can occur at any level, but surfaced all at once. This is actually just the generalized version of Scenario 2.

In the example from scenario 2, let's now assume that you have a 3rd branch with a itinerary_advisor agent that has two sub agents - food_advisor and event_advisor. Each of these two new agents may solicit user feedback on whether they like the food and event suggestions, etc.

Under this scenario, the two new agents exist one step below travel_advisor and hotel_advisor, but we want to surface all interrupts at once. In this case the solution from scenario 2 holds with one modification.

Under scenario 2, your builder may have lines like this:

builder.add_conditional_edge(travel_advisor, trigger_human_or_not, [human_node, END])

builder.add_conditional_edge(hotel_advisor, trigger_human_or_not, [human_node, END])

Since both nodes are at the same step, they are triggered simultaneously and human_node executes after both complete. In scenario 3, you can force human_node to wait for the two new agents to complete as well by doing something like this:

builder.add_conditional_edge([travel_advisor, hotel_advisor, food_advisor, events_advisor], trigger_human_or_not, [human_node, END])

Let me know if this makes sense/is helpful for your use case.

[VMinB12]

I didn't yet answer your first question correctly, let me do that now.
Actually, I want a third option that you didn't list yet (as I understand it).
The options are:

1. All interrupts are surfaced at once and then a user sends back a single response addressing all interrupts? It means that aggregation happens at the level of catching the interrupts, before they are surfaced to the user.
2. All interrupts at a single step are captured at once, surfaced to the user individually, user responds to each interrupt, then all interrupts are passed back in a single "Command" input to the graph to resume. It means that aggregation happens after the interrupts are surfaced to the user, and the graph receives a single aggregated Command.
3. All interrupts at a single step are captured at once, surfaced to the user individually, user responds to each interrupt individually. It means that no aggregation happens.

For my use case, I need option 3. My use case specifically is having multiple agents that can handle user requests concurrently. A single agent may get stuck, but that shouldn't stop the other agents from continuing their work. When one or more agents get stuck, the user can help each agent individually. Once an agent's interrupt is resolved, it should immediately be able to continue, but the other interrupted agents will still be stuck until the user handles those interrupts.

Rereading my initial message, I see it came across as needing some kind of aggregation, let me edit that message for clarity. In fact, what I need is no aggregation at all.

It is my understanding that this is not currently supported, do you agree?
I think a slight modification to the Interrupt-Command interface is needed, where we receive an interrupt_id and we have to pass it as Command(resume=value_from_human, interrupt_id=interrupt_id).

From my testing, it seems that currently, Command just resolves the first non-resolved interrupt that was triggered. That is problematic, because the user may want to resolve the second interrupt before resolving the first interrupt. It seems to me that this particular feature is not supported yet.

I might be confused about a few points though, so please do correct me! Imo, some additional documentation on concurrent interrupts would be great. Currently the docs seem focused on use cases where only 1 interrupt triggers at a time.

[VMinB12]

I realized after more testing that my code example makes it seam that only one interrupt is surfaced at a time, but that isn't actually true. It's just that the original def stream implementation immediately sends a command the moment a single interrupt is surfaced and it doesn't wait for multiple interrupts to be triggered. It can actually lead to some race conditions when you use async.

Here's an updated version of def stream:

def stream(
    app: CompiledStateGraph,
    inputs: dict,
    config: RunnableConfig | None = None,
    stream_mode: StreamMode | None = None,
):
    _config = config or {"configurable": {"thread_id": str(uuid4())}}
    chunks = []
    for chunk in app.stream(
        inputs,
        _config,
        stream_mode=stream_mode,
    ):
        chunks.append(chunk)
        print(chunk)
    if any("__interrupt__" in chunk for chunk in chunks):
        print("\nInterrupted! Continuing...\n")
        stream(
            app,
            inputs=Command(resume="continue"),
            config=_config,
            stream_mode=stream_mode,
        )

Now, if we run the code, we get:

                                                     +-----------+                                                  
                                                  ***| __start__ |*****                                             
                                          ********   +-----------+*    ********                                     
                                 *********          ***            ***         *********                            
                         ********                  *                  ***               ********                    
                    *****                        **                      **                     *****               
+----------------------+           +----------------------+           +------------+                +------------+  
| branch_1_A_Interrupt |           | branch_2_A_Interrupt |           | branch_3_A |                | branch_4_A |  
+----------------------+           +----------------------+           +------------+                +------------+  
            *                                  *                             *                             *        
            *                                  *                             *                             *        
            *                                  *                             *                             *        
     +------------+                     +------------+           +----------------------+           +------------+  
     | branch_1_B |*****                | branch_2_B |           | branch_3_B_Interrupt |       ****| branch_4_B |  
     +------------+     *********       +------------+           +----------------------+*******    +------------+  
                                 ********           ***            ***        *********                             
                                         *********     *        ***  *********                                      
                                                  ***** **    *******                                               
                                                      +---------+                                                   
                                                      | __end__ |                                                   
                                                      +---------+                                                   
Message: Branch 3 A
{'__interrupt__': (Interrupt(value='Interruption!', resumable=True, ns=['branch_1_A_Interrupt:e019dda8-edd0-3c15-dead-1b7564526e12']),)}
Message: Branch 4 A
{'__interrupt__': (Interrupt(value='Interruption!', resumable=True, ns=['branch_2_A_Interrupt:23896519-46bb-343f-ecd3-67430d0043e1']),)}
{'branch_3_A': {'integer': 1}}
{'branch_4_A': {'integer': 1}}

Interrupted! Continuing...

{'branch_3_A': {'integer': 1}, '__metadata__': {'cached': True}}
{'branch_4_A': {'integer': 1}, '__metadata__': {'cached': True}}
Message: Branch 1 A with interrupt
{'__interrupt__': (Interrupt(value='Interruption!', resumable=True, ns=['branch_2_A_Interrupt:23896519-46bb-343f-ecd3-67430d0043e1']),)}
{'branch_1_A_Interrupt': {'integer': 1}}

Interrupted! Continuing...

{'branch_1_A_Interrupt': {'integer': 1}, '__metadata__': {'cached': True}}
{'branch_3_A': {'integer': 1}, '__metadata__': {'cached': True}}
{'branch_4_A': {'integer': 1}, '__metadata__': {'cached': True}}
Message: Branch 2 A with interrupt
{'branch_2_A_Interrupt': {'integer': 1}}
Message: Branch 1 B
Message: Branch 2 B
{'branch_1_B': {'integer': 1}}
Message: Branch 4 B
{'__interrupt__': (Interrupt(value='Interruption!', resumable=True, ns=['branch_3_B_Interrupt:d4b25fc0-4e87-bb65-40c0-7ad65118833e']),)}
{'branch_2_B': {'integer': 1}}
{'branch_4_B': {'integer': 1}}

Interrupted! Continuing...

{'branch_1_B': {'integer': 1}, '__metadata__': {'cached': True}}
{'branch_2_B': {'integer': 1}, '__metadata__': {'cached': True}}
{'branch_4_B': {'integer': 1}, '__metadata__': {'cached': True}}
Message: Branch 3 B with interrupt
{'branch_3_B_Interrupt': {'integer': 1}}

Let's walk through these steps:
Now it is clear that in branch 1 A and branch 2 A both surface their interrupts the moment they occur. The interrupt in branch 3 B is not surfaced yet, because we are still on superstep A, which cannot complete until the interrupts of branch 1 A and branch 2 A are handled.
Then, we send our first Command. From the logs, we see that it was sent to Branch 1 A. I am not sure what controls this, will the first command always be sent to the first interrupt to be triggered? I would also want to be able to send a Command for Branch 2 A without resolving the interrupt for Branch 1 A yet. I don't know how to do that at the moment.
Then we wait for branch 1 A to complete and we send another Command which is received by branch 2 A.
Now all interrupts on superstep 1 have been handled and we move on to superstep 2.
This triggers the interrupt on branch 3 B.
Again, we wait for the other branches to complete.
Then we sent a Command for branch 3 B.
Finally, the full execution can complete.

This is great! It's very close to the behaviour I need. The only thing that I'm missing at the moment is the ability to send a Command to a particular interrupt. How can we do that?