本教程将带你了解如何使用 langgraph 库构建和执行一个状态图, 并在调用或流式传输图时设置图的递归限制。。我们将通过一个具体的例子来解释程序的运行逻辑。

1. 导入必要的库

首先，我们需要导入一些必要的库：

import operator
from typing import Annotated, Any

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END

• operator：用于提供操作符函数，如 operator.add。
• typing 和 typing_extensions：用于类型注解。
• langgraph.graph：提供状态图相关的类和方法。

2. 定义状态类型

我们定义一个 State 类型，它是一个 TypedDict，包含一个 aggregate 键，其值是一个列表，并且使用 operator.add 作为归约函数，使其具有追加功能。

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

3. 定义节点函数

接下来，我们定义四个节点函数，每个函数都接受一个状态并返回一个新的状态。

def node_a(state):
    return {"aggregate": ["I'm A"]}

def node_b(state):
    return {"aggregate": ["I'm B"]}

def node_c(state):
    return {"aggregate": ["I'm C"]}

def node_d(state):
    print("I'm D")
    return {"aggregate": ["I'm A"]}

• node_a、node_b 和 node_c 分别向状态中的 aggregate 列表追加字符串。
• node_d 打印 “I’m D” 并向 aggregate 列表追加 “I’m A”。

4. 构建状态图

我们使用 StateGraph 类来构建状态图。

builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_edge(START, "a")
builder.add_node("b", node_b)
builder.add_node("c", node_c)
builder.add_node("d", node_d)
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()

• 创建一个 StateGraph 实例。
• 添加节点 a、b、c 和 d。
• 添加边来定义节点之间的执行顺序：
  • 从 START 到 a。
  • 从 a 到 b 和 c。
  • 从 b 和 c 到 d。
  • 从 d 到 END。
• 编译图以生成可执行的图对象。

5. 执行状态图

我们尝试执行状态图，并设置递归限制。

5.1 设置递归限制为 3

from langgraph.errors import GraphRecursionError

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

输出：

Recursion Error

由于递归限制为 3，图在执行过程中达到递归限制，抛出 GraphRecursionError。

5.2 设置递归限制为 4

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

这次没有输出，因为递归限制为 4，图能够成功执行。

总结

通过本教程，我们学习了如何使用 langgraph 库构建和执行一个状态图。我们定义了状态类型、节点函数，构建了状态图，并尝试在不同递归限制下执行图。希望这个教程对你理解状态图的工作原理有所帮助。
参考链接：https://langchain-ai.github.io/langgraph/how-tos/recursion-limit/
如果有任何问题，欢迎在评论区提问。