03_状态管理和节点

发表于 更新于 分类于 ， Changyan： 本文字数： 3.5k 阅读时长 ≈ 13 分钟

深入理解LangGraph的核心机制：状态管理。本节课将详细讲解State的高级用法，包括列表、嵌套字典、状态合并策略等。通过实战案例，学会如何设计合理的状态结构，以及节点间如何安全高效地共享数据。

🎯 学习目标

学完本节课，你将能够：

  1. 掌握State的高级类型定义
  2. 理解状态的合并和累积机制
  3. 学会设计合理的状态结构
  4. 实现带历史记忆的多轮对话
  5. 避免常见的状态管理陷阱

1. 状态的本质

1.1 状态是什么？

回顾： 在Lesson 02中，我们把状态比作快递包裹。现在让我们更深入地理解它。

状态 = 流程的记忆

想象你在餐厅点餐：

  1. 服务员记下你的菜单（状态：订单）
  2. 厨房根据订单做菜（读取状态）
  3. 菜做好了，更新订单状态（修改状态）
  4. 服务员看到状态变化，上菜（响应状态）

在LangGraph中：

1
2
3
4
5
6
state = {
    "order": ["宫保鸡丁", "米饭"],  # 订单内容
    "status": "cooking",            # 当前状态
    "table_number": 5,               # 桌号
    "special_requests": "少辣"       # 特殊要求
}

1.2 状态的生命周期

1
2
3
[初始状态] → [节点1修改] → [节点2修改] → [节点3修改] → [最终状态]
    ↓              ↓              ↓              ↓
  {"input": ""}  {"step1": ""}  {"step2": ""}  {"result": ""}

关键点：

  • 状态在整个流程中持续存在
  • 每个节点都能读取和修改状态
  • 状态的修改会传递到下一个节点

2. 基础状态类型

2.1 简单类型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from typing import TypedDict

class State(TypedDict):
    # 字符串
    name: str
    message: str
    
    # 数字
    age: int
    score: float
    
    # 布尔值
    is_valid: bool
    success: bool

2.2 列表类型

1
2
3
4
5
6
7
8
9
10
11
from typing import TypedDict, List

class State(TypedDict):
    # 字符串列表
    tags: List[str]
    
    # 对话历史
    messages: List[str]
    
    # 数字列表
    scores: List[float]

示例：累积对话历史

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def chat_node(state: State) -> State:
    """每次调用都累积历史"""
    user_input = state["user_input"]
    
    # 追加到历史
    if "messages" not in state:
        state["messages"] = []
    
    state["messages"].append(f"用户: {user_input}")
    
    # 生成回复
    reply = f"收到：{user_input}"
    state["messages"].append(f"助手: {reply}")
    
    return state

2.3 字典类型

1
2
3
4
5
6
7
8
9
10
11
from typing import TypedDict, Dict, Any

class State(TypedDict):
    # 简单字典
    user_info: Dict[str, str]
    
    # 嵌套字典
    metadata: Dict[str, Any]
    
    # 统计信息
    stats: Dict[str, int]

示例：存储用户信息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def collect_info_node(state: State) -> State:
    """收集用户信息"""
    state["user_info"] = {
        "name": "小明",
        "age": "25",
        "city": "北京"
    }
    return state

def use_info_node(state: State) -> State:
    """使用用户信息"""
    info = state["user_info"]
    message = f"你好，{info['name']}！来自{info['city']}的朋友。"
    state["message"] = message
    return state

2.4 可选字段

使用 NotRequired 标记可选字段：

1
2
3
4
5
6
7
8
9
10
11
from typing import TypedDict
from typing_extensions import NotRequired

class State(TypedDict):
    # 必需字段
    user_input: str
    
    # 可选字段
    error: NotRequired[str]
    result: NotRequired[str]
    metadata: NotRequired[Dict[str, Any]]

好处：

  • IDE不会警告未初始化的可选字段
  • 代码更清晰，一眼看出哪些是必需的

3. 状态的合并策略

3.1 默认策略：覆盖

默认情况下，同名字段会被覆盖：

1
2
3
4
5
6
7
8
9
def node1(state: State) -> State:
    state["value"] = "A"
    return state

