跳到内容

如何创建一系列步骤

先决条件

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

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

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

摘要

要添加一系列节点,我们使用 的 .add_node.add_edge 方法:API 参考: 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_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}

注意

请注意,在向 状态 发出 更新 时,每个 节点 只需指定它希望 更新 的 键 的值。

默认情况下,这将 覆盖 相应 键 的值。您还可以使用 归约器 (reducers) 来控制 更新 的处理方式—— 例如,您可以将后续 更新 附加到 键 中,而不是 覆盖。更多详情请参见 本指南

定义图

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

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

API 参考: 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__
  • 我们必须指定 图 的 入口点。为此,我们添加一条指向 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}

请注意

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

内置简写

先决条件

.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}

评论