LangGraph 状态管理全解析：从定义到高级应用

创想小编 · 发表于昨天 08:16

作者：佑瞻
在构建复杂的语言模型应用时，状态管理往往是容易被忽视却又至关重要的一环。当我们使用 LangGraph 搭建智能交互系统时，如何高效定义、更新和管理状态直接影响着系统的灵活性和可维护性。今天我们就来深入探讨 LangGraph 中状态管理的核心机制，从基础定义到高级特性，带大家全面掌握这一关键技术。
一、状态的基础定义与更新机制

1.1 状态结构的定义方式

在 LangGraph 中，状态可以通过多种方式定义，包括 TypedDict、Pydantic 模型或数据类。我们先以 TypedDict 为例，看看如何构建一个基础的状态结构：
python
运行

from langchain_core.messages import AnyMessage
from typing_extensions import TypedDict
class State(TypedDict):
messages: list[AnyMessage]
extra_field: int

复制代码

这个状态结构包含两个字段：一个消息列表和一个整数类型的额外字段。默认情况下，图的输入和输出模式由状态决定，这为大多数 LLM 应用提供了灵活的表述方式。
1.2 状态更新的核心原则

当我们需要更新状态时，节点函数应返回新的状态更新，而不是直接修改原始状态。来看一个简单的例子：
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}

复制代码

这个节点函数向消息列表中添加了一条新消息，并设置了额外字段的值。接下来我们需要将这个节点添加到状态图中：
python
运行

from langgraph.graph import StateGraph
builder = StateGraph(State)
builder.add_node(node)
builder.set_entry_point("node")
graph = builder.compile()

复制代码

调用这个图时，我们可以传入初始状态并获取更新后的完整状态：
python
运行

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

复制代码

输出结果会包含我们添加的两条消息和设置的额外字段值。这里需要注意的是，我们只需更新状态的部分键，调用结果会返回完整的状态。
二、使用 Reducers 优化状态更新

2.1 Reducers 的基本概念

每个状态键都可以有独立的 reducer 函数，用于控制节点更新的应用方式。如果没有显式指定 reducer，默认会覆盖该键的值。对于 TypedDict 状态，我们可以通过注解为字段指定 reducer 函数：
python
运行

from typing_extensions import Annotated
from operator import add # 也可以自定义add函数
class State(TypedDict):
messages: Annotated[list[AnyMessage], add]
extra_field: int

复制代码

这样一来，节点函数可以简化为：
python
运行

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

复制代码

2.2 消息状态的特殊处理

在实际应用中，消息列表的更新有更多需要考虑的地方，比如更新现有消息或接受简化的消息格式。LangGraph 提供了内置的 add_messages reducer 来处理这些情况：
python
运行

from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
extra_field: int

复制代码

甚至可以直接使用预构建的 MessagesState：
python
运行

from langgraph.graph import MessagesState
class State(MessagesState):
extra_field: int

复制代码

三、高级状态管理特性

3.1 定义独立的输入输出模式

默认情况下，StateGraph 使用单一模式，但我们也可以定义独立的输入和输出模式：
python
运行

from langgraph.graph import StateGraph, START, END
# 输入模式
class InputState(TypedDict):
question: str
# 输出模式
class OutputState(TypedDict):
answer: str
# 整体模式
class OverallState(InputState, OutputState):
pass
def answer_node(state: InputState):
return {"answer": "bye", "question": state["question"]}
builder = StateGraph(OverallState, input=InputState, output=OutputState)
builder.add_node(answer_node)
builder.add_edge(START, "answer_node")
builder.add_edge("answer_node", END)
graph = builder.compile()
print(graph.invoke({"question": "hi"})) # 输出: {'answer': 'bye'}

复制代码

这里输出结果只包含定义的输出模式字段，实现了数据的有效过滤。
3.2 节点间传递私有状态

有时我们需要节点间交换中间数据，这些数据不需要作为图的主要模式部分：
python
运行

