LangGraph Parallel Execution
Fan-out · Fan-in · reducer

1What is parallel execution? Why is it needed?

In the first three modules of LangGraph, the graphs we study arelinear: After node A is executed, node B starts. After node B is executed, node C starts. This "single lane" execution method is simple and predictable, but has a natural bottleneck:Each step must wait for the previous step to complete。

In real-life scenarios, many tasks areindependent of each otherYes, it can be done at the same time. For example:

• Retrieve data from Wikipedia and Web search engines simultaneously, then let LLM combine the two data to answer questions
• Perform sentiment analysis and keyword extraction on an article at the same time, and then merge the results
• Call multiple external APIs at the same time and aggregate the results

If these independent tasks are executed serially, the total time consumption = task 1 time + task 2 time + ...; and ifParallel execution, the total time taken ≈ the time of the slowest task. When each subtask needs to wait for LLM or network IO, the speedup effect brought by parallelization is very significant.

What do Fan-out and Fan-in mean?

These two terms come from circuit design and have the following meanings in LangGraph:

2Linear Graphs vs. Parallel Graphs: From Sequential Execution to Fan-out / Fan-in

Let’s look at the linear graph first: covering the state one by one

The notebook first uses a linear graph for baseline comparison. Each node puts its own valuecoverdropstateFields (note: no reducer here):

from typing import Any, List
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    # 注意：没有 Reducer，默认行为是"覆盖"
    state: List[str]

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['state']}")
        return {"state": [self._value]}  # 直接覆盖，不追加

# 构建线性图：a → b → c → d
builder = StateGraph(State)
builder.add_node("a", ReturnNodeValue("I'm 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_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", "c")
builder.add_edge("c", "d")
builder.add_edge("d", END)
graph = builder.compile()

The running results are as follows. Because every nodecoverThe value of the previous node, only the value of D is left in the final state:

Change to parallel graph: Fan-out to b and c, Fan-in to d

Now change b and c to be executed in parallel. The key changes are onlyEdge connection method——FromaConnect simultaneouslybandc, thenbandcAll even tod：

# Fan-out：a → b 且 a → c（两条边同时从 a 出发）
builder.add_edge("a", "b")
builder.add_edge("a", "c")

# Fan-in：b → d 且 c → d（两条边汇聚到 d）
builder.add_edge("b", "d")
builder.add_edge("c", "d")

3Parallel conflict: the same field is written by multiple nodes at the same time

If the parallel graph directly uses State without reducer,Will report an error. This is an important security mechanism of LangGraph, and understanding it is critical.

Why the conflict?

In parallel steps,bandc Execute simultaneously within the same "step", each of them has tostateWrite a value to the field. Without the reducer, LangGraph doesn't know which one to keep - the two writes conflict with each other, so it throws an exception instead of quietly discarding one of them.

from langgraph.errors import InvalidUpdateError

try:
    graph.invoke({"state": []})
except InvalidUpdateError as e:
    print(f"An error occurred: {e}")

The root cause of the error: LangGraph’s step concept

LangGraph divides the execution of the graph into "steps". In a parallel scenario, multiple nodes may run concurrently in the same step. When these nodes all update the same State field, LangGraph needs an explicit rule to decide how to merge these updates - this rule isReducer。

4reducer: a safe merging mechanism for parallel output

reducer is one attached to the State fieldmerge function. When multiple nodes write values ​​to the same field in the same step, LangGraph will use the reducer function to combine all written values ​​into a final value.

use operator.addas reducer

Python standard libraryoperator.addApplying to a list is equivalent to list splicing (list1 + list2). Using it as a reducer, the output lists of all parallel nodes will be concatenated in sequence:

import operator
from typing import Annotated

class State(TypedDict):
    # Annotated 的第二个参数是 Reducer 函数
    # operator.add 作用于列表时 = 列表拼接（追加）
    state: Annotated[list, operator.add]

# 其余代码（节点定义、边连接）与之前完全相同
builder = StateGraph(State)
builder.add_node("a", ReturnNodeValue("I'm 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"))

# Fan-out + Fan-in
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "d")
builder.add_edge("c", "d")
builder.add_edge("d", END)
graph = builder.compile()

