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 上面的例子虽然展示了状态的概念,但太过简单。在实际应用中,我们需要:
更复杂的状态结构 ——不只是对话历史,可能还有用户偏好、中间计算结果、错误状态等
状态更新规则 ——什么时候覆盖、什么时候追加、什么时候清空
多步骤流程 ——一个Agent可能要经历”理解→规划→执行→验证”多个阶段
可视化调试 ——能看到状态是如何一步步变化的
LangGraph 就是为解决这些问题而生的。
核心概念:TypedDict vs Pydantic,reducer概念 在深入代码之前,我们需要理解LangGraph中状态管理的三个核心概念。
概念1:State(状态)是什么? 在LangGraph中,State就是一个Python类,用来定义你的Agent需要记住的所有信息 。
打个比方:State就像是Agent的”记忆笔记本”,每一页记录不同类型的信息。有的页面记对话历史,有的记用户偏好,有的记当前任务进度…
LangGraph提供了两种方式来定义State:TypedDict 和Pydantic 。
概念2:TypedDict方式 TypedDict是Python标准库的一部分(Python 3.8+),它让我们可以用类似字典的方式定义结构化数据。
1 2 3 4 5 6 7 from typing import TypedDict, List , Annotatedclass 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, Fieldfrom typing import List 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中最精妙的设计之一。
问题 :当多个节点都想修改同一个状态时,应该怎么做?
简单方案(有问题) :直接覆盖
更好的方案 :告诉LangGraph如何”合并”状态更新
这就是Reducer 的作用——它定义了当新值来临时,应该如何与旧值合并。
1 2 3 4 5 6 7 8 9 10 from typing import Annotatedfrom operator import addclass ChatState (TypedDict ): messages: Annotated[List [str ], add] count: int
内置的reducer:
operator.add —— 用于列表,表示追加
1 2 3 old = [1 , 2 ] new = [3 , 4 ] result = [1 , 2 , 3 , 4 ]
自定义reducer函数 —— 你可以写任意逻辑
1 2 3 4 5 def keep_max (existing, new ): return max (existing, new) class State (TypedDict ): score: Annotated[int , keep_max]
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 osfrom typing import TypedDict, List , Annotated, Literal from operator import addfrom datetime import datetimefrom dotenv import load_dotenvfrom langgraph.graph import StateGraph, ENDfrom langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage, AIMessage, SystemMessageload_dotenv() class ChatState (TypedDict ): """ 定义聊天机器人的状态结构 每个字段代表Agent需要记忆的一种信息 """ messages: Annotated[List [dict ], add] user_name: str turn_count: int current_intent: str emotion: str 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 "" 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_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" ])) 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希望很快再见到你!👋" 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" ] if intent == "farewell" : return "exit" if turn_count >= 20 : return "exit" return "continue" 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 () 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 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 if state["messages" ]: last_msg = state["messages" ][-1 ] if last_msg["role" ] == "assistant" : print (f"🤖 AI: {last_msg['content' ]} " ) if state["current_intent" ] == "farewell" : print ("\n💡 提示:检测到告别意图,下次输入将结束对话。" ) 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__" : run_with_streaming()
步骤3:配置API密钥 创建 .env 文件:
1 OPENAI_API_KEY=your_openai_api_key_here
步骤4:运行
你会看到流式执行的结果,展示状态是如何一步步变化的。
代码要点解析
状态定义 :
1 2 3 4 class ChatState (TypedDict ): messages: Annotated[List [dict ], add] user_name: str turn_count: int
节点函数 :每个节点接收当前状态,返回状态更新
1 2 3 def intent_classifier (state: ChatState ) -> ChatState: return {"current_intent" : intent, "emotion" : emotion}
流式执行 :可以观察每一步的状态变化
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 def step_a (state: State ): return {"current_step" : "processing" } def step_b (state: State ): return {"current_step" : "completed" }
模式2:追加(Append) 使用Annotated和add实现,适用于需要保留历史的字段。
适用场景:
1 2 3 4 5 6 from typing import Annotatedfrom operator import addclass State (TypedDict ): messages: Annotated[List [dict ], add] logs: Annotated[List [str ], add]
示例:
1 2 3 4 5 6 7 8 9 10 11 12 def node_a (state: State ): return {"messages" : [{"role" : "ai" , "content" : "你好" }]} def node_b (state: State ): return {"messages" : [{"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]
示例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:] 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 from langgraph.graph import StateGraphprint (app.get_graph().draw_mermaid())
把输出的Mermaid语法粘贴到 Mermaid Live Editor 就能看到漂亮的流程图。
技巧5:使用LangSmith追踪(生产环境) 1 2 3 4 5 6 7 8 9 import osos.environ["LANGCHAIN_TRACING_V2" ] = "true" os.environ["LANGCHAIN_API_KEY" ] = "your_langsmith_api_key" os.environ["LANGCHAIN_PROJECT" ] = "my-chatbot" result = app.invoke(state)
在LangSmith界面上,你可以看到:
完整的执行轨迹
每个节点的输入输出
状态变化的时序图
性能分析
总结与下篇预告 今天学到了什么?
为什么需要状态管理 ——让Agent拥有”记忆”,支持上下文对话
TypedDict vs Pydantic ——两种定义状态的方式,根据需求选择
Reducer概念 ——控制状态如何更新,避免数据丢失
完整实战 ——构建了一个带历史记录的对话机器人
调试技巧 ——多种方法观察状态变化
关键要点回顾 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学会”思考”》
我们会学习:
如何根据状态动态决定下一步
构建更复杂的工作流(循环、分支、并行)
实际案例:智能客服路由系统
敬请期待!
练习与思考
动手练习 :修改示例代码,添加一个”记忆用户喜好”的功能(如记住用户喜欢的颜色)
思考题 :在你的项目中,哪些数据适合用”追加”模式,哪些适合”覆盖”模式?
进阶挑战 :尝试用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》。