LangGraph State状态管理:Agent的”记忆系统”

从零开始理解LangGraph最核心的概念——State状态管理,让你的Agent拥有真正的”记忆”。

封面

引言:为什么需要状态管理?

想象一下,你和一位朋友聊天。你说:”我昨天去看了《星际穿越》。”朋友回答:”那部电影太棒了!”然后你接着问:”你觉得结局是什么意思?”

这里有个关键问题——朋友怎么知道”那部电影”指的是《星际穿越》?

因为他记住了之前的对话。

这就是状态管理的本质:让程序记住之前发生过的事情

无状态的痛苦

让我们先看一个没有状态管理的简单”机器人”:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 一个无状态的"机器人"
def simple_bot(user_input):
# 每次都只处理当前输入,完全不记得之前说过什么
if "你好" in user_input:
return "你好!很高兴见到你。"
elif "天气" in user_input:
return "今天天气不错。"
else:
return "我不太明白你的意思。"

# 模拟对话
print("用户:你好")
print("机器人:", simple_bot("你好"))

print("\n用户:今天天气怎么样?")
print("机器人:", simple_bot("今天天气怎么样?"))

print("\n用户:那你喜欢这种天气吗?")
print("机器人:", simple_bot("那你喜欢这种天气吗?"))
# 输出:我不太明白你的意思。
# 问题:机器人完全忘记了刚才在聊天气!

看到了吗?当用户问”那你喜欢这种天气吗?”时,机器人一脸懵逼——它根本不知道”这种天气”是什么。

这就是无状态的痛苦:每次对话都是全新的开始,没有上下文,没有记忆,就像金鱼一样(据说金鱼只有7秒记忆)。

有状态的美好

现在让我们看看有状态管理的世界:

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
# 一个有记忆的机器人
class StatefulBot:
def __init__(self):
self.history = [] # 这就是状态——记住对话历史

def chat(self, user_input):
# 保存用户输入到记忆中
self.history.append({"role": "user", "content": user_input})

# 基于完整历史生成回复
if len(self.history) == 1:
response = "你好!很高兴见到你。"
elif "天气" in user_input:
response = "今天天气不错。"
elif "这种天气" in user_input and any("天气" in h["content"] for h in self.history[:-1]):
response = "我喜欢晴朗的天气,让人心情愉快!"
else:
response = "我不太明白你的意思。"

# 保存机器人的回复到记忆中
self.history.append({"role": "bot", "content": response})
return response

# 模拟有记忆的对话
bot = StatefulBot()

print("用户:你好")
print("机器人:", bot.chat("你好"))

print("\n用户:今天天气怎么样?")
print("机器人:", bot.chat("今天天气怎么样?"))

print("\n用户:那你喜欢这种天气吗?")
print("机器人:", bot.chat("那你喜欢这种天气吗?"))
# 输出:我喜欢晴朗的天气,让人心情愉快!

现在机器人记得刚才在聊天气了!这就是状态的魔力。

从简单机器人到LangGraph

上面的例子虽然展示了状态的概念,但太过简单。在实际应用中,我们需要:

  1. 更复杂的状态结构——不只是对话历史,可能还有用户偏好、中间计算结果、错误状态等
  2. 状态更新规则——什么时候覆盖、什么时候追加、什么时候清空
  3. 多步骤流程——一个Agent可能要经历”理解→规划→执行→验证”多个阶段
  4. 可视化调试——能看到状态是如何一步步变化的

LangGraph就是为解决这些问题而生的。

核心概念:TypedDict vs Pydantic,reducer概念

在深入代码之前,我们需要理解LangGraph中状态管理的三个核心概念。

概念1:State(状态)是什么?

在LangGraph中,State就是一个Python类,用来定义你的Agent需要记住的所有信息

打个比方:State就像是Agent的”记忆笔记本”,每一页记录不同类型的信息。有的页面记对话历史,有的记用户偏好,有的记当前任务进度…

LangGraph提供了两种方式来定义State:TypedDictPydantic

概念2:TypedDict方式

TypedDict是Python标准库的一部分(Python 3.8+),它让我们可以用类似字典的方式定义结构化数据。

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

# 定义一个简单的状态
class ChatState(TypedDict):
messages: List[str] # 对话历史
user_name: str # 用户名
turn_count: int # 轮数计数

