深入 ‘Parallel Execution’ (Fan-out/Fan-in)：如何在 LangGraph 中实现多个专家节点的同步并行与结果汇聚？

深入 LangGraph 中的并行执行：实现多专家节点的同步并行与结果汇聚

在构建复杂的人工智能应用时，我们经常面临一个挑战：如何有效地整合多个专业领域的知识，并以高效、可扩展的方式处理用户请求。传统的顺序执行流程可能导致响应延迟，尤其是在涉及多个耗时操作（如调用不同的大型语言模型、查询外部数据库或执行复杂计算）时。

LangGraph，作为LangChain生态系统中的一个强大工具，通过其图结构和状态管理能力，为解决这一问题提供了优雅的方案。今天，我们将深入探讨如何在LangGraph中实现“并行执行”模式，特别是“扇出”（Fan-out）和“扇入”（Fan-in）机制，从而实现多个专家节点的同步并行处理与结果的智能汇聚。

一、并行执行的必要性与 Fan-out/Fan-in 模式概览

1.1 为什么我们需要并行执行？

想象一个复杂的客户服务场景：用户提交了一个问题。为了提供最佳答案，系统可能需要：

• 分析用户意图： 识别问题是关于销售、技术支持还是账单。
• 查询知识库： 针对识别出的意图，并行查询多个专业知识库。
• 调用外部API： 例如，检查订单状态、用户账户信息。
• 生成多角度回复： 让不同的AI专家（例如，一个销售专家、一个技术专家）分别从其专业角度生成回复草稿。

如果这些步骤都串行执行，整个流程将非常缓慢。并行执行能够显著缩短整体响应时间，提高用户体验。

1.2 Fan-out/Fan-in 模式：核心思想

Fan-out/Fan-in 模式是并行处理中的一种常见且强大的模式，它包含两个主要阶段：

• Fan-out (扇出)： 在此阶段，一个单一的输入或任务被分解成多个独立的子任务，并分发给多个处理单元（在LangGraph中，即多个专家节点）并行执行。这些子任务通常是独立的，或至少是可以在不相互依赖的情况下同时进行的。
• Fan-in (扇入)： 在此阶段，等待所有并行执行的子任务完成。一旦所有子任务的结果都可用，它们将被收集、整合、分析或汇总，以生成一个最终的、统一的结果。

这种模式的优势在于它能够最大化并行度，并在需要整合多源信息时提供一个清晰的结构。

二、LangGraph 基础回顾与异步编程的重要性

在深入实现之前，我们快速回顾LangGraph的核心概念，并强调异步编程在实现真正并行中的关键作用。

2.1 LangGraph 核心概念

• StateGraph： LangGraph 的核心，它定义了一个有向图，其中的节点表示计算步骤，边表示状态的流转。StateGraph 管理一个共享的 state 对象，节点通过修改这个 state 来进行通信。
• 节点 (Nodes)： 图中的基本计算单元。每个节点都是一个函数或可调用对象，它接收当前 state 作为输入，并返回修改后的 state。
• 边 (Edges)： 连接节点的路径，定义了图的执行流。边可以是条件性的，根据 state 的值决定下一个执行的节点。
• TypedDict 或 Pydantic 模型作为 State： 为了清晰和类型安全，通常使用 TypedDict 或 Pydantic 模型来定义 StateGraph 的状态结构。
• 异步执行： LangGraph 内部是基于 asyncio 构建的。这意味着我们可以定义 async 节点，并利用 await 来等待异步操作，从而实现非阻塞的I/O和并行执行的潜力。

2.2 异步编程：实现并行执行的关键

在Python中，真正的并行计算通常涉及多进程。但对于I/O密集型任务（如网络请求、数据库查询、LLM API调用），asyncio 提供的协程（coroutine）是实现并发和伪并行（concurrent but not truly parallel CPU-bound tasks）的绝佳选择。当一个协程 await 一个I/O操作时，Python解释器可以切换到执行另一个协程，从而有效地利用等待时间。

