如何创建和控制循环¶

先决条件

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

在创建包含循环的图时，我们需要一种机制来终止执行。这通常通过添加一个条件边，在达到某个终止条件时路由到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构建的LLM应用程序——更多关于如何开始的信息，请参阅这里。

定义图¶

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

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

API Reference: StateGraph | START | END

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

API Reference: StateGraph | START | END

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