def node2(state: State) -> State:
    state["value"] = "B"  # 覆盖了A
    return state

# 结果：state["value"] = "B"

3.2 累积策略：列表追加

对于列表，通常希望追加而不是覆盖：

1
2
3
4
5
6
from typing import TypedDict, List, Annotated
from operator import add

class State(TypedDict):
    # 使用Annotated指定累积策略
    messages: Annotated[List[str], add]

效果：

1
2
3
4
5
6
7
8
9
def node1(state: State) -> State:
    state["messages"] = ["消息1"]
    return state

def node2(state: State) -> State:
    state["messages"] = ["消息2"]  # 不会覆盖，而是追加
    return state

# 结果：state["messages"] = ["消息1", "消息2"]

3.3 自定义合并策略

1
2
3
4
5
6
7
8
9
10
11
from typing import TypedDict, Annotated

def merge_dicts(left: dict, right: dict) -> dict:
    """自定义字典合并逻辑"""
    result = left.copy()
    result.update(right)
    return result

class State(TypedDict):
    # 使用自定义合并函数
    metadata: Annotated[Dict[str, Any], merge_dicts]

4. 实战：多轮对话机器人

4.1 需求分析

构建一个记住上下文的对话机器人：

  • 用户可以多次提问
  • AI能记住之前的对话
  • 可以引用历史信息

4.2 代码实现

创建 01_chat_with_memory.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
"""
带记忆的多轮对话机器人
学习目标：理解列表累积和对话历史管理
"""
import os
from typing import TypedDict, List, Annotated
from operator import add
from langgraph.graph import StateGraph, END
from langchain_community.llms import Tongyi
from dotenv import load_dotenv

load_dotenv()


# 定义状态
class State(TypedDict):
    """对话状态"""
    user_input: str                              # 当前用户输入
    messages: Annotated[List[str], add]         # 对话历史（自动累积）
    ai_response: str                             # AI回复


# 初始化LLM
def get_llm() -> Tongyi:
    """获取通义千问实例"""
    return Tongyi(
        model="qwen-turbo",
        api_key=os.getenv("DASHSCOPE_API_KEY"),
        temperature=0.7,
    )


# 节点1：记录用户输入
def record_user_input(state: State) -> State:
    """将用户输入添加到历史"""
    user_msg = f"用户: {state['user_input']}"
    state["messages"] = [user_msg]  # 会自动追加，不会覆盖
    print(f"[记录] {user_msg}")
    return state


# 节点2：AI生成回复
def generate_response(state: State) -> State:
    """基于历史生成回复"""
    llm = get_llm()
    
    # 构造包含历史的提示词
    history = "\n".join(state["messages"])
    prompt = f"""
你是一个友好的助手，能记住之前的对话内容。

对话历史：
{history}

请基于对话历史，回复用户的最新问题。如果问题涉及之前的内容，要体现出你记得。
直接输出回复内容，不要有其他说明。
"""
    
    print("[AI思考中...]")
    response = llm.invoke(prompt)
    
    state["ai_response"] = response
    
    # 记录AI回复到历史
    ai_msg = f"助手: {response}"
    state["messages"] = [ai_msg]  # 会追加到历史
    print(f"[AI回复] {response}\n")
    
    return state


# 构建图
def create_graph() -> StateGraph:
    """创建对话图"""
    graph = StateGraph(State)
    
    graph.add_node("record", record_user_input)
    graph.add_node("respond", generate_response)
    
    graph.set_entry_point("record")
    graph.add_edge("record", "respond")
    graph.add_edge("respond", END)
    
    return graph


def main() -> None:
    """多轮对话示例"""
    app = create_graph().compile()
    
    # 初始状态（空历史）
    state = {"messages": []}
    
    # 模拟多轮对话
    conversations = [
        "我叫小明，今年25岁",
        "我喜欢编程和阅读",
        "你还记得我叫什么吗？",
        "我今年多大了？",
        "我有哪些爱好？"
    ]
    
    print("="*60)
    print("多轮对话演示")
    print("="*60)
    
    for i, user_input in enumerate(conversations, 1):
        print(f"\n{'='*60}")
        print(f"第 {i} 轮对话")
        print(f"{'='*60}")
        
        # 调用图
        state["user_input"] = user_input
        state = app.invoke(state)
        
        print(f"\n当前历史记录数: {len(state['messages'])}")
    
    # 打印完整历史
    print(f"\n{'='*60}")
    print("完整对话历史")
    print(f"{'='*60}")
    for msg in state["messages"]:
        print(msg)


