parallel-execution-in-langgraph

Parallel Execution in LangGraph

Parallel execution allows multiple nodes to process state concurrently, with results automatically merged using state reducers.

Fan-Out/Fan-In Pattern

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Overwrite
import operator

class State(TypedDict):
    results: Annotated[list, operator.add]  # Accumulates results via concatenation
    input_data: str

def branch_1(state: State) -> dict:
    return {"results": [1, 2, 3]}

def branch_2(state: State) -> dict:
    return {"results": [4, 5, 6]}

def branch_3(state: State) -> dict:
    return {"results": [7, 8, 9]}

def combine_results(state: State) -> dict:
    """
    Process aggregated results from parallel branches.

    Note: Results are already combined by the state reducer (operator.add).
    This node performs post-processing (sorting) and uses Overwrite to
    replace the accumulated results rather than appending to them.
    """
    sorted_results = sorted(state["results"])
    return {"results": Overwrite(value=sorted_results)}

def fan_out(state: State) -> list[str]:
    """Route to multiple branches for parallel execution."""
    return ["branch_1", "branch_2", "branch_3"]

workflow = StateGraph(State)
workflow.add_node("branch_1", branch_1)
workflow.add_node("branch_2", branch_2)
workflow.add_node("branch_3", branch_3)
workflow.add_node("combine", combine_results)

# Fan-out from START to parallel branches (path_map ensures correct visualization)
workflow.add_conditional_edges(START, fan_out, ["branch_1", "branch_2", "branch_3"])

# Fan-in: all branches must complete before combine (list syntax)
workflow.add_edge(["branch_1", "branch_2", "branch_3"], "combine")
workflow.add_edge("combine", END)

# Compile the graph before invocation
graph = workflow.compile()

# Invoke the graph
result = graph.invoke({"input_data": "test", "results": []})

Result: result["results"] = [1, 2, 3, 4, 5, 6, 7, 8, 9] (sorted by combine_results after accumulation)