from langgraph.graph import StateGraph, START, END
# 公共状态
class OverallState(TypedDict):
a: str
# 节点1的输出（包含私有数据）
class Node1Output(TypedDict):
private_data: str
def node_1(state: OverallState) -> Node1Output:
return {"private_data": "set by node_1"}
# 节点2的输入（只需要私有数据）
class Node2Input(TypedDict):
private_data: str
def node_2(state: Node2Input) -> OverallState:
return {"a": "set by node_2"}
def node_3(state: OverallState) -> OverallState:
return {"a": "set by node_3"}
builder = StateGraph(OverallState).add_sequence([node_1, node_2, node_3])
builder.add_edge(START, "node_1")
graph = builder.compile()
response = graph.invoke({"a": "set at start"})
print(f"Output: {response}") # 输出: {'a': 'set by node_3'}

复制代码

这个例子中，私有数据只在节点 1 和节点 2 之间传递，节点 3 只能访问公共状态。
3.3 使用 Pydantic 模型进行状态验证

Pydantic 模型可以为状态添加运行时验证：
python
运行

from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END
class OverallState(BaseModel):
a: str
def node(state: OverallState):
return {"a": "goodbye"}
builder = StateGraph(OverallState)
builder.add_node(node)
builder.add_edge(START, "node")
builder.add_edge("node", END)
graph = builder.compile()
graph.invoke({"a": "hello"}) # 正常运行
try:
graph.invoke({"a": 123}) # 会抛出异常
except Exception as e:
print("异常信息:", e)

复制代码

这里需要注意，目前图的输出不会是 Pydantic 模型实例，运行时验证只针对节点输入，且错误跟踪不显示具体节点。
3.4 添加运行时配置

我们可以在调用图时配置参数，而不是将参数混入状态中：
python
运行

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END
# 配置模式
class ConfigSchema(TypedDict):
my_runtime_value: str
# 状态模式
class State(TypedDict):
my_state_value: str
def node(state: State, config: RunnableConfig):
value = config["configurable"]["my_runtime_value"]
return {"my_state_value": 1 if value == "a" else 2}
builder = StateGraph(State, config_schema=ConfigSchema)
builder.add_node(node)
builder.add_edge(START, "node")
builder.add_edge("node", END)
graph = builder.compile()
print(graph.invoke({}, {"configurable": {"my_runtime_value": "a"}})) # 输出: {'my_state_value': 1}
print(graph.invoke({}, {"configurable": {"my_runtime_value": "b"}})) # 输出: {'my_state_value': 2}

复制代码

3.5 节点重试策略

对于调用 API、数据库或 LLM 等可能失败的操作，我们可以为节点添加重试策略：
python
运行

from langgraph.pregel import RetryPolicy
builder.add_node(
"node_name",
node_function,
retry=RetryPolicy(), # 使用默认重试策略
)

复制代码

默认情况下，会重试除特定异常外的所有错误，对于 HTTP 请求库的异常，只重试 5xx 状态码。
3.6 节点缓存机制

为避免重复执行昂贵操作，我们可以为节点添加缓存：
python
运行

from langgraph.types import CachePolicy
from langgraph.cache.memory import InMemoryCache
builder.add_node(
"node_name",
node_function,
cache_policy=CachePolicy(ttl=120), # 缓存有效期120秒
)
graph = builder.compile(cache=InMemoryCache()) # 启用内存缓存

复制代码

也可以使用 SqliteCache 实现持久化缓存。
总结与实践建议

通过本文，我们全面了解了 LangGraph 中状态管理的各个方面，从基础的状态定义和更新，到使用 reducers 优化更新过程，再到高级的输入输出模式定义、私有状态传递、运行时配置、重试策略和节点缓存等特性。这些功能为构建复杂的语言模型应用提供了强大的支持。
如果本文对你有帮助，别忘了点赞收藏，关注我，一起探索更高效的开发方式～

原文地址：https://blog.csdn.net/The_Thieves/article/details/148798756

25位大厂高管转战生成式AI创业！吸金猛，不

LangGraph 状态管理全解析：从定义到高级应用

LangGraph(一)——构建聊天机器人

关于我们

产品与服务

解决方案

产品与服务