LangGraph智能体记忆管理与多轮对话实战

LangGraph智能体记忆管理与多轮对话实战 LangGraph智能体记忆管理与多轮对话实战写在前面做AI Agent开发时多轮对话的记忆管理是个绕不开的问题。本文从短期记忆、长期记忆、消息裁剪、消息总结四个维度结合LangGraph的底层API手把手带你实现一个记得住事的智能体。一、记忆管理的四种方案在开始写代码之前先理清楚LangGraph中处理记忆的几种思路方案载体适用场景优缺点短期记忆CheckpointerMemorySaver/RedisSaver同一线程内的连续对话简单直接但受上下文窗口限制长期记忆BaseStoreInMemoryStore/RedisStore跨线程、跨会话的用户画像支持向量检索但需要手动管理消息裁剪trim_messages快速丢弃旧消息简单可控但会丢失信息消息总结LLM生成摘要长对话压缩保留关键信息但依赖摘要质量二、短期记忆让Agent记住刚才聊了啥短期记忆是最基础的用Checkpointer就能搞定。核心思路是同一个thread_id就是同一个会话。2.1 预构建Agent方式最简单的方式几行代码就能跑起来importdotenvfromlanggraph.checkpoint.memoryimportInMemorySaverfromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelimportos dotenv.load_dotenv()# 初始化模型llminit_chat_model(deepseek-chat,model_providerdeepseek,api_keyos.getenv(DEEPSEEK_API_KEY))# 关键传入checkpointercheckpointerInMemorySaver()agentcreate_agent(modelllm,tools[],checkpointercheckpointer)# 同一个thread_id 同一个会话config{configurable:{thread_id:user-001}}# 第一轮自我介绍msg1agent.invoke({messages:[(user,你好我叫二狗喜欢学习。)]},config)msg1[messages][-1].pretty_print()# 第二轮Agent能记住你说的话msg2agent.invoke({messages:[(user,我叫什么我喜欢做什么)]},config)msg2[messages][-1].pretty_print()运行结果# 第一轮回复 你好二狗很高兴认识你 # 第二轮回复 - Agent记住了 根据我们刚刚的对话—— **你叫**二狗 **你喜欢**学习2.2 底层API方式如果需要更灵活的控制可以用StateGraph自己搭建fromtypingimportTypedDict,Annotatedfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphfromlanggraph.graph.messageimportadd_messagesfromlangchain.chat_modelsimportinit_chat_modelimportdotenv dotenv.load_dotenv(overrideTrue)# 定义状态结构classState(TypedDict):messages:Annotated[list,add_messages]# 构建图graph_builderStateGraph(State)llminit_chat_model(deepseek-chat,model_providerdeepseek,api_keyos.getenv(DEEPSEEK_API_KEY))defchatbot(state:State):return{messages:[llm.invoke(state[messages])]}graph_builder.add_node(chatbot,chatbot)graph_builder.add_edge(START,chatbot)graph_builder.add_edge(chatbot,END)# 传入checkpointermemoryMemorySaver()graphgraph_builder.compile(checkpointermemory)# 多轮对话config{configurable:{thread_id:chat-1}}msg1graph.invoke({messages:[你好我叫二狗喜欢学习。]},configconfig)msg1[messages][-1].pretty_print()msg2graph.invoke({messages:[我叫什么我喜欢做什么]},configconfig)msg2[messages][-1].pretty_print()输出# 第一轮 你好二狗很高兴认识你爱学习的你听起来就是个有趣的灵魂。 # 第二轮 - 记住了 你叫二狗喜欢学习 看来我得记牢这个学霸认证啦三、长期记忆跨会话也能认出你短期记忆有个局限换了个thread_id就失忆了。长期记忆通过BaseStore实现跨线程记忆用户换个会话Agent也能认出他。核心实现importuuidfromtypingimportTypedDict,Annotatedfromlangchain_core.runnablesimportRunnableConfigfromlanggraph.constantsimportEND,STARTfromlanggraph.graphimportStateGraph,MessagesState,add_messagesfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.store.memoryimportInMemoryStorefromlanggraph.store.baseimportBaseStorefromlangchain.chat_modelsimportinit_chat_modelimportdotenv dotenv.load_dotenv()modelinit_chat_model(deepseek-chat,model_providerdeepseek,api_keyos.getenv(DEEPSEEK_API_KEY))classState(TypedDict):messages:Annotated[list,add_messages]defsave_memory(store:BaseStore,user_id:str,content:str):保存记忆到Storenamespace(memories,user_id)store.put(namespace,str(uuid.uuid4()),{data:content})defrecall_memories(store:BaseStore,user_id:str,query:str,limit:int5):从Store中检索相关记忆namespace(memories,user_id)memoriesstore.search(namespace,queryquery,limitlimit)return[m.value[data]forminmemories]defchatbot(state:MessagesState,config:RunnableConfig,*,store:BaseStore):user_idconfig[configurable][user_id]# 检索历史记忆querystate[messages][-1].content related_memoriesrecall_memories(store,user_id,query)# 把记忆注入系统提示system_msg(你是一个友好的聊天助手。\nf以下是关于用户的记忆:\n{chr(10).join(related_memories)ifrelated_memorieselse暂无})# 保存当前消息到记忆save_memory(store,user_id,query)responsemodel.invoke([{role:system,content:system_msg}]state[messages])return{messages:response}# 构建图builderStateGraph(State)builder.add_node(chatbot)builder.add_edge(START,chatbot)builder.add_edge(chatbot,END)# 同时使用Checkpointer和StorecheckpointerInMemorySaver()storeInMemoryStore()graphbuilder.compile(checkpointercheckpointer,storestore)# 测试两个不同的thread但同一个user_idconfig1{configurable:{thread_id:1,user_id:1}}msg1graph.invoke({messages:[{role:user,content:我叫钢蛋喜欢学习。}]},config1)print(第一次回复)msg1[messages][-1].pretty_print()# 换了thread_id但user_id相同Agent还能记得config2{configurable:{thread_id:2,user_id:1}}msg2graph.invoke({messages:[{role:user,content:我叫什么我喜欢做什么}]},config2)print(第二次回复)msg2[messages][-1].pretty_print()输出第一次回复 你好崔亮很高兴认识你。你提到喜欢学习这真是一个很棒的兴趣 第二次回复 你叫钢蛋你喜欢学习。关键点虽然thread_id从1变成了2但因为user_id相同Agent还是记住了用户信息。四、消息裁剪简单粗暴的上下文控制当对话太长时最直接的办法就是砍掉旧消息。LangGraph提供了trim_messages函数来实现这个功能。fromlangchain_core.messages.utilsimporttrim_messages,count_tokens_approximatelyfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.prebuiltimportcreate_react_agentfromlangchain.chat_modelsimportinit_chat_modelimportdotenv dotenv.load_dotenv()modelinit_chat_model(deepseek-chat,model_providerdeepseek,api_keyos.getenv(DEEPSEEK_API_KEY))defpre_model_hook(state): 钩子函数在调用模型前裁剪消息 trimmed_messagestrim_messages(state[messages],strategylast,# 保留最新的token_countercount_tokens_approximately,max_tokens300,# 最多300个tokenstart_onhuman,# 从人类消息开始裁剪end_on(human,tool),# 裁剪到人类消息或工具消息)return{llm_input_messages:trimmed_messages}checkpointerInMemorySaver()agentcreate_react_agent(model,tools[],pre_model_hookpre_model_hook,checkpointercheckpointer,)config{configurable:{thread_id:user-001}}# 第一轮msg1agent.invoke({messages:[(user,你好我叫迪迦)]},config)msg1[messages][-1].pretty_print()# 发送多轮消息like_list[唱,跳,rap,篮球]forlikeinlike_list:agent.invoke({messages:[(user,f我喜欢做的事是{like})]},config)# 查询记忆 - 由于裁剪可能丢失早期信息msg2agent.invoke({messages:[(user,我叫什么我喜欢做的事是什么)]},config)msg2[messages][-1].pretty_print()裁剪策略说明参数说明strategylast保留最新的消息max_tokens300限制总token数start_onhuman裁剪时从人类消息开始end_on(human, tool)裁剪到人类消息或工具消息为止优缺点✅ 简单直接保证上下文不超限❌ 容易丢失重要信息比如早期介绍的名字五、消息总结智能压缩历史消息总结比裁剪更聪明它用LLM把旧消息压缩成摘要既控制了token数又保留了关键信息。核心思路设定阈值比如消息数 6超过阈值时把旧消息喂给LLM生成摘要保留摘要 最近几条消息完整实现importosfromtypingimportTypedDict,Annotatedfromdotenvimportload_dotenvfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportSystemMessagefromlangchain.chat_modelsimportinit_chat_model load_dotenv()modelinit_chat_model(deepseek-chat,model_providerdeepseek,api_keyos.getenv(DEEPSEEK_API_KEY))classState(TypedDict):messages:Annotated[list,add_messages]defcreate_summary(model,messages):用LLM生成对话摘要summary_prompt[SystemMessage(content请将以下对话历史压缩成一段简短的摘要保留关键信息如用户姓名、偏好等。),*messages]responsemodel.invoke(summary_prompt)returnresponse.content# 阈值消息数超过6条触发总结summary_threshold6defsummary_node(state:State):messagesstate[messages]msg_countlen(messages)print(f\n[总结节点] 当前消息数:{msg_count})ifmsg_countsummary_threshold:print(f触发总结消息数{msg_count} 阈值{summary_threshold})# 保留最近2条其余总结recent_messagesmessages[-2:]old_messagesmessages[:-2]# 生成摘要summary_contentcreate_summary(model,old_messages)print(f摘要:{summary_content[:100]}...)# 构建新消息列表summarized_messages[SystemMessage(contentf【历史对话摘要】{summary_content}),*recent_messages]print(f消息数变化:{msg_count}-{len(summarized_messages)})return{messages:summarized_messages}else:print(f未达到阈值继续累积)returnstatedefchat_node(state:State):messagesstate[messages]ifnotany(isinstance(m,SystemMessage)forminmessages):messages[SystemMessage(content你是一个友好的AI助手。)]messages responsemodel.invoke(messages)return{messages:[response]}defshould_summarize(state:State):msg_countlen(state[messages])returnsummaryifmsg_countsummary_thresholdelsechat# 构建图graph_builderStateGraph(State)graph_builder.add_node(summary,summary_node)graph_builder.add_node(chat,chat_node)graph_builder.add_edge(START,summary)graph_builder.add_conditional_edges(summary,should_summarize,[summary,chat])graph_builder.add_edge(chat,END)memoryInMemorySaver()graphgraph_builder.compile(checkpointermemory)# 测试config{configurable:{thread_id:user-003}}# 第一轮graph.invoke({messages:[(user,你好我叫小蠡喜欢学习。)]},config)# 多轮对话触发总结like_list[唱,跳,rap,篮球,看电影,打游戏]forlikeinlike_list:graph.invoke({messages:[(user,f我喜欢做的事是{like})]},config)# 查询记忆 - 总结后是否还记得msg_finalgraph.invoke({messages:[(user,我叫什么我喜欢做的事是什么)]},config)print(\n最终回复:)print(msg_final[messages][-1].content)运行输出[总结节点] 当前消息数: 3 未达到阈值继续累积 [总结节点] 当前消息数: 5 未达到阈值继续累积 [总结节点] 当前消息数: 7 触发总结消息数 7 阈值 6 摘要: 用户叫小蠡喜欢学习还喜欢唱、跳、rap、篮球... 消息数变化: 7 - 3 最终回复: 你叫小蠡喜欢学习喜欢的活动包括唱、跳、rap、篮球...六、总结对比方案实现复杂度信息保留度适用场景短期记忆⭐⭐⭐⭐简单多轮对话长期记忆⭐⭐⭐⭐⭐⭐用户画像、个性化服务消息裁剪⭐⭐快速原型、token敏感场景消息总结⭐⭐⭐⭐⭐⭐长对话、复杂任务实际项目建议短期对话为主 → 用Checkpointer就够了需要跨会话记忆 → 加上BaseStore对话很长 → 消息总结 短期记忆配合使用Token预算紧张 → 消息裁剪兜底七、踩坑记录Checkpointer和Store的区别- Checkpointer保存的是图的运行状态短期记忆Store保存的是用户数据长期记忆两者要配合使用消息裁剪的边界-start_onhuman和end_on参数很重要否则可能裁剪掉不完整的消息对总结阈值的设置- 太小会频繁总结影响性能太大又起不到压缩作用建议根据实际对话长度调整Store的命名空间- 用(memories, user_id)这样的结构可以实现用户隔离避免数据串线如果觉得有帮助点个赞再走呗