在上一节中，我们了解了 LangGraph 的核心原语：状态、节点、边和命令。有了这些原语，我们就可以构建简单的线性流程。但在实际的智能体系统中，流程往往不是简单的线性执行，而是会根据不同的情况和条件，出现分支、并行、循环等复杂的控制流。

本节我们将重点探讨 LangGraph 中的流程控制机制，特别是如何利用 LangGraph 实现并行分支 (Parallel Branching)，构建能够并发执行多个任务、提高系统效率的智能体系统。

在最简单的 LangGraph 流程中，我们使用普通边将节点串联起来，形成一条线性的执行路径。
但线性流程有其局限性：

• •无法处理条件判断：线性流程无法根据状态或外部条件，动态地选择不同的执行路径
• •无法实现并行执行：线性流程只能串行地执行节点，无法同时执行多个独立的任务
• •效率较低：对于一些可以并行处理的任务，线性流程的串行执行方式会浪费计算资源

为了克服这些局限性，LangGraph 提供了强大的流程控制机制，其中分支 (Branching) 和 并发 (Concurrency) 是核心要素。

1. 并行分支：扇出 (Fan-out) 与扇入 (Fan-in)

LangGraph 实现并行分支的核心机制是扇出 (Fan-out) 和 扇入 (Fan-in)。

• •扇出 (Fan-out)：从一个节点出发，同时触发多个下游节点，使得流程并行地向多个方向分支
• •扇入 (Fan-in)：将多个并行分支的流程汇聚到一个共同的下游节点，实现并行流程的同步和汇合

要实现扇出，最简单的方式是为一个节点添加多个出边，将该节点同时连接到多个下游节点。当 LangGraph 执行到该节点时，会同时、并发地触发所有出边指向的下游节点。

示例 1-1：扇出和扇入的完整并行分支流程

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

# 定义状态，使用 operator.add Reducer 处理并行分支的状态更新
class ParallelState(TypedDict):
    aggregate: Annotated[list, operator.add]

def node_a(state: ParallelState):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"]}

def node_b(state: ParallelState):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}

def node_c(state: ParallelState):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}

def node_d(state: ParallelState):
    print(f'Adding "D" to {state["aggregate"]}')
    return {"aggregate": ["D"]}

# 构建并行分支图
builder = StateGraph(ParallelState)

# 添加节点
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("node_c", node_c)
builder.add_node("node_d", node_d)

# 添加边：实现扇出和扇入
builder.add_edge(START, "node_a")  # 从 START 到 node_a
builder.add_edge("node_a", "node_b")  # node_a 扇出到 node_b
builder.add_edge("node_a", "node_c")  # node_a 扇出到 node_c (实现扇出)
builder.add_edge("node_b", "node_d")  # node_b 扇入到 node_d
builder.add_edge("node_c", "node_d")  # node_c 扇入到 node_d (实现扇入)
builder.add_edge("node_d", END)  # node_d 到 END

# 编译图
parallel_graph = builder.compile()

print("并行分支图构建完成")
print("流程: START -> node_a -> (node_b, node_c 并行) -> node_d -> END")

# 执行并行分支流程
print("\n执行并行分支流程:")
result = parallel_graph.invoke({"aggregate": []})
print(f"\n最终结果: {result}")

并行分支图构建完成
流程: START -> node_a -> (node_b, node_c 并行) -> node_d -> END

执行并行分支流程:
Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "D" to ['A', 'B', 'C']

最终结果: {'aggregate': ['A', 'B', 'C', 'D']}

💡 并行分支核心概念解析：

• •扇出效果：节点 A 执行完后，同时触发节点 B 和 C，两个节点并发执行
• •扇入同步：节点 D 必须等待节点 B 和 C 都执行完成后才开始执行，起到同步点作用
• •状态 Reducer：使用operator.addReducer 确保并行分支的状态更新能正确合并
• •执行效率：并行执行比串行执行节省时间，如果 B 需要 3 秒，C 需要 5 秒，并行执行总共只需 5 秒

可以看到执行结果中，节点 B 和 C 是并发执行的，然后节点 D 汇聚了两个分支的结果。

2. 并发 (Concurrency) 而非并行 (Parallelism)

需要明确的是，LangGraph 实现的并非真正的并行 (Parallelism)，而是并发 (Concurrency)。理解这个区别对于正确使用 LangGraph 非常重要：

并发 vs 并行的本质区别