if __name__ == "__main__":
    main()

4.3 运行效果

1
python 01_chat_with_memory.py

预期输出：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
==================================================
第 1 轮对话
==================================================
[记录] 用户: 我叫小明，今年25岁
[AI思考中...]
[AI回复] 你好，小明！很高兴认识你。25岁正是美好的年纪呢！

当前历史记录数: 2

==================================================
第 3 轮对话
==================================================
[记录] 用户: 你还记得我叫什么吗？
[AI思考中...]
[AI回复] 当然记得！你叫小明，刚才你告诉我的。😊

当前历史记录数: 6

4.4 代码要点解析

1️⃣ 列表自动累积

1
2
# 关键：使用Annotated + add
messages: Annotated[List[str], add]

这样定义后：

1
2
3
4
5
6
7
# 节点1
state["messages"] = ["消息1"]  # 添加消息1

# 节点2
state["messages"] = ["消息2"]  # 添加消息2，不会覆盖

# 结果：["消息1", "消息2"]

2️⃣ 状态在多次invoke间传递

1
2
3
4
5
6
7
# 第一次调用
state = app.invoke({"user_input": "你好", "messages": []})
# state["messages"] = ["用户: 你好", "助手: 你好！"]

# 第二次调用：传入前一次的state
state = app.invoke(state)
# state["messages"] = ["用户: 你好", "助手: 你好！", "用户: ...", "助手: ..."]

核心理念： 状态是持续累积的，不是每次都从零开始。

5. 状态设计最佳实践

5.1 原则1：扁平化优于嵌套

❌ 过度嵌套（不推荐）：

1
2
class State(TypedDict):
    data: Dict[str, Dict[str, Dict[str, Any]]]  # 太复杂！

✅ 扁平化（推荐）：

1
2
3
4
class State(TypedDict):
    user_name: str
    user_age: int
    user_city: str

5.2 原则2：明确字段用途

✅ 清晰的命名：

1
2
3
4
5
class State(TypedDict):
    user_input: str          # 输入
    processed_text: str      # 处理后的文本
    intent: str              # 识别的意图
    final_response: str      # 最终回复

❌ 模糊的命名：

1
2
3
4
class State(TypedDict):
    data1: str
    data2: str
    result: str

5.3 原则3：只存必要数据

❌ 存储过多：

1
2
3
4
5
6
7
def node(state: State) -> State:
    # 把所有中间变量都存到state
    state["temp1"] = ...
    state["temp2"] = ...
    state["temp3"] = ...
    state["debug_info"] = ...
    # 状态变得臃肿

✅ 只存必要数据：

1
2
3
4
5
6
7
def node(state: State) -> State:
    # 中间变量在函数内部
    temp1 = ...
    temp2 = ...
    
    # 只把需要传递的数据放入state
    state["result"] = final_result

5.4 原则4：使用类型注解

✅ 完整的类型注解：

1
2
3
4
5
6
7
from typing import TypedDict, List, Dict, Optional

class State(TypedDict):
    user_input: str
    messages: List[str]
    metadata: Dict[str, str]
    error: Optional[str]  # 可能为None

5.5 原则5：分离输入和输出

1
2
3
4
5
6
7
8
9
10
11
12
class State(TypedDict):
    # 输入字段（只读）
    user_query: str
    user_id: str
    
    # 中间结果
    intent: str
    entities: List[str]
    
    # 输出字段
    response: str
    status: str

6. 高级示例：信息收集机器人

6.1 需求

构建一个信息收集机器人：

  1. 依次询问：姓名、年龄、城市
  2. 收集完毕后，生成总结
  3. 可以看到收集进度

6.2 实现

创建 02_info_collector.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
"""
信息收集机器人
学习目标：状态驱动的多步骤流程
"""
from typing import TypedDict, List, Dict
from typing_extensions import NotRequired
from langgraph.graph import StateGraph, END