在Fan-out场景中，多个专家节点可能都需要执行耗时的I/O操作。通过将这些节点定义为 async 函数，LangGraph 可以同时调度它们，并在后台等待所有I/O操作完成，显著提升效率。

三、构建 Fan-out/Fan-in 架构：一个客户服务Triage的例子

为了具体化 Fan-out/Fan-in 模式，我们构建一个客户服务Triage系统。当一个用户查询进入时，系统会：

1. 分发 (Dispatch)： 根据查询内容，识别可能相关的专家领域。
2. 并行分析 (Fan-out)： 将查询发送给多个相关的专家节点（例如，销售专家、技术支持专家、账单专家）并行处理。每个专家会生成一个独立的分析结果。
3. 结果汇聚 (Fan-in)： 等待所有专家完成分析，然后将它们的独立结果收集起来，生成一个最终的综合性回复。

3.1 定义共享状态 AgentState

首先，我们定义LangGraph的状态。这个状态将贯穿整个Triage流程，并保存所有节点产生的信息。

from typing import TypedDict, List, Dict, Any, Optional
import asyncio
import time
from langchain_core.messages import BaseMessage
from langchain_core.pydantic_v1 import Field
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    """
    Represent the state of our multi-agent customer service workflow.
    """
    query: str  # The initial customer query
    active_experts: List[str]  # List of expert names that were dispatched
    expert_results: Dict[str, str]  # Dictionary to store results from each expert {expert_name: result_string}
    final_analysis: Optional[str]  # The aggregated final response
    error: Optional[str] # To store any errors during execution

# 初始状态
initial_state: AgentState = {
    "query": "",
    "active_experts": [],
    "expert_results": {},
    "final_analysis": None,
    "error": None
}

• query: 用户的原始查询。
• active_experts: 记录当前调度了哪些专家，这对于汇聚阶段判断是否所有专家都已完成至关重要。
• expert_results: 一个字典，键是专家名称，值是该专家处理查询后返回的结果。这是并行执行的关键输出。
• final_analysis: 所有专家结果汇聚后的最终分析或回复。
• error: 用于捕获和传递流程中的错误信息。

3.2 定义专家节点 (Expert Nodes)

每个专家节点都是一个异步函数，模拟一个LLM调用或其他耗时操作。它们接收当前状态，提取相关信息，执行其专业任务，然后将结果存储回 expert_results。

为了模拟真实世界的LLM调用延迟，我们将使用 asyncio.sleep。

# 专家节点模拟
async def sales_expert_node(state: AgentState) -> AgentState:
    print(f"[{time.time():.2f}] 销售专家开始处理...")
    query = state["query"]
    # 模拟LLM调用或其他耗时操作
    await asyncio.sleep(2) # 模拟2秒延迟

    # 模拟生成销售相关回复
    sales_advice = f"销售专家建议：针对 '{query}'，我们有最新的优惠套餐和解决方案，可以为您带来更高性价比。请问您对哪个产品系列更感兴趣？"

    current_results = state["expert_results"]
    current_results["sales_expert"] = sales_advice

    print(f"[{time.time():.2f}] 销售专家处理完成。")
    return {**state, "expert_results": current_results}

async def tech_support_expert_node(state: AgentState) -> AgentState:
    print(f"[{time.time():.2f}] 技术支持专家开始处理...")
    query = state["query"]
    # 模拟LLM调用或其他耗时操作
    await asyncio.sleep(3) # 模拟3秒延迟

    # 模拟生成技术支持相关回复
    tech_advice = f"技术支持专家建议：针对 '{query}'，请您提供更详细的错误信息或产品型号。常见问题排查可以参考我们的帮助文档。我们有专门的工程师团队随时准备协助您。"

    current_results = state["expert_results"]
    current_results["tech_support_expert"] = tech_advice

    print(f"[{time.time():.2f}] 技术支持专家处理完成。")
    return {**state, "expert_results": current_results}