Now it runs successfully:

Illustration of the working principle of reducer

5Unequal length branches: How LangGraph waits for all parallel paths to complete

In actual scenarios, multiple parallel branches often have different lengths—for example, one branch has only one node, and another branch has two nodes connected in series. How does LangGraph handle this situation?

Construct parallel graphs of unequal lengths

In this example, fromaFan-out has two branches: branch one is b → b2 (two nodes connected in series), branch two is c (only one node). Both branches are Fan-in.d：

builder = StateGraph(State)

builder.add_node("a",  ReturnNodeValue("I'm 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(START, "a")
builder.add_edge("a", "b")    # 支路一：a → b → b2
builder.add_edge("a", "c")    # 支路二：a → c
builder.add_edge("b", "b2")

# 列表语法：等 b2 和 c 都完成后，才触发 d
builder.add_edge(["b2", "c"], "d")
builder.add_edge("d", END)
graph = builder.compile()

Execution result:

Key observation: LangGraph's "step" scheduling mechanism

As can be seen from the output:

• Step 1:aexecute
• Step 2:bandc At the same timeexecution (since they are bothadirect successor)
• Step 3:b2execute(btriggered after completion), whilecCompleted waiting
• Step 4:dExecution——only ifb2andcTriggered after everything is completed

6Custom reducer: Control the merging order of parallel outputs

use operator.addWhen merging parallel output, the order in which each node writesDetermined by LangGraph internal scheduling, unpredictable. Sometimes we want the merged results to have a certain order, which requiresCustom reducer。

Problem description

Output from previous section["I'm A", "I'm B", "I'm C", "I'm B2", "I'm D"], C comes before B2, but we probably want the results to be sorted alphabetically.

Write a custom reducer function

def sorting_reducer(left, right):
    """合并两个列表并按字母顺序排序"""
    # 确保 left 和 right 都是列表
    if not isinstance(left, list):
        left = [left]
    if not isinstance(right, list):
        right = [right]
    # 合并后排序
    return sorted(left + right, reverse=False)

class State(TypedDict):
    # 使用自定义 Reducer
    state: Annotated[list, sorting_reducer]

The other code (definition of nodes and edges) is the same as the previous section. Just replace the reducer in the State definition:

Pay attention to the end result["I'm A", "I'm B", "I'm B2", "I'm C", "I'm D"]Alphabetically sorted - even if C is printed before B during execution, the sorting reducer ensures that the final result is ordered.

Signature convention of reducer function

7Real case: parallel search on Wikipedia + Web and then use LLM to summarize answers

The previous examples all used simple strings to demonstrate the parallel mechanism. Now let’s use a real AI scenario to demonstrate the practical value of parallel execution:Data is retrieved from Wikipedia and Web search engines at the same time, and then LLM combines the two data to answer user questions.。

State definition

import operator
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    question: str              # 用户问题（不需要 Reducer，只写一次）
    answer:   str              # LLM 的最终回答
    context:  Annotated[list, operator.add]  # 检索到的资料（需要 Reducer 追加）

contextThe field will besearch_webandsearch_wikipediaTwo parallel nodes write at the same time, so they must be addedoperator.add Reducer。questionandanswerWill only be written by one node, no reducer is required.

Complete implementation of three nodes

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_community.document_loaders import WikipediaLoader
from langchain_tavily import TavilySearch
from langchain_deepseek import ChatDeepSeek

llm = ChatDeepSeek(model="deepseek-v4-pro", temperature=0)

# ──── 节点 1：Web 搜索 ────
def search_web(state):
    """从 Web 搜索引擎检索资料"""
    tavily_search = TavilySearch(max_results=3)
    data = tavily_search.invoke({"query": state['question']})
    search_docs = data.get("results", data)

    # 格式化为带 URL 的文档字符串
    formatted = "\n\n---\n\n".join([
        f'<Document href="../../../langGraph/module-4/{doc["url"]}">\n{doc["content"]}\n</Document>'
        for doc in search_docs
    ])
    return {"context": [formatted]}  # 返回列表，Reducer 会追加

# ──── 节点 2：Wikipedia 搜索 ────
def search_wikipedia(state):
    """从 Wikipedia 检索资料"""
    search_docs = WikipediaLoader(
        query=state['question'], load_max_docs=2
    ).load()

    # 格式化为带来源的文档字符串
    formatted = "\n\n---\n\n".join([
        f'<Document source="{doc.metadata["source"]}">\n{doc.page_content}\n</Document>'
        for doc in search_docs
    ])
    return {"context": [formatted]}  # 返回列表，Reducer 会追加

# ──── 节点 3：LLM 综合回答 ────
def generate_answer(state):
    """基于检索到的上下文，让 LLM 回答问题"""
    context  = state["context"]   # 已合并的两份资料
    question = state["question"]

    answer_template = "Answer the question {question} using this context: {context}"
    instructions = answer_template.format(question=question, context=context)

    answer = llm.invoke([
        SystemMessage(content=instructions),
        HumanMessage(content="Answer the question.")
    ])
    return {"answer": answer}

Build parallel graphs

builder = StateGraph(State)

builder.add_node("search_web",       search_web)
builder.add_node("search_wikipedia", search_wikipedia)
builder.add_node("generate_answer",  generate_answer)

# 从 START 同时 Fan-out 到两个搜索节点（并行）
builder.add_edge(START, "search_wikipedia")
builder.add_edge(START, "search_web")

# 两个搜索节点都完成后，Fan-in 到 generate_answer
builder.add_edge("search_wikipedia", "generate_answer")
builder.add_edge("search_web",       "generate_answer")
builder.add_edge("generate_answer",  END)

graph = builder.compile()

Run the example

result = graph.invoke({"question": "How were Nvidia's Q2 2025 earnings"})
print(result['answer'].content)

The core value embodied in this case

• 1

  Parallel acceleration

  Web search and Wikipedia searchsimultaneously, the total time taken is approximately equal to the slower of the two, rather than the sum of the two.

• 2

  Automatic merge

  The results of both search nodes are passedoperator.addreducer automatically splices intocontextlist,generate_answerWhat the node receives isTwo complete documents, no need to merge manually.

• 3

  Higher quality information

  LLM's web-based timely content + Wikipedia's authoritative knowledge base provide comprehensive answers from two sources, which are more comprehensive and accurate than a single source.

8LangGraph Studio deployment and API calls

The constructed parallel graph can be directly displayed in the notebookinvoke, can also be deployed to LangGraph API Server and called asynchronously through HTTP interface or SDK.

Start LangGraph Dev Server locally

existmodule-4/studio/Execute in the directory:

# 在终端里运行（不是 Python 代码）
langgraph dev

Open the Studio UI address in the browser and you can visually see the graph structure, real-time execution status, and input/output of each node, which is very useful when debugging complex parallel graphs.

Asynchronous streaming calls via SDK

from langgraph_sdk import get_client

client = get_client(url="http://127.0.0.1:2024")

# 创建新线程（每个线程有独立的状态）
thread = await client.threads.create()

input_question = {"question": "How were Nvidia Q2 2025 earnings?"}

# 流式订阅执行过程，stream_mode="values" 表示每个步骤后推送完整 State
async for event in client.runs.stream(
    thread["thread_id"],
    assistant_id="parallelization",  # 对应 studio 里的图名称
    input=input_question,
    stream_mode="values"
):
    # 检查 answer 字段是否已经出现
    if event.data is not None:
        answer = event.data.get('answer', None)
        if answer:
            print(answer['content'])

Why use API Server instead of invoke directly?

Summary of knowledge panorama in this section