# 定义状态
class State(TypedDict):
    """信息收集状态"""
    # 要收集的字段
    to_collect: List[str]
    
    # 当前询问的字段
    current_field: NotRequired[str]
    
    # 已收集的信息
    collected_info: Dict[str, str]
    
    # 用户输入
    user_input: NotRequired[str]
    
    # 系统提示
    prompt: NotRequired[str]
    
    # 是否完成
    is_complete: bool


# 字段的友好名称和提示
FIELD_PROMPTS = {
    "name": "请问您叫什么名字？",
    "age": "请问您多大了？",
    "city": "请问您来自哪个城市？"
}


# 节点1：确定下一个要收集的字段
def determine_next_field(state: State) -> State:
    """确定下一个要收集的字段"""
    collected = state["collected_info"]
    to_collect = state["to_collect"]
    
    # 找出还未收集的字段
    remaining = [f for f in to_collect if f not in collected]
    
    if remaining:
        # 还有未收集的
        next_field = remaining[0]
        state["current_field"] = next_field
        state["prompt"] = FIELD_PROMPTS[next_field]
        state["is_complete"] = False
        print(f"\n[系统] {state['prompt']}")
    else:
        # 全部收集完毕
        state["is_complete"] = True
        print("\n[系统] 信息收集完毕！")
    
    return state


# 节点2：保存用户输入
def save_input(state: State) -> State:
    """保存用户输入到已收集信息"""
    if "current_field" in state and "user_input" in state:
        field = state["current_field"]
        value = state["user_input"]
        state["collected_info"][field] = value
        print(f"[保存] {field} = {value}")
    
    return state


# 节点3：生成总结
def generate_summary(state: State) -> State:
    """生成收集结果的总结"""
    info = state["collected_info"]
    
    summary = f"""
{'='*50}
📋 信息收集完成
{'='*50}
姓名: {info.get('name', '未知')}
年龄: {info.get('age', '未知')}
城市: {info.get('city', '未知')}
{'='*50}
"""
    print(summary)
    return state


# 路由函数：决定下一步
def route_next(state: State) -> str:
    """根据是否完成决定路由"""
    if state["is_complete"]:
        return "summary"  # 完成 → 生成总结
    else:
        return "collect"  # 未完成 → 继续收集


# 构建图
def create_graph() -> StateGraph:
    """创建信息收集图"""
    graph = StateGraph(State)
    
    # 添加节点
    graph.add_node("determine", determine_next_field)
    graph.add_node("collect", save_input)
    graph.add_node("summary", generate_summary)
    
    # 设置流程
    graph.set_entry_point("determine")
    
    # determine → 根据是否完成决定路由
    graph.add_conditional_edges(
        "determine",
        route_next,
        {
            "collect": "collect",    # 继续收集
            "summary": "summary"     # 生成总结
        }
    )
    
    # collect → determine（循环）
    graph.add_edge("collect", "determine")
    
    # summary → END
    graph.add_edge("summary", END)
    
    return graph


def main() -> None:
    """运行信息收集"""
    app = create_graph().compile()
    
    # 初始状态
    state = {
        "to_collect": ["name", "age", "city"],
        "collected_info": {},
        "is_complete": False
    }
    
    # 模拟用户输入
    user_inputs = ["小明", "25", "北京"]
    
    print("="*50)
    print("信息收集机器人")
    print("="*50)
    
    for user_input in user_inputs:
        # 先调用一次，显示提示
        state = app.invoke(state)
        
        # 用户输入
        print(f"[用户] {user_input}")
        state["user_input"] = user_input
        
        # 再调用一次，保存输入
        state = app.invoke(state)


if __name__ == "__main__":
    main()

6.3 运行效果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
==================================================
信息收集机器人
==================================================

[系统] 请问您叫什么名字？
[用户] 小明
[保存] name = 小明

[系统] 请问您多大了？
[用户] 25
[保存] age = 25

[系统] 请问您来自哪个城市？
[用户] 北京
[保存] city = 北京

[系统] 信息收集完毕！

==================================================
📋 信息收集完成
==================================================
姓名: 小明
年龄: 25
城市: 北京
==================================================

6.4 关键技术点

1️⃣ 循环流程

1
2
3
4
# collect节点执行完后，回到determine节点
graph.add_edge("collect", "determine")

