如何为并行节点执行创建分支

先决条件

本指南假设您熟悉以下内容：

• 节点
• 边缘
• 减少器

并行执行节点对于加快整体图操作速度至关重要。LangGraph 提供了对节点并行执行的本机支持，这可以显著增强基于图的工作流的性能。这种并行化是通过扇出和扇入机制实现的，利用了标准边缘和有条件边缘。以下是一些示例，展示了如何创建适合您的分支数据流。

设置环境

首先，让我们安装所需的包

%%capture --no-stderr
%pip install -U langgraph

使用LangSmith为LangGraph开发设置环境

注册LangSmith，可以快速发现并解决您的LangGraph项目中的问题，提高项目性能。LangSmith允许您使用跟踪数据来调试、测试和监控使用LangGraph构建的LLM应用程序——更多关于如何开始的信息，请参阅这里。

如何并行运行图节点

在这个示例中，我们从Node A分发到B和C，然后汇聚到D。通过我们的状态，我们指定了reducer的加法操作。这将合并或累积状态中特定键的值，而不是简单地覆盖现有值。对于列表，这意味着将新列表与现有列表连接起来。有关使用reducer更新状态的更多详细信息，请参阅此指南。

import operator
from typing import Annotated, Any

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    # The operator.add reducer fn makes this append-only
    aggregate: Annotated[list, operator.add]


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


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


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


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


builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)
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()

API Reference: StateGraph | START | END

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

通过reducer，你可以看到每个节点中添加的值是被累积起来的。

graph.invoke({"aggregate": []}, {"configurable": {"thread_id": "foo"}})

Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "D" to ['A', 'B', 'C']

{'aggregate': ['A', 'B', 'C', 'D']}

Note

在上述示例中，节点 "b" 和 "c" 在同一个 超级步骤 中并发执行。由于它们处于同一个步骤中，因此节点 "d" 在节点 "b" 和 "c" 完成后执行。

重要的是，来自并行超级步骤的更新可能不会按一致的顺序进行。如果您需要来自并行超级步骤的更新以一致、预定的顺序出现，则应将输出写入状态中的单独字段，并与用于排序的值一起保存。

异常处理？

LangGraph 在“超级步骤”中执行节点，这意味着虽然并行分支可以并行执行，但整个超级步骤是事务性的。如果这些分支中的任何一个抛出异常，没有任何更新会被应用到状态中（整个超级步骤会出错）。

重要的是，当使用检查点器时，成功节点的结果会在超级步骤中保存，并在恢复时不会重复执行。

如果你有容易出错的代码（也许想处理不可靠的API调用），LangGraph 提供了两种方法来解决这个问题：

1. 你可以在节点中编写常规的Python代码来捕获和处理异常。
2. 你可以设置一个重试策略来指示图重试抛出某些类型异常的节点。只有失败的分支会被重试，因此你不必担心执行重复的工作。

这些方法结合起来，让你能够执行并行执行并完全控制异常处理。

并行节点的扇出和扇入（包含额外步骤）

上面的例子展示了当每条路径只有一个步骤时如何进行扇出和扇入。但如果一条路径包含多个步骤又该如何处理呢？让我们在“b”分支中添加一个节点 b_2：

def b_2(state: State):
    print(f'Adding "B_2" to {state["aggregate"]}')
    return {"aggregate": ["B_2"]}


builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(b_2)
builder.add_node(c)
builder.add_node(d)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "b_2")
builder.add_edge(["b_2", "c"], "d")
builder.add_edge("d", END)
graph = builder.compile()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

graph.invoke({"aggregate": []})

Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "B_2" to ['A', 'B', 'C']
Adding "D" to ['A', 'B', 'C', 'B_2']

{'aggregate': ['A', 'B', 'C', 'B_2', 'D']}

Note

在上面的例子中，节点 "b" 和 "c" 在同一个 超级步骤 中并发执行。那么下一步会发生什么？

我们使用 add_edge(["b_2", "c"], "d") 这里强制节点 "d" 只在节点 "b_2" 和 "c" 都执行完毕后才运行。如果我们添加两条独立的边， 节点 "d" 将会运行两次：一次是在节点 b2 完成后，另一次是在节点 c 完成后（无论这两个节点的完成顺序如何）。

条件分支

如果您的分发分支不是确定性的，您可以直接使用 add_conditional_edges。

import operator
from typing import Annotated, Sequence

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    aggregate: Annotated[list, operator.add]
    # Add a key to the state. We will set this key to determine
    # how we branch.
    which: str


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


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


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


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


def e(state: State):
    print(f'Adding "E" to {state["aggregate"]}')
    return {"aggregate": ["E"]}


builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)
builder.add_node(e)
builder.add_edge(START, "a")


def route_bc_or_cd(state: State) -> Sequence[str]:
    if state["which"] == "cd":
        return ["c", "d"]
    return ["b", "c"]


intermediates = ["b", "c", "d"]
builder.add_conditional_edges(
    "a",
    route_bc_or_cd,
    intermediates,
)
for node in intermediates:
    builder.add_edge(node, "e")

builder.add_edge("e", END)
graph = builder.compile()

API Reference: StateGraph | START | END

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))

graph.invoke({"aggregate": [], "which": "bc"})

Adding "A" to []
Adding "B" to ['A']
Adding "C" to ['A']
Adding "E" to ['A', 'B', 'C']

{'aggregate': ['A', 'B', 'C', 'E'], 'which': 'bc'}

graph.invoke({"aggregate": [], "which": "cd"})

Adding "A" to []
Adding "C" to ['A']
Adding "D" to ['A']
Adding "E" to ['A', 'C', 'D']

{'aggregate': ['A', 'C', 'D', 'E'], 'which': 'cd'}

下一步

• 继续阅读Graph API 基础指南。
• 学习如何创建map-reduce分支，在这些分支中，不同的状态可以分发到节点的多个实例中。

Comments