• •并行 (Parallelism)：同时执行多个独立的任务，需要多个物理计算资源（多核 CPU、多台机器）真正地同时运行不同的代码
• •并发 (Concurrency)：在单计算资源上，"看似同时"执行多个任务，通过时间片轮转或异步 IO，CPU 在多个任务之间快速切换

LangGraph 的 Superstep 执行模型

LangGraph 借鉴了 Apache Pregel 分布式图计算框架的 Superstep (超步) 概念：

• •Superstep：图执行的基本迭代单元，代表 LangGraph 图执行过程中的一个"步骤"
• •并发执行：在每个 Superstep 中，LangGraph 会尽可能地并发执行所有"就绪"的节点
• •同步屏障：当 Superstep 内的所有节点都执行完成后，进行全局同步，应用状态更新，进入下一个 Superstep

示例 2-1：Superstep 执行模式演示

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

class SuperstepState(TypedDict):
    superstep_log: Annotated[list, operator.add]
    timestamp: str

def superstep_node_a(state: SuperstepState):
    print(f"Superstep 1: 执行节点 A")
    time.sleep(0.1)  # 模拟处理时间
    return {"superstep_log": ["Superstep 1: Node A executed"]}

def superstep_node_b(state: SuperstepState):
    print(f"Superstep 2: 执行节点 B (并发)")
    time.sleep(0.2)  # 模拟处理时间
    return {"superstep_log": ["Superstep 2: Node B executed"]}

def superstep_node_c(state: SuperstepState):
    print(f"Superstep 2: 执行节点 C (并发)")
    time.sleep(0.15)  # 模拟处理时间
    return {"superstep_log": ["Superstep 2: Node C executed"]}

def superstep_node_d(state: SuperstepState):
    print(f"Superstep 3: 执行节点 D (同步后)")
    time.sleep(0.1)  # 模拟处理时间
    return {"superstep_log": ["Superstep 3: Node D executed"]}

# 构建 Superstep 演示图
builder = StateGraph(SuperstepState)

# 添加节点
builder.add_node("superstep_a", superstep_node_a)
builder.add_node("superstep_b", superstep_node_b) 
builder.add_node("superstep_c", superstep_node_c)
builder.add_node("superstep_d", superstep_node_d)

# 添加边：演示 Superstep 执行
builder.add_edge(START, "superstep_a")          # Superstep 1: A 执行
builder.add_edge("superstep_a", "superstep_b")  # Superstep 2: A -> B (并发)
builder.add_edge("superstep_a", "superstep_c")  # Superstep 2: A -> C (并发) 
builder.add_edge("superstep_b", "superstep_d")  # Superstep 3: B -> D (同步)
builder.add_edge("superstep_c", "superstep_d")  # Superstep 3: C -> D (同步)
builder.add_edge("superstep_d", END)

superstep_graph = builder.compile()

print("Superstep 执行模式演示:")
print("Superstep 1: 执行 A")
print("Superstep 2: 并发执行 B 和 C (等待同步)")  
print("Superstep 3: 同步后执行 D")
print()

# 执行并记录时间
start_time = time.time()
result = superstep_graph.invoke({"superstep_log": []})
end_time = time.time()

print(f"\\n执行完成，总时间: {end_time - start_time:.3f} 秒")
print("执行日志:")
for log in result["superstep_log"]:
    print(f"  - {log}")

Superstep 执行模式演示:
Superstep 1: 执行 A
Superstep 2: 并发执行 B 和 C (等待同步)
Superstep 3: 同步后执行 D

Superstep 1: 执行节点 A
Superstep 2: 执行节点 B (并发)
Superstep 2: 执行节点 C (并发)
Superstep 3: 执行节点 D (同步后)
\n执行完成，总时间: 0.422 秒
执行日志:
  - Superstep 1: Node A executed
  - Superstep 2: Node B executed
  - Superstep 2: Node C executed
  - Superstep 3: Node D executed

💡 Superstep 核心特性：

• •并发性：同一个 Superstep 内的多个节点可以并发执行，提高执行效率
• •同步性：每个 Superstep 结束时有全局同步点，保证状态更新的原子性和一致性
• •迭代性：图执行是 Superstep 的迭代过程，每个 Superstep 在前一个基础上计算

从上面的例子可以看到：

• • Superstep 1: 只有节点 A 执行
• • Superstep 2: 节点 B 和 C 并发执行（等待较慢的任务完成）
• • Superstep 3: 同步后节点 D 执行

总执行时间约等于每个 Superstep 中最慢节点的时间之和，而非所有节点时间的总和。