# 形成循环：determine → collect → determine → collect → ...

2️⃣ 条件路由

1
2
3
4
5
6
7
8
9
10
11
def route_next(state: State) -> str:
    """返回下一个节点的名称"""
    if state["is_complete"]:
        return "summary"
    else:
        return "collect"

graph.add_conditional_edges("determine", route_next, {
    "collect": "collect",
    "summary": "summary"
})

3️⃣ 状态驱动

整个流程由状态驱动：

  • to_collect - 定义要收集什么
  • collected_info - 记录已收集的内容
  • is_complete - 决定是否结束

7. 常见陷阱和解决方案

陷阱1：忘记返回state

❌ 错误：

1
2
3
def my_node(state: State):
    state["result"] = "done"
    # 忘记return了！

✅ 正确：

1
2
3
def my_node(state: State) -> State:
    state["result"] = "done"
    return state

陷阱2：直接修改列表引用

❌ 错误：

1
2
3
4
5
def my_node(state: State) -> State:
    messages = state["messages"]
    messages.append("new")  # 修改了引用
    # 但没有触发state更新
    return state

✅ 正确：

1
2
3
4
5
6
7
8
def my_node(state: State) -> State:
    # 方式1：重新赋值
    state["messages"] = state["messages"] + ["new"]
    
    # 方式2：使用累积策略
    state["messages"] = ["new"]  # 会自动追加
    
    return state

陷阱3：状态字段过多

❌ 过度设计：

1
2
3
4
5
class State(TypedDict):
    field1: str
    field2: str
    # ... 20个字段
    field20: str

✅ 分组设计：

1
2
3
4
5
6
7
8
class UserInfo(TypedDict):
    name: str
    age: int
    city: str

class State(TypedDict):
    user: UserInfo
    result: str

陷阱4：在state中存储函数或对象

❌ 不要这样做：

1
2
3
def my_node(state: State) -> State:
    state["llm"] = ChatOpenAI()  # 不要存储对象！
    state["func"] = my_function  # 不要存储函数！

✅ 正确做法：

1
2
3
4
5
6
7
8
# 在节点外部创建
llm = ChatOpenAI()

def my_node(state: State) -> State:
    # 直接使用外部变量
    response = llm.invoke(state["input"])
    state["output"] = response
    return state

8. 练习题

练习1：购物车（基础）

实现一个购物车状态管理：

  • 可以添加商品（名称、价格）
  • 可以删除商品
  • 计算总价

练习2：问答机器人（进阶）

实现一个三轮问答：

  • 第1轮：问”你的爱好是什么？”
  • 第2轮：问”为什么喜欢这个爱好？”
  • 第3轮：根据前两轮的回答，给出建议

练习3：表单验证（高级）

实现一个带验证的表单收集：

  • 收集：用户名、邮箱、手机号
  • 验证：
    • 用户名：3-20个字符
    • 邮箱：包含@和.
    • 手机号：11位数字
  • 如果验证失败，重新询问该字段

9. 小结

核心要点

✅ 状态的本质： 流程的记忆，在节点间传递数据

✅ 状态类型：

  • 简单类型：str, int, bool
  • 复合类型：List, Dict
  • 可选字段：NotRequired

✅ 合并策略：

  • 默认：覆盖
  • 列表：使用 Annotated[List[T], add] 自动追加
  • 自定义：实现自己的合并函数

✅ 设计原则：

  1. 扁平化优于嵌套
  2. 明确字段用途
  3. 只存必要数据
  4. 使用类型注解
  5. 分离输入和输出

✅ 常见陷阱：

  • 忘记返回state
  • 直接修改列表引用
  • 状态字段过多
  • 存储不可序列化的对象

📚 下一步

现在你已经掌握了状态管理的核心技能！但流程控制不止简单的顺序执行，还需要条件分支、循环、并行等高级功能。

下一课： Lesson 04: 图的流转控制

我们将学习：

  • 条件边：根据结果选择不同路径
  • 循环：让流程回到之前的节点
  • 并行：同时执行多个节点
  • 子图：组织复杂的流程

📖 参考资料

  • LangGraph State管理文档
  • Python TypedDict文档
  • typing_extensions文档