如何创建和控制循环¶

先决条件

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

在创建包含循环的图形时，我们需要一种终止执行的机制。这通常通过添加一个条件边，当达到某个终止条件时，该边会路由到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")

设置¶

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

pip install -U langgraph

为LangGraph开发设置LangSmith

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

定义图¶

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

API Reference: StateGraph | START | END

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

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

带分支的循环¶

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

API Reference: StateGraph | START | END

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()))

该图看起来比较复杂，但实际上可以被看作是一个由超级步骤组成的循环：

节点A
节点B
节点C和节点D
节点A
...

我们有一个包含四个超级步骤的循环，在这个循环中，节点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']

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

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