async def billing_expert_node(state: AgentState) -> AgentState:
    print(f"[{time.time():.2f}] 账单专家开始处理...")
    query = state["query"]
    # 模拟LLM调用或其他耗时操作
    await asyncio.sleep(1) # 模拟1秒延迟

    # 模拟生成账单相关回复
    billing_advice = f"账单专家建议：针对 '{query}'，请您提供订单号或账户信息以便查询。您可以登录用户中心查看详细账单或支付记录。对于任何费用疑问，我们都会为您核对。"

    current_results = state["expert_results"]
    current_results["billing_expert"] = billing_advice

    print(f"[{time.time():.2f}] 账单专家处理完成。")
    return {**state, "expert_results": current_results}

每个专家节点都遵循相同的模式：

1. 打印开始信息。
2. 从 state 中获取 query。
3. await asyncio.sleep(X) 模拟耗时操作。
4. 生成专家特有的回复。
5. 将回复存储到 state["expert_results"] 字典中，键是专家自己的名称。
6. 打印完成信息。
7. 返回更新后的 state。

3.3 分发节点 (Dispatcher Node) – 实现 Fan-out

分发节点的职责是根据用户的查询，决定哪些专家应该被激活，并将这些专家的名称记录到 active_experts 列表中。最关键的是，它将返回一个列表或元组的节点名称，指示 LangGraph 应该并行执行这些节点。

async def dispatcher_node(state: AgentState) -> List[str]:
    print(f"[{time.time():.2f}] 分发器开始分析查询并调度专家...")
    query = state["query"].lower()

    active_experts = []

    # 简单的关键词匹配来模拟意图识别和专家调度
    if "购买" in query or "价格" in query or "优惠" in query or "产品" in query:
        active_experts.append("sales_expert")
    if "问题" in query or "故障" in query or "技术" in query or "安装" in query:
        active_experts.append("tech_support_expert")
    if "账单" in query or "费用" in query or "支付" in query or "退款" in query:
        active_experts.append("billing_expert")

    # 如果没有匹配到特定专家，则默认调用所有或一个通用专家
    if not active_experts:
        print(f"[{time.time():.2f}] 未匹配到特定专家，默认调度所有专家。")
        active_experts = ["sales_expert", "tech_support_expert", "billing_expert"]

    print(f"[{time.time():.2f}] 分发器调度专家: {active_experts}")

    # 更新状态，记录被调度的专家
    state["active_experts"] = active_experts

    # 返回一个列表，LangGraph 会并行执行这些节点
    return active_experts

dispatcher_node 的返回类型是 List[str]。当 LangGraph 的一个节点返回一个字符串列表时，它会理解为需要并行执行列表中指定的所有节点。这是实现 Fan-out 的核心机制。

3.4 汇聚节点 (Aggregator Node) – 实现 Fan-in

汇聚节点的职责是等待所有 active_experts 的结果都出现在 expert_results 中，然后将这些结果整合生成最终的 final_analysis。由于 LangGraph 会在所有并行路径完成后自动将控制权传递给下一个节点，所以汇聚节点不需要主动“等待”；它被调用时，expert_results 应该已经包含了所有已完成专家的结果。

async def aggregator_node(state: AgentState) -> AgentState:
    print(f"[{time.time():.2f}] 汇聚器开始收集并分析专家结果...")

    active_experts = state["active_experts"]
    expert_results = state["expert_results"]
    query = state["query"]

    # 检查是否所有活跃专家的结果都已收集
    # 实际上，在LangGraph中，当一个节点有多个前驱节点时，它只会在所有前驱节点执行完毕后才会被调用。
    # 所以此处不需要额外的等待逻辑，可以直接处理结果。

    missing_results = [expert for expert in active_experts if expert not in expert_results]
    if missing_results:
        # 这通常不应该发生，除非图结构或状态更新有误
        print(f"[{time.time():.2f}] 警告：缺少以下专家的结果：{missing_results}")
        final_analysis = "抱歉，部分专家未能完成分析。请稍后再试或提供更多信息。"
    else:
        # 整合所有专家结果
        print(f"[{time.time():.2f}] 所有专家结果已收集，开始生成最终分析。")

        analysis_parts = []
        analysis_parts.append(f"针对您的查询：'{query}'，以下是我们的专家团队提供的综合分析：n")

        for expert_name in active_experts:
            result = expert_results.get(expert_name, "无结果")
            analysis_parts.append(f"--- {expert_name.replace('_', ' ').title()} ---n{result}n")

        analysis_parts.append("n希望这些信息能帮助到您！")
        final_analysis = "n".join(analysis_parts)

    print(f"[{time.time():.2f}] 汇聚器完成最终分析。")
    return {**state, "final_analysis": final_analysis}

