Parallel node fan-out and fan-in with extra steps

[Freezaa9]

Hi !
Following this example, I wanna had some parallel execution into my graph:
https://github.com/langchain-ai/langgraph/blob/main/examples/branching.ipynb

The "issue" is if one parallel branch has more than one steps, 2 executions supersteps will be created (which is normal):

But if you want to overcome this you need to add steps "b" and "b2" into a child subgraph:

I understand the behavior of the first example, but is there a way to set "d" to wait for the 2 "supersteps" ? Or we need to play around with subgraph for that ?

Thanks for your help :)

### Code:
`from langgraph.graph import StateGraph
from typing_extensions import TypedDict
from typing import Annotated
import operator

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

from typing import Any

class ReturnNodeValue:
def init(self, node_secret: str):
self._value = node_secret

def __call__(self, state: State) -> Any:
    print(f"Adding {self._value} to {state['aggregate']}")
    return {"aggregate": [self._value]}

builder = StateGraph(State)
builder.add_node("a", ReturnNodeValue("I'm A"))
builder.set_entry_point("a")
builder.add_node("b", ReturnNodeValue("I'm B"))
builder.add_node("b2", ReturnNodeValue("I'm B2"))
builder.add_node("c", ReturnNodeValue("I'm C"))
builder.add_node("d", ReturnNodeValue("I'm D"))
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "b2")
builder.add_edge("b2", "d")
builder.add_edge("c", "d")
builder.set_finish_point("d")
graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

graph.invoke({"aggregate": []})

child_builder = StateGraph(State)
child_builder.set_entry_point("b")
child_builder.add_node("b", ReturnNodeValue("I'm B"))
child_builder.add_node("b2", ReturnNodeValue("I'm B2"))
child_builder.add_edge("b", "b2")
child_builder.set_finish_point("b2")

builder = StateGraph(State)
builder.add_node("a", ReturnNodeValue("I'm A"))
builder.set_entry_point("a")
builder.add_node("child", child_builder.compile())
builder.add_node("c", ReturnNodeValue("I'm C"))
builder.add_node("d", ReturnNodeValue("I'm D"))
builder.add_edge("a", "child")
builder.add_edge("a", "c")
builder.add_edge("child", "d")
builder.add_edge("c", "d")
builder.set_finish_point("d")
graph = builder.compile()
from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png())

graph.invoke({"aggregate": []}))`

[hinthornw]

Thanks for starting this discussion ,and for the detailed example! This is something @nfcampos and I have debated a bit - will follow up with you on the best recommendations.

If I understand correctly, the desired behavior would be something like a "barrier (or in python threading)" rather than the current supersteps. E.g., you'd want fan-out, and then for the fan-in node you'd like for it to wait until all branches converge on that node (or something similar)

[Freezaa9]

Yes exactly @hinthornw !

[nfcampos]

hi @Freezaa9 I think you can achieve what you want with

builder = StateGraph(State)
builder.add_node("a", ReturnNodeValue("I'm A"))
builder.set_entry_point("a")
builder.add_node("b", ReturnNodeValue("I'm B"))
builder.add_node("b2", ReturnNodeValue("I'm B2"))
builder.add_node("c", ReturnNodeValue("I'm C"))
builder.add_node("d", ReturnNodeValue("I'm D"))
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "b2")
builder.add_edge(["b2", "c"], "d")
builder.set_finish_point("d")
graph = builder.compile()

[Freezaa9]

Thanks @nfcampos this make sens and work great.
I will send a PR for the branching notebook to add this if you agree (https://github.com/langchain-ai/langgraph/blob/main/examples/branching.ipynb)

[jasontjahjono]

@nfcampos it seems like when I look at the langsmith, it looks like b and c runs parallelly, but b2 waits for c to finish before running. How do I make such that b2 does not wait for c to finish (just needs to wait for b to finish)

In my experiment, i put b to execute in 0.5s, c sleeps for 7s then continue, and b2 sleeps for 4 seconds then continue. Expected execution time = 7s, but it's 11s (7+4). is this expected?

https://smith.langchain.com/public/a4c61b59-4217-4dbf-b832-e00bb8fa0bb0/r

[gambastyle]

Ya I have the same question,
What if I wanted b2 to continue when b finished (not limited by c)
and then b and b2 finish with c
but you cant add a list as the second argument
any insights?

[Freezaa9]

You welcome @minki-j ! The solution has been added in the official documentation: https://github.com/langchain-ai/langgraph/blob/main/docs/docs/how-tos/branching.ipynb

May be @jasontjahjono @gambastyle you're looking for the "Stable Sorting" part of the above documentation / notebook ?

cf

[minki-j]

@Freezaa9 Thanks for asking this issue. I'm facing the same issue and got a lot of help from this thread. @nfcampos 's solution is succinct and worked for my case as well.

[Freezaa9]

You welcome @minki-j !
The solution has been added in the official documentation: https://github.com/langchain-ai/langgraph/blob/main/docs/docs/how-tos/branching.ipynb

May be @jasontjahjono @gambastyle you're looking for the "Stable Sorting" part of the above documentation / notebook ?