TypedDict的特点:

  • ✅ 标准库,无需额外依赖
  • ✅ 轻量级,性能好
  • ✅ 类型提示友好
  • ❌ 运行时类型检查较弱(只是提示,不强制)

什么时候用TypedDict?

  • 追求简单和性能
  • 不需要复杂的验证逻辑
  • 团队对类型安全要求不是特别严格

概念3:Pydantic方式

Pydantic是一个强大的数据验证库,被FastAPI等流行框架广泛使用。

1
2
3
4
5
6
7
8
9
10
11
12
from pydantic import BaseModel, Field
from typing import List

# 使用Pydantic定义状态
class ChatState(BaseModel):
messages: List[str] = Field(default_factory=list, description="对话历史")
user_name: str = Field(default="", description="用户名")
turn_count: int = Field(default=0, ge=0, description="轮数计数,必须大于等于0")

class Config:
# 允许额外字段
extra = "allow"

Pydantic的特点:

  • ✅ 强大的数据验证(类型、范围、格式等)
  • ✅ 清晰的错误提示
  • ✅ 自动生成文档
  • ✅ 支持默认值、别名等高级功能
  • ❌ 稍微重一点(但通常可以忽略)

什么时候用Pydantic?

  • 需要严格的数据验证
  • 对外暴露API接口
  • 团队协作,需要清晰的契约
  • 需要默认值和复杂配置

TypedDict vs Pydantic 对比

特性 TypedDict Pydantic
性能 ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐
验证能力 ⭐⭐ ⭐⭐⭐⭐⭐
学习成本 ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐
生态兼容 ⭐⭐⭐⭐ ⭐⭐⭐⭐⭐
推荐场景 内部工具、原型 生产服务、API

我的建议:新手从TypedDict开始,熟悉后再根据需要切换到Pydantic。

概念4:Reducer(状态更新器)

这是LangGraph中最精妙的设计之一。

问题:当多个节点都想修改同一个状态时,应该怎么做?

简单方案(有问题):直接覆盖

1
2
3
# 节点A设置 state["count"] = 5
# 节点B设置 state["count"] = 10
# 结果:count = 10(A的修改被覆盖了!)

更好的方案:告诉LangGraph如何”合并”状态更新

这就是Reducer的作用——它定义了当新值来临时,应该如何与旧值合并。

1
2
3
4
5
6
7
8
9
10
from typing import Annotated
from operator import add

# 定义reducer
class ChatState(TypedDict):
# messages字段使用"追加"策略
messages: Annotated[List[str], add]

# count字段默认使用"覆盖"策略
count: int

内置的reducer:

  1. operator.add —— 用于列表,表示追加

    1
    2
    3
    old = [1, 2]
    new = [3, 4]
    result = [1, 2, 3, 4] # 追加
  2. 自定义reducer函数 —— 你可以写任意逻辑

    1
    2
    3
    4
    5
    def keep_max(existing, new):
    return max(existing, new)

    class State(TypedDict):
    score: Annotated[int, keep_max]
  3. lambda x, y: y —— 覆盖(默认行为)

为什么reducer这么重要?

因为在LangGraph的并发执行中,多个节点可能同时产生状态更新。Reducer让你能精确控制这些更新如何合并,避免数据丢失。

实战代码:带历史记录的对话机器人

理论说得够多了,让我们动手写一个完整的、带历史记录的对话机器人!

项目结构

1
2
3
4
langgraph-chatbot/
├── chatbot.py # 主程序
├── requirements.txt # 依赖
└── .env # 环境变量(API密钥)

步骤1:安装依赖

创建 requirements.txt

1
2
3
langgraph>=0.2.0
langchain-openai>=0.2.0
python-dotenv>=1.0.0

安装:

1
pip install -r requirements.txt

步骤2:完整代码

创建 chatbot.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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
"""
LangGraph 状态管理实战:带历史记录的对话机器人
作者:Cypher
日期:2026-02-20
"""

import os
from typing import TypedDict, List, Annotated, Literal
from operator import add
from datetime import datetime

from dotenv import load_dotenv
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# 加载环境变量
load_dotenv()

# ========== 1. 定义状态 ==========

