如何从节点更新图状态¶
本指南演示如何在LangGraph中定义和更新状态。我们将展示：
我们将在示例中使用消息。这代表了适用于许多LLM应用的状态的灵活表述形式。更多详情请参阅我们的概念页面。
设置¶
首先，让我们安装langgraph:
使用LangSmith进行更好的调试
注册 LangSmith 可以快速发现并解决您的 LangGraph 项目中的问题，并提高其性能。LangSmith 允许您使用跟踪数据来调试、测试和监控使用 LangGraph 构建的大型语言模型应用——更多关于如何开始的信息，请参阅文档。
示例图¶
定义状态¶
在LangGraph中，状态可以是TypedDict、Pydantic模型或数据类。下面我们将使用TypedDict。有关如何使用Pydantic的详细信息，请参阅此指南。
默认情况下，图具有相同的输入和输出模式，并且状态确定该模式。有关如何定义不同的输入和输出模式的详细信息，请参阅此指南。
让我们考虑一个简单的例子：
API Reference: AnyMessage
from langchain_core.messages import AnyMessage
from typing_extensions import TypedDict
class State(TypedDict):
messages: list[AnyMessage]
extra_field: int
此状态跟踪一组消息对象，以及一个额外的整数字段。
定义图结构¶
让我们构建一个只有一个节点的例子图。我们的节点只是一个Python函数，它读取图的状态并对其进行更新。该函数的第一个参数始终是状态：
API Reference: AIMessage
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}
此节点只是将一条消息添加到我们的消息列表中，并填充一个额外的字段。
Important
节点应该直接返回状态更新，而不是修改状态。
接下来，我们定义一个包含此节点的简单图。我们使用StateGraph来定义一个操作此状态的图。然后，我们使用add_node来填充我们的图。
API Reference: StateGraph
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 提供了用于可视化您的图的内置工具。让我们来检查一下我们的图。有关可视化的详细信息，请参阅此指南。
在这种情况下，我们的图只是执行一个节点。
使用图表¶
让我们继续进行一个简单的调用：
API Reference: HumanMessage
from langchain_core.messages import HumanMessage
result = graph.invoke({"messages": [HumanMessage("Hi")]})
result
{'messages': [HumanMessage(content='Hi', additional_kwargs={}, response_metadata={}),
AIMessage(content='Hello!', additional_kwargs={}, response_metadata={})],
'extra_field': 10}
注意：
- 我们通过更新状态中的单个键来启动调用。
- 在调用结果中接收整个状态。
为了方便，我们经常通过美化打印来检查消息对象的内容：
================================ 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}
API Reference: START
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()
================================ Human Message =================================
Hi
================================== Ai Message ==================================
Hello!
消息状态¶
在实际操作中，更新消息列表时还需要考虑以下几点：
LangGraph 包含一个内置的 add_messages 减速器来处理这些情况：
API Reference: 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()
================================ Human Message =================================
Hi
================================== Ai Message ==================================
Hello!
MessagesState 以方便使用，因此我们可以拥有：