3.5 构建 LangGraph

现在，我们将所有节点和逻辑连接起来，形成一个 StateGraph。

# 1. 构建图
workflow = StateGraph(AgentState)

# 2. 添加节点
workflow.add_node("dispatcher", dispatcher_node)
workflow.add_node("sales_expert", sales_expert_node)
workflow.add_node("tech_support_expert", tech_support_expert_node)
workflow.add_node("billing_expert", billing_expert_node)
workflow.add_node("aggregator", aggregator_node)

# 3. 设置入口点
workflow.set_entry_point("dispatcher")

# 4. 设置边
# 从分发器到各个专家（Fan-out）
# LangGraph 会自动根据dispatcher_node的返回值（List[str]）来创建这些动态边并并行执行。
# 因此，我们不需要显式地为每个可能的专家创建条件边。
# 相反，我们创建一条从dispatcher到aggregator的“虚拟”边，
# 但实际上，LangGraph的并行机制是通过dispatcher的返回值来触发的，
# 并且只有当所有被触发的专家都完成后，才会隐式地汇聚到下一个共同的节点。
# 在这里，我们将所有专家都连接到aggregator。

# 从 dispatcher 到所有潜在的专家节点，然后所有专家节点都导向 aggregator
# 这里利用了LangGraph的特性：当一个节点返回多个后续节点时，这些节点会被并行执行。
# 然后，我们可以将所有这些并行执行的节点都连接到同一个汇聚节点。
# LangGraph会确保汇聚节点只在所有前驱节点都完成后才执行。

# dispatcher 实际上返回的是一个字符串列表，每个字符串是下一个要调用的节点名称。
# 例如，如果 dispatcher 返回 ["sales_expert", "tech_support_expert"]
# 那么 sales_expert 和 tech_support_expert 会被并行调用。
# 而从这些专家节点到 aggregator 的边，确保了 aggregator 是在它们都完成后才被调用的。

# 假设dispatcher返回的列表中的每个元素都会作为下一个要执行的节点，
# 并且这些节点会并行执行。
# 然后，我们需要将这些并行执行的节点连接到同一个汇聚节点。
# 这样，汇聚节点就会在所有这些并行任务完成后被调用。

# 明确地，从每个专家节点到汇聚节点添加边
workflow.add_edge("sales_expert", "aggregator")
workflow.add_edge("tech_support_expert", "aggregator")
workflow.add_edge("billing_expert", "aggregator")

# 设置图的最终结束点
workflow.add_edge("aggregator", END)

# 编译图
app = workflow.compile()

关于 Fan-out 边的处理：
在 LangGraph 中，当一个节点（例如 dispatcher_node）返回一个字符串列表时，LangGraph 会自动将列表中的每个字符串视为一个要并行执行的下一个节点。这意味着你不需要为 dispatcher 到 sales_expert、tech_support_expert 等显式添加条件边。LangGraph 会隐式地处理这种并行调度。

而 Fan-in 的实现则体现在：所有并行执行的专家节点（sales_expert, tech_support_expert, billing_expert）都将它们的输出传递给同一个 aggregator 节点。LangGraph 的调度器会确保 aggregator 节点只在所有指向它的前驱节点（即所有被调用的专家节点）都执行完毕后才会被调用。这是 LangGraph 自动处理结果汇聚的关键机制。

3.6 运行图并观察并行执行

现在我们来运行这个图，并观察其执行流程和时间消耗。

# 运行图
async def run_triage_example(query_text: str):
    print(f"n--- 处理查询: '{query_text}' ---")
    start_time = time.perf_counter()

    result = await app.ainvoke({"query": query_text})

    end_time = time.perf_counter()
    duration = end_time - start_time

    print("n--- 最终结果 ---")
    print(result["final_analysis"])
    print(f"n总耗时: {duration:.2f} 秒")

    return result