class ChatState(TypedDict):
"""
定义聊天机器人的状态结构

每个字段代表Agent需要记忆的一种信息
"""
# messages: 对话历史,使用add reducer表示追加而非覆盖
messages: Annotated[List[dict], add]

# user_name: 用户名(覆盖模式)
user_name: str

# turn_count: 对话轮数(覆盖模式)
turn_count: int

# current_intent: 当前识别的用户意图(覆盖模式)
current_intent: str

# emotion: 检测到的情绪(覆盖模式)
emotion: str


# ========== 2. 定义节点(Node) ==========

# 初始化语言模型
# 注意:请确保设置了 OPENAI_API_KEY 环境变量
llm = ChatOpenAI(
model="gpt-3.5-turbo",
temperature=0.7,
api_key=os.getenv("OPENAI_API_KEY")
)


def intent_classifier(state: ChatState) -> ChatState:
"""
节点1:意图识别
分析用户最后一条消息,识别意图和情绪
"""
# 获取最后一条用户消息
last_message = state["messages"][-1]["content"] if state["messages"] else ""

# 简单的规则匹配(实际项目中可以用LLM做更复杂的分析)
intent = "general"
emotion = "neutral"

if any(word in last_message.lower() for word in ["你好", "hi", "hello"]):
intent = "greeting"
elif any(word in last_message.lower() for word in ["再见", "拜拜", "bye"]):
intent = "farewell"
elif any(word in last_message.lower() for word in ["帮助", "help", "怎么做"]):
intent = "help"
elif "?" in last_message or "?" in last_message:
intent = "question"

# 情绪检测
if any(word in last_message for word in ["棒", "好", "赞", "😊", "开心"]):
emotion = "happy"
elif any(word in last_message for word in ["差", "烂", "糟", "😠", "生气"]):
emotion = "angry"
elif any(word in last_message for word in ["难过", "😢", "伤心"]):
emotion = "sad"

# 返回状态更新(注意:只返回要修改的字段)
return {
"current_intent": intent,
"emotion": emotion,
"turn_count": state.get("turn_count", 0) + 1
}


def generate_response(state: ChatState) -> ChatState:
"""
节点2:生成回复
基于历史对话和识别的意图生成AI回复
"""
messages = state["messages"]
intent = state["current_intent"]
emotion = state["emotion"]
user_name = state.get("user_name", "")

# 构建系统提示词
system_prompt = f"""你是一个友好的AI助手。当前信息:
- 用户意图: {intent}
- 用户情绪: {emotion}
- 用户称呼: {user_name if user_name else '未设置'}

请根据用户的情绪和意图,给出恰当的回复。保持友好、有帮助的语气。"""

# 转换消息格式为LangChain格式
langchain_messages = [SystemMessage(content=system_prompt)]
for msg in messages:
if msg["role"] == "user":
langchain_messages.append(HumanMessage(content=msg["content"]))
elif msg["role"] == "assistant":
langchain_messages.append(AIMessage(content=msg["content"]))

# 调用LLM生成回复
try:
response = llm.invoke(langchain_messages)
ai_message = response.content
except Exception as e:
ai_message = f"抱歉,我遇到了一点问题:{str(e)[:50]}..."

# 根据意图做特殊处理
if intent == "greeting" and not user_name:
ai_message += "\n\n对了,我还不知道该怎么称呼你呢?"
elif intent == "farewell":
ai_message += "\n\n希望很快再见到你!👋"

# 返回新的AI消息(会被add reducer追加到messages列表)
return {
"messages": [{"role": "assistant", "content": ai_message}]
}


def check_exit(state: ChatState) -> Literal["continue", "exit"]:
"""
条件边:检查是否应该结束对话
返回"exit"表示结束,"continue"表示继续
"""
intent = state["current_intent"]
turn_count = state["turn_count"]

# 如果用户说再见,或者对话超过20轮,结束对话
if intent == "farewell":
return "exit"
if turn_count >= 20:
return "exit"

return "continue"


# ========== 3. 构建图 ==========

# 创建状态图
workflow = StateGraph(ChatState)

# 添加节点
workflow.add_node("intent_classifier", intent_classifier)
workflow.add_node("generate_response", generate_response)

# 添加边
# 从意图分类器到回复生成器
workflow.add_edge("intent_classifier", "generate_response")

