Skip to content

如何创建步骤序列

前提条件

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

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

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

概述

要添加一系列节点,我们使用.add_node.add_edge方法: API Reference: START | StateGraph

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:

pip install -U langgraph

使用LangSmith进行更好的调试

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

构建图

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

  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

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

默认情况下,这将**覆盖**相应键的值。您也可以使用reducers来控制更新的处理方式——例如,您可以将连续的更新附加到一个键上而不是覆盖它。更多详情请参阅此指南

定义图

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

然后我们将使用add_nodeadd_edge来填充我们的图并定义其控制流。

API Reference: START | StateGraph

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__
  • 我们必须指定图的入口点。为此,我们添加一个与起始节点相连的边。
  • 当没有更多节点可以执行时,图将停止运行。

接下来,我们将图编译。这会对图的结构进行一些基本检查(例如,识别孤立节点)。如果我们通过检查点器为应用程序增加了持久性,它也会在这里传递。

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