# 示例查询
async def main():
    # 场景1: 触发所有专家 (或大多数)
    await run_triage_example("我有一个关于产品升级和账单支付的问题，需要技术支持。")
    # 期望：dispatcher 调度 sales, tech_support, billing。最长耗时 3 秒 (tech_support)

    print("n" + "="*80 + "n")

    # 场景2: 只触发部分专家
    await run_triage_example("我正在考虑购买你们的新产品，请问最近有什么优惠吗？")
    # 期望：dispatcher 调度 sales。最长耗时 2 秒 (sales)

    print("n" + "="*80 + "n")

    # 场景3: 触发所有专家 (因为没有特定匹配)
    await run_triage_example("请问如何提高我的账户安全性？")
    # 期望：dispatcher 调度 sales, tech_support, billing。最长耗时 3 秒 (tech_support)

if __name__ == "__main__":
    asyncio.run(main())

预期输出分析：

当我们运行第一个查询 "我有一个关于产品升级和账单支付的问题，需要技术支持。" 时：

• dispatcher_node 会识别出关键词 "产品升级" (sales, tech_support), "账单支付" (billing), "技术支持" (tech_support)。
• 它会调度 sales_expert (2秒), tech_support_expert (3秒), billing_expert (1秒) 三个节点。
• 由于它们是并行执行的，整个并行阶段的耗时将由最长的那个任务决定，即 tech_support_expert 的 3 秒。
• aggregator_node 会在所有专家完成后的第 3 秒左右被调用。
• 因此，总耗时应该略高于 3 秒（加上节点间的调度和打印时间）。

当我们运行第二个查询 "我正在考虑购买你们的新产品，请问最近有什么优惠吗？" 时：

• dispatcher_node 只会调度 sales_expert (2秒)。
• 总耗时将略高于 2 秒。

通过观察时间戳和总耗时，我们可以清晰地看到并行执行带来的效率提升。

3.7 流程可视化（可选，但有助于理解）

虽然我们不能插入图片，但可以描述一下这个图的逻辑结构：

节点名称	类型	职责	输入（State）	输出（State）	下一步（返回）
dispatcher	决策	分析 query，决定激活哪些专家。	query	active_experts	List[str] (专家节点名称列表)
sales_expert	专家	模拟销售咨询，生成销售回复。	query	expert_results["sales_expert"]	"aggregator"
tech_support_expert	专家	模拟技术支持，生成技术支持回复。	query	expert_results["tech_support_expert"]	"aggregator"
billing_expert	专家	模拟账单查询，生成账单回复。	query	expert_results["billing_expert"]	"aggregator"
aggregator	汇聚	收集所有 expert_results，生成 final_analysis。	active_experts, expert_results	final_analysis	END

执行流：

1. START -> dispatcher
2. dispatcher 返回 ["sales_expert", "tech_support_expert", "billing_expert"] (或子集)。
3. LangGraph 并行执行 sales_expert, tech_support_expert, billing_expert。
4. 每个专家节点更新 state["expert_results"]。
5. 一旦所有被调度的专家节点都完成，LangGraph 自动将控制权交给 aggregator。
6. aggregator 处理 state["expert_results"]，生成 final_analysis。
7. aggregator -> END。

四、高级考量与最佳实践

4.1 动态 Fan-out

我们当前的分发器已经实现了基于关键词的动态 Fan-out。在更复杂的场景中，dispatcher_node 可以：

• 使用一个单独的 LLM 调用进行意图识别。
• 查询外部配置服务来获取可用的专家列表及其触发条件。
• 根据用户历史、优先级等因素动态调整调度的专家。

4.2 错误处理与容错

在并行执行中，一个专家的失败不应该导致整个流程崩溃。

