How to make multiple branches execute in parallel with conditional branches

[fuyuku]

import operator
from collections.abc import Sequence
from typing import Annotated
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict


class State(TypedDict):
    aggregate: Annotated[list, operator.add]
    which: str


class ReturnNodeValue:
    def __init__(self, node_secret: str) -> None:
        self._value = node_secret

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


builder = StateGraph(State)
builder.add_node("a", ReturnNodeValue("I'm A"))
builder.add_edge(START, "a")
builder.add_node("b", ReturnNodeValue("I'm B"))
builder.add_node("c", ReturnNodeValue("I'm C"))
builder.add_node("d", ReturnNodeValue("I'm D"))
builder.add_node("e", ReturnNodeValue("I'm E"))
builder.add_edge("a", "d")

builder.add_conditional_edges("a", lambda _: True, {True: "b", False: "c"})
builder.add_edge(["b", "c", "d"], "e")

builder.add_edge("e", END)
graph = builder.compile()
print(graph.invoke({"aggregate": []}))

output:

Adding I'm A to []
Adding I'm D to ["I'm A"]
Adding I'm B to ["I'm A"]
{'aggregate': ["I'm A", "I'm B", "I'm D"]}

---

builder.add_conditional_edges("a", lambda _: True, {True: "b", False: "c"})
builder.add_edge(["b", "c", "d"], "e")

Since "c" will never execute, it will always be stuck here
'e' will never be executed

In reality, this is often dynamic, so we can only add them all.

But there is always one between 'b' and 'c' that won't execute, causing this place to wait forever.

Is there a way to change here, delete the one that will never be executed, or any other method？

[vbarda]

Could you please provide an example graph and say more about what you're trying to achieve? Also, will turn this into a discussion since there is no explicit issue

[fuyuku]

import operator
from collections.abc import Sequence
from typing import Annotated
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict


class State(TypedDict):
    aggregate: Annotated[list, operator.add]
    which: str


class ReturnNodeValue:
    def __init__(self, node_secret: str) -> None:
        self._value = node_secret

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


builder = StateGraph(State)
builder.add_node("a", ReturnNodeValue("I'm A"))
builder.add_edge(START, "a")
builder.add_node("b", ReturnNodeValue("I'm B"))
builder.add_node("c", ReturnNodeValue("I'm C"))
builder.add_node("d", ReturnNodeValue("I'm D"))
builder.add_node("e", ReturnNodeValue("I'm E"))
builder.add_edge("a", "d")

builder.add_conditional_edges("a", lambda _: True, {True: "b", False: "c"})
builder.add_edge(["b", "c", "d"], "e")

builder.add_edge("e", END)
graph = builder.compile()
print(graph.invoke({"aggregate": []}))

output:

Adding I'm A to []
Adding I'm D to ["I'm A"]
Adding I'm B to ["I'm A"]
{'aggregate': ["I'm A", "I'm B", "I'm D"]}

---

builder.add_conditional_edges("a", lambda _: True, {True: "b", False: "c"})
builder.add_edge(["b", "c", "d"], "e")

Since "c" will never execute, it will always be stuck here
'e' will never be executed

In reality, this is often dynamic, so we can only add them all.

But there is always one between 'b' and 'c' that won't execute, causing this place to wait forever.

Is there a way to change here, delete the one that will never be executed, or any other method？