如何从节点更新图状态¶
本指南演示了如何在 LangGraph 中定义和更新状态。我们将展示：
在我们的示例中，我们将使用消息。这代表了许多大语言模型（LLM）应用程序中一种通用的状态表达方式。有关更多详细信息，请参阅我们的概念页面。
准备工作¶
首先，让我们安装 langgraph：
设置 LangSmith 以实现更好的调试
注册 LangSmith 以快速发现问题并提升你的 LangGraph 项目的性能。LangSmith 允许你使用跟踪数据来调试、测试和监控使用 LangGraph 构建的大语言模型应用程序 —— 请在 文档 中了解更多关于如何开始使用的信息。
示例图¶
定义状态¶
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
此节点仅将一条消息追加到我们的消息列表中，并填充一个额外的字段。
重要
节点应直接返回对状态的更新，而不是修改状态。
接下来，让我们定义一个包含此节点的简单图。我们使用 状态图 来定义一个对该状态进行操作的图。然后，我们使用 添加节点 来填充我们的图。
from langgraph.graph import StateGraph
graph_builder = StateGraph(State)
graph_builder.add_node(node)
graph_builder.set_entry_point("node")
graph = graph_builder.compile()
LangGraph 提供了用于可视化图的内置实用工具。让我们来查看一下我们的图。有关可视化的详细信息，请参阅本指南。
在这种情况下，我们的图仅执行一个节点。
使用图¶
让我们进行一个简单的调用：
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}
请注意：
- 我们通过更新状态的单个键来启动调用。
- 我们在调用结果中接收整个状态。
为了方便起见，我们经常通过美化打印来检查消息对象的内容：
================================[1m Human Message [0m=================================
Hi
==================================[1m Ai Message [0m==================================
Hello!
使用归约器处理状态更新¶
状态中的每个键都可以有其独立的归约器函数，该函数控制如何应用来自节点的更新。如果没有显式指定归约器函数，则假定对该键的所有更新都应覆盖它。
对于 TypedDict 状态模式，我们可以通过用归约器函数对状态的相应字段进行注解来定义归约器。
在前面的示例中，我们的节点通过向状态中的 "messages" 键追加一条消息来更新该键。下面，我们为这个键添加一个归约器，以便自动追加更新：
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()
================================[1m Human Message [0m=================================
Hi
==================================[1m Ai Message [0m==================================
Hello!
消息状态¶
实际上，更新消息列表还有其他需要考虑的因素：
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()
input_message = {"role": "user", "content": "Hi"}
result = graph.invoke({"messages": [input_message]})
for message in result["messages"]:
message.pretty_print()
================================[1m Human Message [0m=================================
Hi
==================================[1m Ai Message [0m==================================
Hello!
MessagesState，这样我们可以实现：
后续步骤¶
- 继续阅读Graph API 基础指南。
- 查看有关状态管理的更多详细信息。