📌 目标
1️⃣ State（状态）深入
State 的本质
State 是一个字典，在所有节点间共享和传递。它承载了整个应用的数据状态。
状态合并机制（进阶）
第一章已介绍基本合并规则，这里补充深层理解：
┌─────────────────────────────────────────────────────────┐
│ 状态合并的三个关键点 │
├─────────────────────────────────────────────────────────┤
│ │
│ 1. 部分更新：只返回需要修改的字段 │
│ 2. 保持不变：未返回的字段保持原值 │
│ 3. Reducer 优先：使用 Annotated 定义的字段按 Reducer 合并 │
│ │
└─────────────────────────────────────────────────────────┘
实战示例：游戏状态管理
class GameState(TypedDict):
player: str
score: int
level: int
inventory: list[str]
def level_up_node(state: GameState):
"""升级节点 - 演示部分更新"""
return {
"level": state["level"] + 1,
# player、score、inventory 自动保持不变
}
def add_item_node(state: GameState):
"""添加物品 - 演示列表更新"""
return {
"inventory": state["inventory"] + ["宝剑"]
}
2️⃣ Node（节点）深入
Node 函数规范（进阶）
💡 基础规范已在 01_快速入门 中介绍
高级模式：条件更新
def smart_node(state: MyState):
"""根据条件决定更新策略"""
if state["value"] > 100:
return {"status": "high", "processed": True}
elif state["value"] > 50:
return {"status": "medium", "processed": True}
else:
return {"status": "low"} # processed 保持不变
常见陷阱与解决方案
# ❌ 陷阱1：直接修改可变对象
def bad_list_update(state: MyState):
state["items"].append("new") # 直接修改原列表
return {}
# ✅ 正确：返回新列表
def good_list_update(state: MyState):
return {"items": state["items"] + ["new"]}
# ❌ 陷阱2：忘记处理 None 值
def risky_node(state: MyState):
return {"value": state["maybe_none"].upper()} # 可能报错
# ✅ 正确：安全处理
def safe_node(state: MyState):
value = state.get("maybe_none") or "default"
return {"value": value.upper()}
3️⃣ Edge（边）深入
边的类型速览
┌─────────────────────────────────────────────────────────┐
│ 边的类型 │
├─────────────────────────────────────────────────────────┤
│ 1. 普通边 (Normal Edge) - 固定路径 │
│ 2. 条件边 (Conditional Edge) - 动态路由 │
│ 3. 特殊边 - START/END │
└─────────────────────────────────────────────────────────┘
普通边示例
# 固定流程：A → B → C → END
workflow.add_edge(START, "A")
workflow.add_edge("A", "B")
workflow.add_edge("B", "C")
workflow.add_edge("C", END)
条件边预览
📖 条件边详解请参见 03_条件分支与循环
# 条件边：根据 State 动态选择路径
workflow.add_conditional_edges(
"decision_node",
route_function, # 返回下一个节点名称
{"option1": "node1", "option2": "node2"}
)
4️⃣ Reducer（状态归约器）
问题：默认合并的局限
默认情况下，节点返回的值会直接覆盖 State 中对应的字段：
class ScoreState(TypedDict):
score: int
def level1_node(state: ScoreState):
return {"score": 10} # score 变成 10
def level2_node(state: ScoreState):
return {"score": 20} # score 变成 20（覆盖了之前的10）
# 最终 score = 20，而不是 30
解决方案：使用 Reducer
Reducer 可以自定义状态的合并方式：
from typing import Annotated
import operator
class ScoreState(TypedDict):
player_name: str
# 使用 operator.add 作为 Reducer：累加而不是覆盖
score: Annotated[int, operator.add]
actions: Annotated[list[str], operator.add]
Reducer 工作原理
┌─────────────────────────────────────────────────────────┐
│ Reducer 工作原理 │
├─────────────────────────────────────────────────────────┤
│ │
│ 没有 Reducer（默认行为）： │
│ │
│ State: {"score": 10} │
│ 节点返回: {"score": 20} │
│ 结果: {"score": 20} ← 直接覆盖 │
│ │
│ ────────────────────────────────────────────── │
│ │
│ 有 Reducer (operator.add)： │
│ │
│ State: {"score": 10} │
│ 节点返回: {"score": 20} │
│ Reducer: 10 + 20 = 30 │
│ 结果: {"score": 30} ← 累加 │
│ │
└─────────────────────────────────────────────────────────┘
完整示例：游戏积分系统
对应代码：graph_node_edge2.py
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from typing import Annotated
import operator
class ScoreState(TypedDict):
player_name: str
score: Annotated[int, operator.add] # 累加积分
actions: Annotated[list[str], operator.add] # 累加动作列表
def level1_node(state: ScoreState):
"""第一关：获得10分"""
return {
"score": 10,
"actions": ["完成第一关"]
}
def level2_node(state: ScoreState):
"""第二关：获得20分"""
return {
"score": 20,
"actions": ["完成第二关"]
}
def level3_node(state: ScoreState):
"""第三关：获得30分"""
return {
"score": 30,
"actions": ["完成第三关"]
}
# 构建图
workflow = StateGraph(ScoreState)
workflow.add_node("level1", level1_node)
workflow.add_node("level2", level2_node)
workflow.add_node("level3", level3_node)
workflow.add_edge(START, "level1")
workflow.add_edge("level1", "level2")
workflow.add_edge("level2", "level3")
workflow.add_edge("level3", END)
app = workflow.compile()
# 运行
result = app.invoke({
"player_name": "玩家小明",
"score": 0,
"actions": []
})
print(f"玩家: {result['player_name']}")
print(f"总分: {result['score']}") # 输出: 60 (10+20+30)
print(f"完成动作: {result['actions']}") # 输出: ["完成第一关", "完成第二关", "完成第三关"]
常用 Reducer
| | |
|---|
operator.add | | int |
operator.mul | | int |
| | |
自定义 Reducer
def custom_reducer(left: int, right: int) -> int:
"""取最大值"""
return max(left, right)
class MyState(TypedDict):
max_score: Annotated[int, custom_reducer]
# 每次更新都会取最大值
5️⃣ 内置的特殊 State
MessagesState
LangGraph 提供了 MessagesState，专门用于对话场景：
from langgraph.graph import MessagesState
class ChatState(MessagesState):
"""继承 MessagesState，自动包含 messages 字段"""
user_name: str
conversation_count: int
# MessagesState 等价于：
class MessagesState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
add_messages Reducer 的作用：
📊 完整流程图
┌─────────────────────────────────────────────────────────────┐
│ LangGraph 核心概念 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ │
│ │ START │ │
│ └────┬────┘ │
│ │ Edge │
│ ↓ │
│ ┌─────────┐ │
│ │ Node1 │ ← 读取 State，处理后返回更新 │
│ └────┬────┘ │
│ │ State 合并（可能使用 Reducer） │
│ ↓ │
│ ┌─────────┐ │
│ │ Node2 │ │
│ └────┬────┘ │
│ │ │
│ ↓ │
│ ┌─────────┐ │
│ │ END │ │
│ └─────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘
🎯 小结
