跳到内容

如何从节点更新图状态

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

  1. 如何使用状态来定义图的schema
  2. 如何使用reducers来控制状态更新的处理方式。

在我们的示例中，我们将使用messages。这代表了许多 LLM 应用程序中通用的状态表示形式。有关更多详细信息，请参阅我们的概念页面。

设置

首先，让我们安装 langgraph

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

设置 LangSmith 以获得更好的调试体验

注册 LangSmith 以快速发现问题并提高您的 LangGraph 项目的性能。LangSmith 使您可以使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序——阅读文档，了解如何开始使用。

示例图

定义状态

LangGraph 中的状态可以是 TypedDict、Pydantic 模型或 dataclass。下面我们将使用 TypedDict。有关使用 Pydantic 的详细信息，请参阅本指南。

默认情况下，图将具有相同的输入和输出 schema，并且状态决定了该 schema。有关如何定义不同的输入和输出 schema，请参阅本指南。

让我们考虑一个简单的示例

from langchain_core.messages import AnyMessage
from typing_extensions import TypedDict


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

API 参考：AnyMessage

此状态跟踪 message 对象的列表，以及一个额外的整数字段。

定义图结构

让我们构建一个包含单个节点的示例图。我们的节点只是一个 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 参考：AIMessage

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

重要提示

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

接下来，让我们定义一个包含此节点的简单图。我们使用 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 参考：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 参考：HumanMessage

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

请注意

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

为了方便起见，我们经常通过漂亮打印来检查 message 对象的内容

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

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

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

Hello!

使用 reducers 处理状态更新

状态中的每个键都可以有其自己的独立 reducer 函数，该函数控制如何应用来自节点的更新。如果未显式指定 reducer 函数，则假定对该键的所有更新都应覆盖它。

对于 TypedDict 状态 schema，我们可以通过使用 reducer 函数注释状态的相应字段来定义 reducers。

在前面的示例中，我们的节点通过将消息附加到状态中的 "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 参考：START

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

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

Hello!

MessagesState

在实践中，更新消息列表还有其他注意事项

  • 我们可能希望更新状态中的现有消息。
  • 我们可能想要接受消息格式的简写形式，例如 OpenAI 格式。

LangGraph 包含一个内置的 reducer 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 参考：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

下一步

评论