如何创建和控制循环

前提条件

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

• 图
• 递归限制
• 节点

在创建带有循环的图时，我们需要一种终止执行的机制。最常见的做法是添加一条条件边，一旦达到某个终止条件，就将路径导向END节点。

你还可以在调用或流式传输图时设置图的递归限制。递归限制设置了图在抛出错误之前允许执行的超级步数量。更多关于递归限制概念的信息请点击此处查看。

让我们考虑一个带有循环的简单图，以便更好地理解这些机制是如何工作的。

Tip

若要返回状态的最后一个值而不是收到递归限制错误，请阅读此操作指南。

总结

创建循环时，你可以包含一条指定终止条件的条件边：

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)

def route(state: State) -> Literal["b", END]:
    if termination_condition(state):
        return END
    else:
        return "a"

builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "a")
graph = builder.compile()

要控制递归限制，请在配置中指定 "recursion_limit"。这将抛出一个 GraphRecursionError，你可以捕获并处理该错误：

from langgraph.errors import GraphRecursionError

try:
    graph.invoke(inputs, {"recursion_limit": 3})
except GraphRecursionError:
    print("Recursion Error")

准备工作

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

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

为 LangGraph 开发设置 LangSmith

注册 LangSmith 以快速发现问题并提升你的 LangGraph 项目的性能。LangSmith 允许你使用跟踪数据来调试、测试和监控使用 LangGraph 构建的大语言模型应用程序 — 点击 此处 了解更多关于如何开始使用的信息。

定义图

让我们定义一个带有简单循环的图。请注意，我们使用条件边来实现终止条件。

import operator
from typing import Annotated, Literal

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'Node A sees {state["aggregate"]}')
    return {"aggregate": ["A"]}


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


# Define nodes
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)


# Define edges
def route(state: State) -> Literal["b", END]:
    if len(state["aggregate"]) < 7:
        return "b"
    else:
        return END


builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "a")
graph = builder.compile()

from IPython.display import Image, display

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

这种架构类似于一个ReAct 代理，其中节点 "a" 是一个工具调用模型，节点 "b" 代表工具。

在我们的 route 条件边中，我们指定当状态中的 "aggregate" 列表长度超过一个阈值时，流程应该结束。

调用该图时，我们会看到在达到终止条件之前，我们会在节点 "a" 和 "b" 之间交替执行，一旦达到终止条件，流程就会终止。

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

Node A sees []
Node B sees ['A']
Node A sees ['A', 'B']
Node B sees ['A', 'B', 'A']
Node A sees ['A', 'B', 'A', 'B']
Node B sees ['A', 'B', 'A', 'B', 'A']
Node A sees ['A', 'B', 'A', 'B', 'A', 'B']

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

施加递归限制

在某些应用场景中，我们无法保证能达到给定的终止条件。在这些情况下，我们可以设置图的递归限制。在经过给定数量的超步后，这将引发一个 GraphRecursionError 异常。然后我们可以捕获并处理这个异常：

from langgraph.errors import GraphRecursionError

try:
    graph.invoke({"aggregate": []}, {"recursion_limit": 4})
except GraphRecursionError:
    print("Recursion Error")

Node A sees []
Node B sees ['A']
Node A sees ['A', 'B']
Node B sees ['A', 'B', 'A']
Recursion Error

请注意，这次我们在第四步之后终止。

带分支的循环

为了更好地理解递归限制是如何工作的，让我们考虑一个更复杂的例子。下面我们实现一个循环，但每一步会扩展为两个节点：

import operator
from typing import Annotated, Literal

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    aggregate: Annotated[list, operator.add]


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


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


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


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


# Define nodes
builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)


# Define edges
def route(state: State) -> Literal["b", END]:
    if len(state["aggregate"]) < 7:
        return "b"
    else:
        return END


builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "c")
builder.add_edge("b", "d")
builder.add_edge(["c", "d"], "a")
graph = builder.compile()

from IPython.display import Image, display

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

这个图看起来很复杂，但可以将其概念化为 超级步 的循环：

1. 节点 A
2. 节点 B
3. 节点 C 和节点 D
4. 节点 A
5. ...

我们有一个包含四个超级步的循环，其中节点 C 和节点 D 是并发执行的。

像之前一样调用这个图，我们会发现，在达到终止条件之前，我们完成了两整“圈”：

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

Node A sees []
Node B sees ['A']
Node D sees ['A', 'B']
Node C sees ['A', 'B']
Node A sees ['A', 'B', 'C', 'D']
Node B sees ['A', 'B', 'C', 'D', 'A']
Node D sees ['A', 'B', 'C', 'D', 'A', 'B']
Node C sees ['A', 'B', 'C', 'D', 'A', 'B']
Node A sees ['A', 'B', 'C', 'D', 'A', 'B', 'C', 'D']

然而，如果我们将递归限制设置为 4，我们只会完成一圈，因为每一圈包含 4 个超级步：

from langgraph.errors import GraphRecursionError

try:
    result = graph.invoke({"aggregate": []}, {"recursion_limit": 4})
except GraphRecursionError:
    print("Recursion Error")

Node A sees []
Node B sees ['A']
Node C sees ['A', 'B']
Node D sees ['A', 'B']
Node A sees ['A', 'B', 'C', 'D']
Recursion Error

Comments