环企首页
LangGraph 中状态与多重状态详解
📦 一、状态（State）的基本概念
在 LangGraph 中，状态（State） 是一个在所有节点之间共享的、可变的数据结构。它会在图执行的每个步骤中依次传递、更新和扩展。
from typing import TypedDict, Annotated, List
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage
class MyState(TypedDict):
messages: Annotated[List[BaseMessage], add_messages] # 使用 reducer
user_name: str # 普通字段
step_count: int # 普通字段
状态的核心特点
| 特性 | 说明 |
|---|---|
| 共享性 | 所有节点都能读取和修改状态的任意字段 |
| 不可变性 | 节点返回的更新是增量更新，不是完全替换 |
| Reducer 函数 | 决定如何合并新旧值（如追加、累加、覆盖等） |
| 类型安全 | 通过 TypedDict 定义结构，IDE 有类型提示 |
🧩 二、Reducer 机制（归约函数）
Reducer 是 LangGraph 最核心的设计之一。当你认为“节点返回的更新应该如何合并到现有状态”时，就需要 Reducer。
1. 内置 Reducer：add_messages
from typing import Annotated
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages] # 消息会追加，不会覆盖
效果：
- 节点1 返回
{"messages": [msg1]} - 节点2 返回
{"messages": [msg2]} - 最终状态
messages = [msg1, msg2]（追加）
2. 无 Reducer（默认覆盖）
class State(TypedDict):
user_name: str # 没有 Annotated，默认覆盖
# 节点1 返回 {"user_name": "Alice"}
# 节点2 返回 {"user_name": "Bob"}
# 最终: user_name = "Bob"（覆盖）
3. 自定义 Reducer
from typing import Annotated
def sum_reducer(left: int, right: int) -> int:
"""累加 reducer"""
return left + right
class State(TypedDict):
total: Annotated[int, sum_reducer] # 每次更新都会累加
# 节点1 返回 {"total": 10} → total = 10
# 节点2 返回 {"total": 5} → total = 15
4. 更复杂的自定义 Reducer（合并列表）
def merge_lists(left: list, right: list) -> list:
"""合并列表，去重保留顺序"""
seen = set()
result = []
for item in left + right:
if item not in seen:
seen.add(item)
result.append(item)
return result
class State(TypedDict):
visited_nodes: Annotated[list, merge_lists]
🔄 三、节点的状态更新模式
节点必须返回一个字典，代表要对状态进行的增量更新。
模式1：更新单个字段
def node_a(state: State) -> dict:
return {"user_name": "Alice"} # 只更新 user_name
模式2：更新多个字段
def node_b(state: State) -> dict:
return {
"user_name": "Bob",
"step_count": state["step_count"] + 1
}
模式3：更新时依赖当前状态
def node_c(state: State) -> dict:
messages = state["messages"]
last_message = messages[-1] if messages else None
response = process(last_message)
return {"messages": [response]} # 追加到 messages 列表
模式4：不更新任何状态（但可以执行副作用）
def node_d(state: State) -> dict:
print(f"Current user: {state['user_name']}")
return {} # 返回空字典，不修改状态
🎯 四、多层级状态（嵌套状态）
LangGraph 支持复杂的嵌套状态结构，适合大型应用。
1. 简单嵌套
class UserState(TypedDict):
name: str
age: int
class AppState(TypedDict):
user: UserState # 嵌套
messages: Annotated[list, add_messages]
def update_user(state: AppState) -> dict:
return {"user": {"name": "Alice", "age": 25}} # 整体替换 user
def increment_age(state: AppState) -> dict:
new_age = state["user"]["age"] + 1
return {"user": {"age": new_age}} # 只更新 age 字段（合并）
2. 深度嵌套 + Reducer
from typing import Annotated, TypedDict
class Metrics(TypedDict):
tokens: Annotated[int, sum_reducer] # 累加 token
steps: Annotated[int, sum_reducer] # 累加 step
class NodeMetrics(TypedDict):
node_name: str
metrics: Metrics
class State(TypedDict):
overall_metrics: Annotated[Metrics, lambda l, r: Metrics(
tokens=l["tokens"] + r["tokens"],
steps=l["steps"] + r["steps"]
)]
🧠 五、子图状态（Subgraph State）
当图中包含子图时，父状态和子状态可以映射转换。
子图定义
class SubgraphState(TypedDict):
query: str
result: str
def subgraph_node(state: SubgraphState) -> dict:
return {"result": f"Processed: {state['query']}"}
父图中调用子图
from langgraph.graph import StateGraph
# 父状态定义
class ParentState(TypedDict):
user_input: str
final_output: str
# 构建子图
subgraph = StateGraph(SubgraphState)
subgraph.add_node("process", subgraph_node)
subgraph.set_entry_point("process")
subgraph.set_finish_point("process")
compiled_subgraph = subgraph.compile()
# 在父图中使用子图（需要做状态映射）
def call_subgraph(state: ParentState) -> dict:
# 将父状态映射为子图输入
sub_input = {"query": state["user_input"]}
sub_result = compiled_subgraph.invoke(sub_input)
# 将子图输出映射回父状态
return {"final_output": sub_result["result"]}
📊 六、状态 Schema 的三种定义方式
| 方式 | 代码示例 | 适用场景 |
|---|---|---|
| TypedDict | class State(TypedDict): |
最常见，类型提示好 |
| Dataclass | @dataclass class State: |
需要默认值、方法时 |
| Pydantic BaseModel | class State(BaseModel): |
需要数据验证时 |
# 方式1: TypedDict
from typing import TypedDict
class State(TypedDict):
messages: list
# 方式2: Dataclass
from dataclasses import dataclass, field
@dataclass
class State:
messages: list = field(default_factory=list)
# 方式3: Pydantic
from pydantic import BaseModel
class State(BaseModel):
messages: list = []
🎬 七、完整示例：带多重状态的 Agent
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, add_messages
from langchain_core.messages import HumanMessage, AIMessage
# 定义状态（包含多个字段）
class AgentState(TypedDict):
messages: Annotated[list, add_messages] # 对话历史
iteration: int # 迭代次数
intermediate_results: list # 中间结果
final_answer: str # 最终答案
# 处理节点
def assistant(state: AgentState) -> dict:
print(f"迭代 {state['iteration'] + 1}: 处理中...")
# 模拟处理逻辑
last_msg = state["messages"][-1].content
result = f"处理结果: {last_msg}"
# 返回更新
return {
"messages": [AIMessage(content=result)],
"iteration": state["iteration"] + 1,
"intermediate_results": [result]
}
def finalizer(state: AgentState) -> dict:
# 汇总所有中间结果
summary = " → ".join(state["intermediate_results"])
return {"final_answer": summary}
# 构建图
builder = StateGraph(AgentState)
builder.add_node("assistant", assistant)
builder.add_node("finalize", finalizer)
builder.set_entry_point("assistant")
builder.add_edge("assistant", "finalize")
builder.set_finish_point("finalize")
graph = builder.compile()
# 运行
result = graph.invoke({
"messages": [HumanMessage(content="计算 3+5")],
"iteration": 0,
"intermediate_results": [],
"final_answer": ""
})
print("最终状态:")
print(f"- 消息数: {len(result['messages'])}")
print(f"- 迭代次数: {result['iteration']}")
print(f"- 中间结果: {result['intermediate_results']}")
print(f"- 最终答案: {result['final_answer']}")
输出：
迭代 1: 处理中...
最终状态:
- 消息数: 2
- 迭代次数: 1
- 中间结果: ['处理结果: 计算 3+5']
- 最终答案: 处理结果: 计算 3+5
💎 总结
| 概念 | 核心要点 |
|---|---|
| 状态 | 节点间共享的 TypedDict，通过返回值增量更新 |
| Reducer | 决定如何合并新旧值（追加/覆盖/累加/自定义） |
add_messages |
最常用的 reducer，自动追加消息列表 |
| 多字段状态 | 可包含多个独立字段，不同字段可同时更新 |
| 嵌套状态 | 支持嵌套 TypedDict / List / Dict 等复杂结构 |
| 子图状态 | 子图有独立状态，通过映射与父状态传递数据 |
最佳实践：
- 对消息列表始终使用
Annotated[list, add_messages] - 计数器、累加器使用自定义 reducer
- 简单覆盖用普通字段
- 大状态拆分为多个字段，减少碰撞和冗余