• 在专家节点内部处理错误： 每个专家节点都应该包含 try-except 块来捕获其内部的错误。
• 将错误信息存入状态： 如果一个专家失败，它可以将错误信息写入 state["expert_results"] 中（例如，expert_results["sales_expert"] = "Error: Sales API failed."），而不是抛出异常。
• 汇聚器处理部分失败： aggregator_node 应该能够识别这些错误信息，并在 final_analysis 中说明哪些专家未能提供有效结果，或者提供一个回退（fallback）方案。

示例：带有错误处理的专家节点

async def resilient_sales_expert_node(state: AgentState) -> AgentState:
    print(f"[{time.time():.2f}] 弹性销售专家开始处理...")
    query = state["query"]
    current_results = state["expert_results"]

    try:
        # 模拟LLM调用或其他耗时操作，可能失败
        await asyncio.sleep(2) 
        if "模拟销售错误" in query: # 模拟特定查询导致错误
            raise ValueError("模拟销售系统连接失败")

        sales_advice = f"销售专家建议：针对 '{query}'，我们有最新的优惠套餐和解决方案..."
        current_results["sales_expert"] = sales_advice

    except Exception as e:
        error_msg = f"销售专家处理失败：{str(e)}"
        print(f"[{time.time():.2f}] {error_msg}")
        current_results["sales_expert"] = error_msg # 存储错误信息
        state["error"] = error_msg # 也可将错误汇总到顶层

    print(f"[{time.time():.2f}] 弹性销售专家处理完成。")
    return {**state, "expert_results": current_results}

# 在workflow中用resilient_sales_expert_node替换sales_expert_node
# workflow.add_node("sales_expert", resilient_sales_expert_node)

汇聚器也需要相应地调整以检查 expert_results 中的错误字符串。

4.3 性能优化与资源管理

• 并发限制： 如果并行调用的外部服务（如 LLM API）有速率限制，你可能需要限制同时进行的请求数量。这可以通过在每个专家节点内部使用 asyncio.Semaphore 或在更顶层管理并发池来实现。
• 异步 I/O： 确保所有耗时的 I/O 操作都使用 asyncio 兼容的库（例如，httpx 而不是 requests）和 await 关键字，以充分利用并发。
• 缓存： 对重复的查询或专家回答进行缓存，避免不必要的重复计算或 API 调用。

4.4 更复杂的汇聚策略

我们的 aggregator_node 只是简单地将所有结果拼接起来。在实际应用中，你可能需要更复杂的策略：

• LLM 驱动的总结： 使用另一个 LLM 作为一个“超级汇聚专家”，将所有专家的原始输出作为输入，生成一个更连贯、更智能的最终总结。
• 投票机制： 如果专家提供的是分类或决策，可以采用投票机制来确定最终结果。
• 加权平均： 如果专家提供的是数值，可以根据专家的可信度或专业性进行加权平均。
• 冲突解决： 如果不同专家给出相互矛盾的建议，汇聚器可能需要识别并尝试解决这些冲突，甚至回溯到用户进行澄清。

4.5 状态管理深挖

随着图的复杂性增加，AgentState 可能会变得非常庞大。

• 嵌套结构： 使用嵌套的 TypedDict 或 Pydantic 模型来组织状态，使其更具可读性和可维护性。
• 清理中间状态： 某些中间结果在汇聚后可能不再需要。考虑在 aggregator_node 或后续节点中清理这些状态，以减少状态的内存占用。
• 状态持久化： 对于长运行或需要断点续传的流程，可能需要将 LangGraph 的状态持久化到数据库中。

五、总结与展望

通过 Fan-out/Fan-in 模式在 LangGraph 中实现并行执行，我们能够显著提升复杂 AI 工作流的效率和响应速度。这种模式通过将任务分解、并行处理和智能汇聚结合，为构建模块化、可扩展的多专家系统提供了坚实的基础。

我们探讨了如何设计共享状态、实现异步专家节点、通过动态返回节点列表实现 Fan-out，以及通过统一的汇聚节点实现 Fan-in。同时，我们也讨论了错误处理、性能优化和更高级的汇聚策略等关键考量。掌握这些技术，你将能够构建更加健壮、高效和智能的 LangGraph 应用，充分发挥其图式编程的强大潜力。