Problems with parallel nodes with delayed conditional edges

[NathanAP]

Hi, I'm creating a parallel node workflow, so I started with this basic example:

But in my case, my node b has a tool that is called by conditional_edge. In this situation, since b called for the tool, the d node is thinking that b ended and it get called twice.

Here is an example (I added a random chance to call for the tool just to exemplify):

import operator
from typing import Annotated, Any

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END

import time

class State(TypedDict):
    # The operator.add reducer fn makes this append-only
    aggregate: Annotated[list, operator.add]


def a(state: State):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"]}


def b(state: State):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}


def c(state: State):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}


def d(state: State):
    print(f'Adding "D" to {state["aggregate"]}')
    return {"aggregate": ["D"]}

def e(state: State):
    print(f'Adding "E" to {state["aggregate"]}')
    return {"aggregate": ["E"]}


def node_tool(state: State):
    print(f'tool start running')
    time.sleep(5)
    print(f'tool end running')
    return None

def should_continue(a):
    import random

    if random.random() < 0.5:
        print("Called!")
        return "call"

    else:
        print("Not called!")
        return "next"


builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)
builder.add_node(e)
builder.add_node(node_tool)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_conditional_edges(source="b", path=should_continue, path_map={"call": "node_tool", "next": "d"})
builder.add_edge("a", "c")
builder.add_edge("c", "d")
builder.add_edge("d", "e")
builder.add_edge("e", END)
graph = builder.compile()

graph.invoke({"aggregate": []}, {"configurable": {"thread_id": "foo"}})

This is one of the possible results:

Adding "A" to []
Adding "B" to ['A']
Called!
Adding "C" to ['A']
Adding "D" to ['A', 'B', 'C']
tool start running
tool end running
Adding "E" to ['A', 'B', 'C', 'D']
Adding "D" to ['A', 'B', 'C', 'D', 'B']

Notice that d was called twice, because meanwhile c ended, b was in a "waiting" status. My guess is that b called for their tool and d though it ended, but the tool itself is still running.

Another case that I tried was removing my e node, like this:

import operator
from typing import Annotated, Any

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END

import time

class State(TypedDict):
    # The operator.add reducer fn makes this append-only
    aggregate: Annotated[list, operator.add]


def a(state: State):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"]}


def b(state: State):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}


def c(state: State):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}


def d(state: State):
    print(f'Adding "D" to {state["aggregate"]}')
    return {"aggregate": ["D"]}

def node_tool(state: State):
    print(f'tool start running')
    time.sleep(5)
    print(f'tool end running')
    return None

def should_continue(a):
    import random

    if random.random() < 0.5:
        print("Called!")
        return "call"

    else:
        print("Not called!")
        return "next"


builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)
builder.add_node(node_tool)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_conditional_edges(source="b", path=should_continue, path_map={"call": "node_tool", "next": "d"})
builder.add_edge("a", "c")
builder.add_edge("c", "d")
builder.add_edge("d", END)
graph = builder.compile()

graph.invoke({"aggregate": []}, {"configurable": {"thread_id": "foo"}})

which resulted in this:

Adding "A" to []
Adding "B" to ['A']
Called!
Adding "C" to ['A']
Adding "D" to ['A', 'B', 'C']
tool start running
tool end running

Notice that d ended before my tool finishes, so it would make it break the flow.

Is there any workaround to this?

[NathanAP]

So what I did to workaround that for now is add a counter on the state.
I added a number (in this case 2) on a and in b. Then, in c I reduce it by 1.
Since my d is called twice, I check on that value expecting it is 0 to do whatever it should.

This seems like a bug for me, tbh, because d should undestand that my node is not finished yet...

[MatheusBarbieri]

I just stumbled upon the same problem. It seems on newer versions of langgraph it is possible to defer a node execution until all incoming branchs are completed:
https://langchain-ai.github.io/langgraph/how-tos/graph-api/#defer-node-execution