# 条件边:从回复生成器判断下一步
workflow.add_conditional_edges(
"generate_response",
check_exit,
{
"continue": "intent_classifier", # 继续对话,等待新输入
"exit": END # 结束对话
}
)

# 设置入口点
workflow.set_entry_point("intent_classifier")

# 编译图
app = workflow.compile()


# ========== 4. 运行机器人 ==========

def print_state(state: ChatState, title: str = "当前状态"):
"""打印当前状态(调试用)"""
print(f"\n{'='*50}")
print(f"📊 {title}")
print(f"{'='*50}")
print(f"对话轮数: {state.get('turn_count', 0)}")
print(f"当前意图: {state.get('current_intent', 'N/A')}")
print(f"情绪状态: {state.get('emotion', 'N/A')}")
print(f"用户名: {state.get('user_name', '未设置')}")
print(f"\n对话历史:")
for i, msg in enumerate(state.get('messages', []), 1):
role = "👤 用户" if msg["role"] == "user" else "🤖 AI"
content = msg["content"][:100] + "..." if len(msg["content"]) > 100 else msg["content"]
print(f" {i}. {role}: {content}")
print(f"{'='*50}\n")


def run_chatbot():
"""运行交互式聊天机器人"""
print("🤖 LangGraph 对话机器人已启动!")
print("输入 'quit' 或 '再见' 结束对话\n")

# 初始化状态
state: ChatState = {
"messages": [],
"user_name": "",
"turn_count": 0,
"current_intent": "",
"emotion": ""
}

while True:
# 获取用户输入
user_input = input("👤 你: ").strip()

if not user_input:
continue

# 检查退出
if user_input.lower() in ["quit", "exit", "再见", "拜拜"]:
print("🤖 AI: 再见!祝你有美好的一天!👋")
break

# 提取用户名(简单规则:如果用户说"我叫XXX")
if "我叫" in user_input:
name = user_input.split("我叫")[-1].strip().rstrip("。!?")
state["user_name"] = name
print(f"🤖 AI: 很高兴认识你,{name}!")
continue

# 添加用户消息到状态
state["messages"].append({"role": "user", "content": user_input})

# 运行图
# 注意:每次只运行一步,因为我们需要等待用户输入
result = app.invoke(state)

# 更新状态
state = result

# 打印AI回复(最后一条消息)
if state["messages"]:
last_msg = state["messages"][-1]
if last_msg["role"] == "assistant":
print(f"🤖 AI: {last_msg['content']}")

# 调试用:打印完整状态
# print_state(state)

# 检查是否应该退出
if state["current_intent"] == "farewell":
print("\n💡 提示:检测到告别意图,下次输入将结束对话。")


# ========== 5. 流式运行示例 ==========

def run_with_streaming():
"""使用流式模式运行,可以观察状态变化"""
print("🤖 LangGraph 流式对话机器人已启动!\n")

state: ChatState = {
"messages": [{"role": "user", "content": "你好!今天天气真不错。"}],
"user_name": "小明",
"turn_count": 0,
"current_intent": "",
"emotion": ""
}

print("初始状态:")
print_state(state)

print("🔄 开始执行图...\n")

# 流式执行,观察每一步的状态变化
for step, (node_name, node_state) in enumerate(app.stream(state)):
print(f"\n📍 步骤 {step + 1}: 节点 '{node_name}'")
print(f" 意图: {node_state.get('current_intent', 'N/A')}")
print(f" 情绪: {node_state.get('emotion', 'N/A')}")
print(f" 轮数: {node_state.get('turn_count', 0)}")

if node_state.get("messages"):
last_msg = node_state["messages"][-1]
if last_msg["role"] == "assistant":
preview = last_msg["content"][:80] + "..."
print(f" AI回复: {preview}")

print("\n✅ 执行完成!")


if __name__ == "__main__":
# 模式1:交互式对话(推荐)
# run_chatbot()

# 模式2:流式演示(用于观察状态变化)
run_with_streaming()

# 取消注释上面的 run_chatbot() 来进行真正的交互式对话

步骤3:配置API密钥

创建 .env 文件:

1
OPENAI_API_KEY=your_openai_api_key_here

步骤4:运行

1
python chatbot.py

