LangGraph Module 8: Parallel Branches and Subgraphs

Step 1: Fan out to parallel nodes

If a node has several outgoing edges, those successors run in the same step, in parallel. Because both write to the same state key, that key needs a reducer to combine their updates.

from operator import add
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    topic: str
    notes: Annotated[list[str], add]
    report: str

def research_web(state: State) -> dict:
    return {"notes": ["web: facts about " + state["topic"]]}

def research_docs(state: State) -> dict:
    return {"notes": ["docs: facts about " + state["topic"]]}

Step 2: Join the branches

Add edges from both parallel nodes into one downstream node. It runs after both have finished, so it sees all their results. The order of items accumulated from parallel nodes is not guaranteed, so sort them if order matters.

def write_report(state: State) -> dict:
    return {"report": "\n".join(sorted(state["notes"]))}

builder = StateGraph(State)
builder.add_node("research_web", research_web)
builder.add_node("research_docs", research_docs)
builder.add_node("write_report", write_report)

builder.add_edge(START, "research_web")
builder.add_edge(START, "research_docs")
builder.add_edge("research_web", "write_report")
builder.add_edge("research_docs", "write_report")
builder.add_edge("write_report", END)

graph = builder.compile()
print(graph.invoke({"topic": "LangGraph", "notes": [], "report": ""})["report"])

Step 3: Package logic as a subgraph

A compiled graph can be used as a node inside another graph. Build and compile the inner graph with its own nodes and edges, then add it to the parent by name. When the parent and child share state keys, data flows between them automatically.

sub_builder = StateGraph(State)
sub_builder.add_node("research_web", research_web)
sub_builder.add_node("research_docs", research_docs)
sub_builder.add_edge(START, "research_web")
sub_builder.add_edge(START, "research_docs")
research_graph = sub_builder.compile()

parent = StateGraph(State)
parent.add_node("research", research_graph)   # a compiled graph as a node
parent.add_node("write_report", write_report)
parent.add_edge(START, "research")
parent.add_edge("research", "write_report")
parent.add_edge("write_report", END)
main_graph = parent.compile()

Step 4: Use agents as nodes

Because LangChain's create_agent returns a compiled graph, you can drop an agent into a larger graph as a node, for example a researcher agent feeding a writer step. To see an inner agent's streamed tokens in the outer graph, enable subgraph streaming when you stream.

Step 5: Know about dynamic fan-out

When the number of parallel tasks is not known in advance, such as one task per document, a routing function can return Send objects, each naming a node and the state to give it. This is the map-reduce pattern. Send is imported from langgraph.types, and the Graph API documentation covers its details.