Skip to content

如何创建步骤序列

前提条件

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

本指南将演示如何构建一个简单的步骤序列。我们将展示:

  1. 如何构建一个顺序图
  2. 用于构建类似图的内置简写方法。

总结

要添加节点序列,我们使用 .add_node.add_edge 方法:

from langgraph.graph import START, StateGraph

graph_builder = StateGraph(State)

# 添加节点
graph_builder.add_node(step_1)
graph_builder.add_node(step_2)
graph_builder.add_node(step_3)

# 添加边
graph_builder.add_edge(START, "step_1")
graph_builder.add_edge("step_1", "step_2")
graph_builder.add_edge("step_2", "step_3")

我们还可以使用内置的简写方法 .add_sequence

graph_builder = StateGraph(State).add_sequence([step_1, step_2, step_3])
graph_builder.add_edge(START, "step_1")

为什么使用 LangGraph 将应用程序步骤拆分为序列? LangGraph 使你可以轻松地为应用程序添加底层持久化层。 这允许在节点执行之间对状态进行检查点保存,因此你的 LangGraph 节点可以控制: 它们还决定了执行步骤如何 流式传输,以及如何使用 LangGraph Studio 对应用程序进行可视化和调试。

安装设置

首先,让我们安装 langgraph:

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

设置 LangSmith 以实现更好的调试

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

构建图

让我们演示一个简单的使用示例。我们将创建一个包含三个步骤的序列:

  1. 在状态的一个键中填充一个值
  2. 更新同一个值
  3. 填充一个不同的值

定义状态

让我们首先定义我们的状态。这决定了图的架构,还可以指定如何应用更新。有关更多详细信息,请参阅本指南

在我们的示例中,我们只跟踪两个值:

from typing_extensions import TypedDict


class State(TypedDict):
    value_1: str
    value_2: int

定义节点

我们的节点只是读取图的状态并对其进行更新的 Python 函数。此函数的第一个参数始终是状态:

def step_1(state: State):
    return {"value_1": "a"}


def step_2(state: State):
    current_value_1 = state["value_1"]
    return {"value_1": f"{current_value_1} b"}


def step_3(state: State):
    return {"value_2": 10}

Note

请注意,在对状态进行更新时,每个节点只需指定其希望更新的键的值即可。

默认情况下,这将**覆盖**相应键的值。你还可以使用归约器来控制更新的处理方式 —— 例如,你可以将后续的更新追加到一个键上。有关更多详细信息,请参阅本指南

定义图

我们使用状态图来定义一个对该状态进行操作的图。

然后,我们将使用添加节点添加边来填充我们的图并定义其控制流。

from langgraph.graph import START, StateGraph

graph_builder = StateGraph(State)

# Add nodes
graph_builder.add_node(step_1)
graph_builder.add_node(step_2)
graph_builder.add_node(step_3)

# Add edges
graph_builder.add_edge(START, "step_1")
graph_builder.add_edge("step_1", "step_2")
graph_builder.add_edge("step_2", "step_3")

指定自定义名称

你可以使用 .add_node 为节点指定自定义名称:

graph_builder.add_node("my_node", step_1)

请注意:

  • .add_edge 接受节点的名称,对于函数而言,默认名称为 node.__name__
  • 我们必须指定图的入口点。为此,我们添加一条与 START 节点 相连的边。
  • 当没有更多节点需要执行时,图的执行将停止。

接下来,我们 编译 我们的图。这会对图的结构进行一些基本检查(例如,识别孤立节点)。如果我们要通过 检查点器 为应用程序添加持久化功能,也会在这里传入该检查点器。

graph = graph_builder.compile()

LangGraph 提供了用于可视化图形的内置实用工具。让我们检查一下我们的序列。有关可视化的详细信息,请参阅本指南

from IPython.display import Image, display

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

使用方法

让我们进行一个简单的调用:

graph.invoke({"value_1": "c"})
{'value_1': 'a b', 'value_2': 10}

请注意:

  • 我们通过为单个状态键提供一个值来启动调用。我们必须始终为至少一个键提供一个值。
  • 我们传入的值被第一个节点覆盖了。
  • 第二个节点更新了该值。
  • 第三个节点填充了一个不同的值。

内置简写形式

前提条件

.add_sequence 要求 langgraph>=0.2.46

为方便起见,LangGraph 包含了一个内置简写形式 .add_sequence

graph_builder = StateGraph(State).add_sequence([step_1, step_2, step_3])
graph_builder.add_edge(START, "step_1")

graph = graph_builder.compile()

graph.invoke({"value_1": "c"})
{'value_1': 'a b', 'value_2': 10}

Comments