你会看到流式执行的结果,展示状态是如何一步步变化的。

代码要点解析

  1. 状态定义

    1
    2
    3
    4
    class ChatState(TypedDict):
    messages: Annotated[List[dict], add] # 使用reducer
    user_name: str # 默认覆盖
    turn_count: int # 默认覆盖
  2. 节点函数:每个节点接收当前状态,返回状态更新

    1
    2
    3
    def intent_classifier(state: ChatState) -> ChatState:
    # 分析意图...
    return {"current_intent": intent, "emotion": emotion}
  3. 流式执行:可以观察每一步的状态变化

    1
    2
    for node_name, node_state in app.stream(state):
    print(f"节点 {node_name} 更新了状态")

状态更新模式:覆盖 vs 追加

在实际应用中,不同的字段需要不同的更新策略。让我们深入理解这两种模式。

模式1:覆盖(Replace)

这是默认行为,适用于大多数字段。

适用场景:

  • 当前步骤、当前意图、当前情绪
  • 用户偏好设置
  • 计算结果(如总分、平均分)
1
2
3
4
class State(TypedDict):
current_step: str # 覆盖:永远只关心当前在哪一步
user_mood: str # 覆盖:只保存最新的情绪
total_score: float # 覆盖:计算后直接替换

示例:

1
2
3
4
5
6
7
8
9
# 节点A
def step_a(state: State):
return {"current_step": "processing"}

# 节点B
def step_b(state: State):
return {"current_step": "completed"} # 覆盖了processing

# 结果:current_step = "completed"

模式2:追加(Append)

使用Annotatedadd实现,适用于需要保留历史的字段。

适用场景:

  • 对话历史
  • 操作日志
  • 中间结果集合
1
2
3
4
5
6
from typing import Annotated
from operator import add

class State(TypedDict):
messages: Annotated[List[dict], add] # 追加:保留所有消息
logs: Annotated[List[str], add] # 追加:保留所有日志

示例:

1
2
3
4
5
6
7
8
9
10
11
12
# 节点A
def node_a(state: State):
return {"messages": [{"role": "ai", "content": "你好"}]}

# 节点B
def node_b(state: State):
return {"messages": [{"role": "ai", "content": "有什么可以帮你的?"}]}

# 结果:messages = [
# {"role": "ai", "content": "你好"},
# {"role": "ai", "content": "有什么可以帮你的?"}
# ]

模式3:自定义Reducer

当内置的reducer不能满足需求时,你可以写自己的。

示例1:保留最大值

1
2
3
4
5
6
def keep_max(existing: int, new: int) -> int:
"""保留较大值"""
return max(existing, new)

class State(TypedDict):
best_score: Annotated[int, keep_max] # 只保留最高分

示例2:合并字典

1
2
3
4
5
6
7
8
9
def merge_dicts(existing: dict, new: dict) -> dict:
"""深度合并两个字典"""
result = existing.copy()
result.update(new)
return result

class State(TypedDict):
user_profile: Annotated[dict, merge_dicts]
# 新的profile字段会合并到旧的,而不是完全替换

示例3:限制列表长度

1
2
3
4
5
6
7
def limited_append(existing: List[str], new: List[str], max_size: int = 10) -> List[str]:
"""追加新元素,但限制总长度"""
combined = existing + new
return combined[-max_size:] # 只保留最后N个

class State(TypedDict):
recent_actions: Annotated[List[str], lambda x, y: limited_append(x, y, 5)]

实战建议

字段类型 推荐模式 原因
对话历史 追加 需要完整上下文
当前状态 覆盖 只关心现在
计数器 覆盖/自定义 通常需要重新计算
配置参数 覆盖 新配置替换旧配置
日志记录 追加 保留完整轨迹
中间结果 视情况 有些要保留,有些要覆盖

调试技巧:打印状态变化轨迹

调试状态管理最头疼的就是”不知道状态现在是什么样”。这里分享几个实用的调试技巧。

技巧1:使用stream()观察每一步

1
2
3
4
5
# 流式执行,可以看到每个节点的输出
for node_name, node_state in app.stream(initial_state):
print(f"\n{'='*40}")
print(f"节点: {node_name}")
print(f"状态更新: {node_state}")

技巧2:添加调试节点

