Skip to content

如何从节点更新图状态

本指南演示如何在LangGraph中定义和更新状态。我们将展示：

  1. 如何使用状态定义图的模式
  2. 如何使用减少器来控制状态更新的处理方式。

我们将使用消息作为示例。这代表了适用于许多LLM应用的状态的灵活表述方式。更多详情请参阅我们的概念页面。

环境搭建

首先，让我们安装langgraph:

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

使用LangSmith进行更好的调试

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

示例图

定义状态

在LangGraph中，状态可以是一个TypedDict、Pydantic模型或数据类。下面我们将使用TypedDict。有关如何使用Pydantic的详细信息，请参阅此指南。

默认情况下，图的输入和输出模式是相同的，状态决定了该模式。有关如何定义不同的输入和输出模式的详细信息，请参阅此指南。

让我们考虑一个简单的例子：

from langchain_core.messages import AnyMessage
from typing_extensions import TypedDict


class State(TypedDict):
    messages: list[AnyMessage]
    extra_field: int

API Reference: AnyMessage

该状态跟踪一系列消息对象，以及一个额外的整数字段。

定义图结构

让我们构建一个只有一个节点的示例图。我们的节点只是一个Python函数，它读取图的状态并对其进行更新。该函数的第一个参数始终是状态：

from langchain_core.messages import AIMessage


def node(state: State):
    messages = state["messages"]
    new_message = AIMessage("Hello!")

    return {"messages": messages + [new_message], "extra_field": 10}

API Reference: AIMessage

这个节点只是将一条消息追加到我们的消息列表中，并填充一个额外的字段。

Important

节点应该直接返回状态更新，而不是修改状态。

接下来，我们定义一个包含此节点的简单图。我们使用StateGraph来定义一个操作此状态的图。然后我们使用add_node来填充我们的图。

from langgraph.graph import StateGraph

graph_builder = StateGraph(State)
graph_builder.add_node(node)
graph_builder.set_entry_point("node")
graph = graph_builder.compile()

API Reference: StateGraph

LangGraph 提供了用于可视化您的图的内置工具。让我们来检查一下我们的图。有关可视化操作的详细信息，请参阅此指南。

from IPython.display import Image, display

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

在这种情况下，我们的图只是执行一个节点。

使用图表

让我们继续进行一个简单的调用：

from langchain_core.messages import HumanMessage

result = graph.invoke({"messages": [HumanMessage("Hi")]})
result

API Reference: HumanMessage

{'messages': [HumanMessage(content='Hi', additional_kwargs={}, response_metadata={}),
  AIMessage(content='Hello!', additional_kwargs={}, response_metadata={})],
 'extra_field': 10}

注意：

  • 我们通过更新状态中的单个键来启动调用。
  • 我们在调用结果中接收整个状态。

为了方便，我们经常通过美化打印来检查消息对象的内容：

for message in result["messages"]:
    message.pretty_print()

================================ Human Message =================================

Hi
================================== Ai Message ==================================

Hello!

使用reducer更新进程状态

状态中的每个键都可以有自己的独立reducer函数，该函数控制从节点收到的更新如何应用。如果没有明确指定reducer函数，则假定对键的所有更新都将覆盖原有的值。

对于TypedDict状态模式，我们可以通过在状态的相应字段上添加reducer函数来定义reducer。

在前面的例子中，我们的节点通过向状态中的"messages"键追加一条消息来更新状态。下面，我们将为此键添加一个reducer，使得更新会自动追加：

from typing_extensions import Annotated


def add(left, right):
    """Can also import `add` from the `operator` built-in."""
    return left + right


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add]
    extra_field: int

现在我们的节点可以简化了：

def node(state: State):
    new_message = AIMessage("Hello!")
    return {"messages": [new_message], "extra_field": 10}

from langgraph.graph import START


graph = StateGraph(State).add_node(node).add_edge(START, "node").compile()

result = graph.invoke({"messages": [HumanMessage("Hi")]})

for message in result["messages"]:
    message.pretty_print()

API Reference: START 

================================ Human Message =================================

Hi
================================== Ai Message ==================================

Hello!

消息状态

在实际操作中，更新消息列表时还需要考虑以下几点：

  • 我们可能希望更新状态中已存在的消息。
  • 我们可能希望接受消息格式的简写形式，例如消息格式，或者OpenAI格式。

LangGraph 包含一个内置的 add_messages 减少器，用于处理这些考虑因素：

from langgraph.graph.message import add_messages


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    extra_field: int


def node(state: State):
    new_message = AIMessage("Hello!")
    return {"messages": [new_message], "extra_field": 10}


graph = StateGraph(State).add_node(node).set_entry_point("node").compile()

API Reference: add_messages

input_message = {"role": "user", "content": "Hi"}

result = graph.invoke({"messages": [input_message]})

for message in result["messages"]:
    message.pretty_print()

================================ Human Message =================================

Hi
================================== Ai Message ==================================

Hello!
这是一个适用于涉及聊天模型的应用程序的状态表示。LangGraph 包括一个预构建的 MessagesState 以方便使用，因此我们可以拥有：

from langgraph.graph import MessagesState


class State(MessagesState):
    extra_field: int

下一步

Comments