3. 递归限制与并行分支

递归限制（Recursion Limit）用于限制 LangGraph 图执行过程中的最大 Superstep 迭代次数，防止图无限循环执行。

重要特性

• •计数单位：递归限制的计数单位是 Superstep，而不是节点
• •并行不影响计数：无论一个 Superstep 内部并发执行了多少个节点，都只计为一次 Superstep 迭代
• •防止无限循环：有效防止图陷入无限循环，保护计算资源

示例 3-1：递归限制是对并行分支流程的限制

import operator
from typing import Annotated, Any
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.errors import GraphRecursionError  # 导入 GraphRecursionError

class State(TypedDict):
    # operator.add 是状态归约器，确保状态键 aggregate 为 append-only 列表
    aggregate: Annotated[list, operator.add]

def node_a(state):
    print("🔄 执行节点 A")
    return {"aggregate": ["I'm A"]}

def node_b(state):
    print("🔄 执行节点 B")
    return {"aggregate": ["I'm B"]}

def node_c(state):
    print("🔄 执行节点 C")
    return {"aggregate": ["I'm C"]}

def node_d(state):
    print("🔄 执行节点 D")
    return {"aggregate": ["I'm D"]}

# 构建图
print("🏗️ 构建并行分支图...")
builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_node("c", node_c)
builder.add_node("d", node_d)

# 添加边 - 创建并行分支结构
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")  # 从 a 扇出到 b 和 c
builder.add_edge("b", "d")
builder.add_edge("c", "d")  # b 和 c 扇入到 d
builder.add_edge("d", END)

graph = builder.compile()
print("✅ 图构建完成！")

print("\n=== 递归限制和并行分支流程的限制 ===\n")

# 测试 1: 正常执行（在递归限制内）
print("📊 测试 1: 正常执行 (recursion_limit=10)")
try:
    result = graph.invoke({"aggregate": []}, {"recursion_limit": 10})
    print(f"✅ 执行成功，最终结果: {result}")
except Exception as e:
    print(f"❌ 执行失败: {e}")

print("\n" + "="*60)

# 测试 2: 设置较低的递归限制来触发 GraphRecursionError
print("📊 测试 2: 递归限制过低 (recursion_limit=3)")
try:
    # 设置 recursion_limit=3，少于完整流程所需的 Superstep 数量
    result = graph.invoke({"aggregate": []}, {"recursion_limit": 3})
    print(f"✅ 执行成功，结果: {result}")
except GraphRecursionError as e:
    print(f"⚠️ 捕获 GraphRecursionError 异常: {e}")
    print("📝 说明: 递归限制过低，无法完成完整的并行分支流程")


🏗️ 构建并行分支图...
✅ 图构建完成！

=== 递归限制和并行分支流程的限制 ===

📊 测试 1: 正常执行 (recursion_limit=10)
🔄 执行节点 A
🔄 执行节点 C
🔄 执行节点 B
🔄 执行节点 D
✅ 执行成功，最终结果: {'aggregate': ["I'm A", "I'm B", "I'm C", "I'm D"]}

============================================================
📊 测试 2: 递归限制过低 (recursion_limit=3)
🔄 执行节点 A
🔄 执行节点 B
🔄 执行节点 C
🔄 执行节点 D
⚠️ 捕获 GraphRecursionError 异常: Recursion limit of 3 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.
For troubleshooting, visit: https://python.langchain.com/docs/troubleshooting/errors/GRAPH_RECURSION_LIMIT
📝 说明: 递归限制过低，无法完成完整的并行分支流程

在设计包含并行分支的 LangGraph 流程时，需要根据流程的复杂度合理地设置recursion_limit参数。

• • 如果设置得过低，则可能导致流程在并行分支执行完成前就因为达到限制而中止，抛出GraphRecursionError异常。
• • 如果设置得过高，则可能无法有效防止流程陷入无限循环，造成不必要的计算资源消耗。
• • 建议通过实验和调优，根据流程的实际迭代次数和复杂度选择一个合适的recursion_limit值。

总结

通过本章的介绍，用户可以了解到如何在LangGraph中创建并行节点执行的分支，利用fan-out和fan-in机制、条件分支以及稳定排序的技术来实现高效的工作流。通过示例代码，用户能够深入理解这些概念，并将在实际应用中得到有效的运用。

参考文献和资料
• LangGraph文档：https://docs.langchain.com/oss/python/langgraph/overview
• 《LangGraph实战》

微信公众号（关注获取最新资讯）：