1
2
3
4
5
6
7
8
9
10
11
12
13
def debug_node(state: ChatState) -> ChatState:
"""一个什么都不做、只打印状态的调试节点"""
print("\n🔍 DEBUG 状态快照:")
for key, value in state.items():
# 截断长内容
display_value = str(value)[:100] + "..." if len(str(value)) > 100 else value
print(f" {key}: {display_value}")
return {} # 不修改任何状态

# 在关键位置插入调试节点
workflow.add_node("debug", debug_node)
workflow.add_edge("intent_classifier", "debug")
workflow.add_edge("debug", "generate_response")

技巧3:状态变化对比

1
2
3
4
5
6
7
8
9
10
11
12
13
def print_state_diff(old_state: dict, new_state: dict):
"""打印两个状态的差异"""
print("\n📊 状态变化:")
all_keys = set(old_state.keys()) | set(new_state.keys())

for key in sorted(all_keys):
old_val = old_state.get(key, "<未设置>")
new_val = new_state.get(key, "<未设置>")

if old_val != new_val:
print(f" {key}:")
print(f" - {old_val}")
print(f" + {new_val}")

技巧4:可视化状态图

1
2
3
4
5
6
7
# LangGraph支持生成Mermaid图表
from langgraph.graph import StateGraph

# ... 构建你的图 ...

# 生成并打印Mermaid语法
print(app.get_graph().draw_mermaid())

把输出的Mermaid语法粘贴到 Mermaid Live Editor 就能看到漂亮的流程图。

技巧5:使用LangSmith追踪(生产环境)

1
2
3
4
5
6
7
8
9
import os

# 设置LangSmith环境变量
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your_langsmith_api_key"
os.environ["LANGCHAIN_PROJECT"] = "my-chatbot"

# 现在每次运行都会自动记录到LangSmith
result = app.invoke(state)

在LangSmith界面上,你可以看到:

  • 完整的执行轨迹
  • 每个节点的输入输出
  • 状态变化的时序图
  • 性能分析

总结与下篇预告

今天学到了什么?

  1. 为什么需要状态管理——让Agent拥有”记忆”,支持上下文对话
  2. TypedDict vs Pydantic——两种定义状态的方式,根据需求选择
  3. Reducer概念——控制状态如何更新,避免数据丢失
  4. 完整实战——构建了一个带历史记录的对话机器人
  5. 调试技巧——多种方法观察状态变化

关键要点回顾

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 状态定义
class State(TypedDict):
messages: Annotated[List[dict], add] # 追加模式
count: int # 覆盖模式(默认)

# 节点函数
def my_node(state: State) -> State:
return {"count": state["count"] + 1} # 返回更新

# 构建图
workflow = StateGraph(State)
workflow.add_node("node", my_node)
workflow.set_entry_point("node")
app = workflow.compile()

# 运行
result = app.invoke({"messages": [], "count": 0})

下篇预告

在下一篇文章中,我们将探讨:

《LangGraph 条件边与路由:让Agent学会”思考”》

我们会学习:

  • 如何根据状态动态决定下一步
  • 构建更复杂的工作流(循环、分支、并行)
  • 实际案例:智能客服路由系统

敬请期待!


练习与思考

  1. 动手练习:修改示例代码,添加一个”记忆用户喜好”的功能(如记住用户喜欢的颜色)

  2. 思考题:在你的项目中,哪些数据适合用”追加”模式,哪些适合”覆盖”模式?

  3. 进阶挑战:尝试用Pydantic重写状态定义,添加字段验证(如turn_count不能为负数)


如果有任何问题,欢迎在评论区留言! 🚀


📚 LangGraph 入门系列

本系列共10篇文章,带你从零掌握 LangGraph:

篇目 标题 链接
01 LangGraph 是什么? 阅读
02 State 状态管理 阅读
03 Node 节点 阅读
04 Tool Calling 阅读
05 Memory 记忆系统 阅读
06 HITL 人机协作 阅读
07 Multi-Agent 阅读
08 Streaming 流式输出 阅读
09 Conditional Edges 阅读
10 Production 生产部署 阅读

当前位置: 第2篇 / 共10篇

本文是LangGraph零基础入门系列的第二篇,第一篇是《LangGraph零基础入门:你的第一个Agent》。