1. 项目概述从“金鱼脑”到“记忆大师”的进化之路如果你最近在折腾AI Agent尤其是基于LangChain或LangGraph构建的智能体大概率踩过同一个坑你精心设计的Agent在对话中像个“金鱼”说完上句就忘了下句更别提记住用户几天前、甚至几分钟前交代的关键信息了。这感觉就像你雇了个能力超群的助手但他每次见面都像初次见面一样需要你从头到尾再介绍一遍自己。这种体验无疑是令人沮丧的也严重制约了Agent在复杂、长周期任务中的实用性。今天我们就来彻底拆解LangGraph中的记忆管理目标很明确——用30分钟系统性地将你的Agent从“七秒记忆”的金鱼升级为能记住上下文、偏好甚至长期目标的“记忆大师”。LangGraph作为构建有状态、多步骤应用Agent的强大框架其核心魅力就在于对“状态”的精细控制。而记忆本质上就是状态的持久化与演化。很多人对LangGraph的记忆理解停留在“有个内存变量”但实际上从短期的工作记忆Working Memory到基于对话历史的上下文记忆Conversation Memory再到可以长期存储和检索的外部记忆Long-term MemoryLangGraph提供了一整套可插拔、可组合的工具箱。掌握它你就能让Agent真正“活”起来不仅能处理单次查询更能进行连贯的、个性化的多轮交互甚至成为用户的数字伴侣。无论你是想构建一个能记住用户偏好的客服机器人一个能跟踪项目进展的编程助手还是一个能进行深度对话的聊天伙伴记忆管理都是你必须跨越的门槛。2. LangGraph记忆体系的核心架构解析2.1 状态State是记忆的基石在LangGraph中一切记忆的起点和归宿都是“状态”State。你可以把它理解为一个共享的、在图的各个节点Nodes间传递的字典dict。这个字典里存放着当前对话或任务执行过程中的所有信息。一个典型的状态可能包含messages对话消息列表、intermediate_steps中间步骤结果、sender发送者信息等键值对。关键在于LangGraph中的图是“有状态”的。这意味着每次图执行run时都会携带并更新这个状态对象。记忆管理的本质就是如何设计这个状态的结构以及如何在不同次执行之间保存和恢复它。LangGraph没有规定你必须用什么它提供了高度的灵活性。最简单的记忆就是利用状态本身来存储本次运行周期内的信息。例如在一个客服Agent中你可以把用户本次会话中提供的订单号、问题描述都暂存在状态里供后续节点查询使用。注意状态设计是记忆系统的蓝图。一开始就要想清楚哪些信息需要被记忆是仅限本次运行还是需要跨会话持久化。一个常见的错误是把所有东西都塞进状态导致状态臃肿影响性能和清晰度。好的做法是区分“运行时临时变量”和“需要记忆的核心信息”。2.2 记忆的三大层次与实现载体根据记忆的寿命和用途我们可以将其分为三个层次LangGraph为每一层都提供了相应的实现思路和工具。第一层短期/工作记忆Working Memory这相当于Agent的“脑内存”完全由当次运行的State对象承载。它存储的是处理当前任务所必需的即时信息。例如在一个检索增强生成RAG流程中从向量数据库查到的相关文档片段就会作为工作记忆暂存在状态里供LLM生成答案时参考。工作记忆的特点是“随用随弃”图运行结束如果没有特意保存这部分记忆就消失了。它的实现直接依赖于你如何在节点函数中读写State。第二层会话/上下文记忆Conversation Memory这是克服“金鱼脑”的关键。它的目标是让Agent能记住在同一会话可能包含多次图的run调用中的历史对话。LangGraph通过与LangChain的深度集成提供了多种开箱即用的“记忆后端”Memory Backend。最常用的是ConversationBufferMemory和ConversationSummaryMemory。ConversationBufferMemory简单粗暴地把所有历史对话消息都保存下来并在每次调用时作为上下文完整地传递给LLM。优点是信息无损缺点是上下文长度增长很快容易触及模型令牌Token限制。ConversationSummaryMemory更智能的选择。它会定期或按策略让LLM对之前的对话历史进行总结只保留总结摘要和最近的几条原始消息。这样既能保留关键信息又能极大地节省令牌数适合长对话。在LangGraph中你可以将这类Memory对象作为图的一个属性或者在状态中维护一个消息列表并在每次运行时智能地管理其长度如只保留最近N轮。第三层长期记忆Long-term Memory这是让Agent成为“大师”的终极武器。长期记忆的目标是跨会话、甚至永久地记住关于用户或实体的关键信息。例如用户说“我喜欢用Python讨厌Java”或者“我的项目截止日期是下周五”。这类信息不应该在会话结束后就丢弃。实现长期记忆通常需要外部存储如数据库SQLite, PostgreSQL、键值存储Redis或向量数据库Chroma, Weaviate。 其工作流程通常是1在对话中识别需要长期记忆的信息如用户偏好、重要事实2将这些信息以结构化的形式如JSON或嵌入向量Embedding存入外部数据库3在未来的会话中根据当前查询从数据库中检索相关的长期记忆并注入到工作记忆或上下文记忆中。LangGraph本身不捆绑特定的长期记忆实现但它与LangChain的VectorStoreRetrieverMemory等组件无缝结合让你可以轻松构建这套流程。2.3 记忆的流动编码、存储与检索一个完整的记忆系统离不开这三个环节。在LangGraph的上下文中编码决定把什么信息、以什么格式存入记忆。对于上下文记忆编码可能就是原始的对话消息对象。对于长期记忆你可能需要设计一个Memory数据模型包含user_id,memory_type,content,embedding向量,timestamp等字段并使用Pydantic来定义。存储决定存到哪里。上下文记忆可能存储在图的checkpointer中或内存变量里长期记忆则要写入外部数据库。LangGraph的Checkpointer接口非常强大它不仅能保存状态快照还能实现记忆的版本管理和回溯对于复杂Agent的调试和恢复至关重要。检索决定用什么方式把记忆找回来。对于上下文记忆检索就是读取历史消息列表。对于长期记忆则可能涉及相似性搜索通过向量、关键字过滤或基于时间的查询。检索到的记忆需要被巧妙地整合到当前的状态或提示词Prompt中让LLM能够利用它们。3. 实战构建一个拥有三层记忆的智能客服Agent理论说得再多不如动手实现。下面我们一步步构建一个具备完整记忆能力的客服Agent。这个Agent将能1记住当前会话中的问题细节工作记忆2保持连贯的多轮对话上下文记忆3记住用户的偏好和过往工单长期记忆。3.1 环境准备与基础图搭建首先确保你的环境已安装必要库langgraph,langchain-openai,langchain-community。我们将使用OpenAI的GPT模型作为大脑。# 基础导入 from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langchain_community.chat_message_histories import ChatMessageHistory from langchain_core.chat_history import BaseChatMessageHistory # 1. 定义状态State class AgentState(TypedDict): # 当前输入的问题 question: str # 累积的对话消息用于上下文记忆 messages: Annotated[List[BaseMessage], operator.add] # 从长期记忆库检索到的相关信息 retrieved_memories: List[str] # 最终答案 answer: str # 2. 初始化模型和基础组件 llm ChatOpenAI(modelgpt-4o, temperature0) system_prompt 你是一个专业的客服助手请根据用户的提问和已知信息包括对话历史和用户档案进行友好、准确的回答。 # 3. 定义节点函数 def retrieve_memories(state: AgentState): 节点检索长期记忆。这里我们先模拟后续替换为真实数据库查询。 user_id user_123 # 假设从状态或上下文中获取真实用户ID # 模拟从数据库检索到该用户的偏好和过往问题 fake_memories [ 用户偏好喜欢通过邮件接收详细报告不喜欢电话沟通。, 历史工单上周曾反馈过‘登录页面加载慢’的问题已解决。 ] return {retrieved_memories: fake_memories} def generate_answer(state: AgentState): 节点生成回答整合所有记忆。 # 构建提示词整合系统指令、长期记忆、上下文历史和当前问题 context_messages state[messages][-5:] # 取最近5条作为上下文防止过长 memory_context \n.join(state[retrieved_memories]) prompt f {system_prompt} 【关于该用户的长期记忆】 {memory_context} 【最近的对话历史】 {context_messages} 用户当前问题{state[question]} 请生成回答 response llm.invoke(prompt) return {answer: response.content, messages: [AIMessage(contentresponse.content)]} # 4. 构建图 workflow StateGraph(AgentState) workflow.add_node(retrieve, retrieve_memories) workflow.add_node(respond, generate_answer) workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, respond) workflow.add_edge(respond, END) app workflow.compile()这是一个最简单的两节点图检索记忆 - 生成回答。但目前我们的上下文记忆state[“messages”]还只是当次运行有效且长期记忆是模拟的。3.2 集成持久化上下文记忆为了让上下文记忆在多次app.invoke()调用间持续我们需要引入ChatMessageHistory和一个存储区如内存字典来管理不同会话的历史。from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver import asyncio # 使用LangGraph的Checkpointer来持久化整个状态包含messages # 这将自动为我们管理会话级别的状态持久化 memory AsyncSqliteSaver.from_conn_string(:memory:) # 使用内存SQLite生产环境可换为文件或PostgreSQL # 重新编译图并传入checkpointer app_with_memory workflow.compile(checkpointermemory) # 异步调用示例 async def run_conversation(): config {configurable: {thread_id: thread_1}} # thread_id 标识唯一会话 # 第一轮 result1 await app_with_memory.ainvoke( {question: 我的登录页面又打不开了和上周的问题一样吗, messages: []}, configconfig ) print(Answer 1:, result1[answer]) # 第二轮状态包括之前的messages会自动从checkpointer恢复 result2 await app_with_memory.ainvoke( {question: 能把解决方案发我邮箱吗, messages: []}, # 这里传入的messages为空但checkpointer会提供完整历史 configconfig ) print(Answer 2:, result2[answer]) # 检查状态中的消息历史会发现包含了第一轮和第二轮的所有消息 print(Full message history length:, len(result2[messages])) # asyncio.run(run_conversation())通过AsyncSqliteSaver作为checkpointer我们实现了两个目标1整个AgentState包括messages在每次调用后都被自动保存2通过相同的thread_id我们可以恢复特定会话的完整状态。这样Agent在第二轮就能“记得”第一轮的对话内容。checkpointer是LangGraph管理有状态工作流的首选方式它比手动管理ChatMessageHistory更强大和统一。3.3 实现真正的长期记忆存储与检索现在我们来替换掉模拟的长期记忆连接一个真实的向量数据库这里以Chroma为例用于存储和检索用户的长期偏好和事实。from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain_core.documents import Document from pydantic import BaseModel from datetime import datetime # 定义长期记忆的数据模型 class LongTermMemory(BaseModel): user_id: str content: str # 记忆内容 memory_type: str preference # 类型如 preference, fact, ticket embedding: List[float] None # 向量嵌入 timestamp: datetime datetime.now() # 初始化向量存储和嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore Chroma(embedding_functionembeddings, persist_directory./chroma_db) def store_long_term_memory(user_id: str, content: str, memory_type: str): 函数将一条信息存储为长期记忆 memory LongTermMemory(user_iduser_id, contentcontent, memory_typememory_type) # 生成向量 doc Document(page_contentmemory.content, metadata{user_id: user_id, type: memory_type, timestamp: memory.timestamp.isoformat()}) vectorstore.add_documents([doc]) print(f已存储长期记忆{content}) def retrieve_long_term_memories(user_id: str, query: str, k3) - List[str]: 函数根据查询检索相关长期记忆 # 可以添加基于user_id的元数据过滤 docs vectorstore.similarity_search(query, kk, filter{user_id: user_id}) return [doc.page_content for doc in docs] # 修改retrieve_memories节点使其调用真实的检索函数 def retrieve_memories_enhanced(state: AgentState): 增强版记忆检索节点 user_id user_123 current_question state[question] # 1. 检索相关长期记忆 long_term_mems retrieve_long_term_memories(user_id, current_question, k2) # 2. 可选也可以在这里触发记忆存储逻辑。 # 例如如果检测到用户表达了明确的偏好如“以后都发邮件”可以调用store_long_term_memory if 邮箱 in current_question or 邮件 in current_question: # 这里需要更精细的NLP判断此处仅作示例 store_long_term_memory(user_id, 用户明确要求通过邮件接收资料和报告。, preference) return {retrieved_memories: long_term_mems} # 更新图 workflow.update_node(retrieve, retrieve_memories_enhanced) app_enhanced workflow.compile(checkpointermemory)现在我们的Agent具备了真正的长期记忆能力。当用户再次提到“邮箱”时检索到的记忆里可能就包含了之前存储的“喜欢邮件沟通”的偏好从而让回答更具个性化。记忆的存储可以在对话中被动触发如本示例也可以设计一个独立的“记忆提炼”节点在对话结束后分析整个会话提取关键信息存入长期记忆。3.4 记忆的整合与提示工程记忆检索回来之后如何有效地呈现给LLM至关重要。直接堆砌字符串可能效果不佳。我们需要进行提示工程。def generate_answer_with_integrated_memory(state: AgentState): 改进的回答生成节点更好地整合记忆 # 获取最近N条对话作为上下文 recent_dialogue state[messages][-6:] # 最近3轮对话假设每轮一问一答 # 格式化长期记忆 long_term_mem_formatted if state[retrieved_memories]: long_term_mem_formatted 以下是与当前对话相关的用户背景信息\n \n- .join([] state[retrieved_memories]) # 构建结构化的提示词 messages_for_llm [ SystemMessage(contentf{system_prompt} {long_term_mem_formatted} 请参考上述背景信息并基于下面的对话历史进行回复。), *recent_dialogue, # 注入历史消息 HumanMessage(contentstate[question]) ] response llm.invoke(messages_for_llm) return {answer: response.content, messages: [AIMessage(contentresponse.content)]}将系统指令、长期记忆、对话历史分层、清晰地组织在提示词中能帮助LLM更好地理解和利用这些信息。对于更复杂的场景你还可以使用ChatPromptTemplate来更灵活地构建提示。4. 高级记忆模式与优化策略4.1 记忆的摘要与压缩对于超长对话即使有checkpointer将全部历史扔给LLM也是不现实的。我们需要摘要压缩策略。ConversationSummaryMemory如前所述这是LangChain内置的解决方案。你可以在状态中维护一个summary字段每隔几轮对话或当令牌数接近上限时调用LLM对之前的历史生成摘要然后用摘要替代部分旧历史。滑动窗口Sliding Window最简单的方法只保留最近N条消息。我们的示例中state[“messages”][-5:]就是滑动窗口的实现。关键信息提取不摘要整个对话而是只提取关键实体、决策、用户声明等结构化信息存入长期记忆或一个独立的“关键点”列表。这需要更复杂的设计但记忆效率最高。4.2 记忆的更新、遗忘与冲突解决记忆不是只增不减的。用户的偏好会改变事实会被修正。更新为长期记忆设计版本机制或支持更新操作。当检测到用户说“我换邮箱了”应能更新之前存储的邮箱偏好。遗忘可以基于时间自动删除旧记忆、相关性低频使用的记忆降权或主动指令“忘记我刚才说的密码”来实现遗忘。冲突解决当检索到两条矛盾的记忆时如用户曾说“喜欢A”后来说“喜欢B”需要在提示词中让LLM知晓这种冲突或设计规则如“以最新为准”在检索层处理。4.3 基于图的记忆路由LangGraph的图结构允许我们实现更精巧的记忆路由。例如可以设计一个专门的“记忆管理”子图或节点。记忆写入路由根据对话内容决定信息是存入短期状态、会话缓存还是长期数据库。例如普通聊天内容不进长期记忆只有明确的用户声明“我住在北京”才触发长期存储。记忆读取路由根据当前查询的类型决定从哪种记忆库检索。例如处理具体技术问题时优先从向量库检索知识文档处理用户个性化问题时优先从用户长期记忆库检索。# 伪代码示例一个简单的记忆路由节点 def memory_router(state: AgentState): query state[question] memories_to_use [] # 规则1如果问题涉及用户个人情况检索长期记忆 if is_personal_question(query): memories_to_use.extend(retrieve_long_term_memories(state[user_id], query)) # 规则2总是加入最近3轮对话作为上下文 memories_to_use.extend(format_recent_messages(state[messages][-6:])) # 规则3如果问题是关于操作步骤从知识库检索 if is_procedural_question(query): memories_to_use.extend(retrieve_knowledge_base(query)) return {retrieved_memories: memories_to_use}5. 常见问题、调试技巧与性能考量5.1 实战中遇到的典型问题记忆混淆Memory Bleeding不同会话thread_id的记忆串了。确保你的checkpointer配置和retriever的过滤条件正确使用了会话ID或用户ID作为隔离键。令牌超限上下文记忆过长导致API调用失败。必须实施摘要、滑动窗口或选择性载入策略。监控state[“messages”]的总令牌数可用tiktoken库估算。记忆检索不准长期记忆检索返回无关内容。优化向量嵌入模型升级到text-embedding-3-large调整检索的相似度阈值或在元数据中添加更丰富的过滤字段如topic,priority。状态污染某个节点错误地修改了不应修改的状态字段。严格定义State的TypedDict结构并在节点函数中只返回需要更新的字段避免直接修改传入的state。5.2 调试与监控技巧可视化状态流使用LangGraph的内置功能或手动打印在每次图执行后输出state的内容观察记忆是如何被累积和使用的。检查点Checkpoint调试利用checkpointer的list和get方法查看历史状态快照这对于调试复杂、多步骤的Agent工作流尤其有用。记忆检索日志记录每次长期记忆检索的查询词和返回结果分析检索相关性。提示词快照在调用LLM前将构建好的完整提示词记录下来这能帮你直观判断记忆是否被正确整合。5.3 性能与成本优化异步操作对于耗时的记忆存储/检索如向量数据库查询使用异步节点async def和异步检查点保存器避免阻塞整个图执行。缓存策略对频繁读取的长期记忆如用户基本资料进行内存缓存减少数据库查询。分层加载不要一次性加载所有可能相关的记忆。采用“召回-精排”两步走先快速检索出候选记忆如基于关键词再让一个小模型或规则对候选进行精排只将最相关的几条注入上下文。记忆重要性评分为每条长期记忆附加一个重要性权重检索时按权重排序。权重可以根据记忆类型、创建时间、使用频率动态调整。记忆管理是构建强大、实用AI Agent的核心。从简单的状态管理到利用checkpointer实现会话持久化再到连接向量数据库构建长期记忆每一步都在为你的Agent注入更多的“灵魂”。没有一劳永逸的方案最好的记忆系统是根据你的具体应用场景、用户需求和成本预算精心设计和调优出来的。我个人的经验是先从实现一个可靠的、基于checkpointer的上下文记忆开始确保多轮对话的连贯性然后逐步引入长期记忆从存储明确的用户声明开始再扩展到更自动化的信息提炼。在调试时多观察、多记录把记忆的输入输出都可视化出来你会更快地发现逻辑中的问题。最后别忘了性能尤其是在记忆检索和提示词长度方面提前做好规划和监控。 LangGraph记忆管理实战：从状态管理到长期记忆构建
1. 项目概述从“金鱼脑”到“记忆大师”的进化之路如果你最近在折腾AI Agent尤其是基于LangChain或LangGraph构建的智能体大概率踩过同一个坑你精心设计的Agent在对话中像个“金鱼”说完上句就忘了下句更别提记住用户几天前、甚至几分钟前交代的关键信息了。这感觉就像你雇了个能力超群的助手但他每次见面都像初次见面一样需要你从头到尾再介绍一遍自己。这种体验无疑是令人沮丧的也严重制约了Agent在复杂、长周期任务中的实用性。今天我们就来彻底拆解LangGraph中的记忆管理目标很明确——用30分钟系统性地将你的Agent从“七秒记忆”的金鱼升级为能记住上下文、偏好甚至长期目标的“记忆大师”。LangGraph作为构建有状态、多步骤应用Agent的强大框架其核心魅力就在于对“状态”的精细控制。而记忆本质上就是状态的持久化与演化。很多人对LangGraph的记忆理解停留在“有个内存变量”但实际上从短期的工作记忆Working Memory到基于对话历史的上下文记忆Conversation Memory再到可以长期存储和检索的外部记忆Long-term MemoryLangGraph提供了一整套可插拔、可组合的工具箱。掌握它你就能让Agent真正“活”起来不仅能处理单次查询更能进行连贯的、个性化的多轮交互甚至成为用户的数字伴侣。无论你是想构建一个能记住用户偏好的客服机器人一个能跟踪项目进展的编程助手还是一个能进行深度对话的聊天伙伴记忆管理都是你必须跨越的门槛。2. LangGraph记忆体系的核心架构解析2.1 状态State是记忆的基石在LangGraph中一切记忆的起点和归宿都是“状态”State。你可以把它理解为一个共享的、在图的各个节点Nodes间传递的字典dict。这个字典里存放着当前对话或任务执行过程中的所有信息。一个典型的状态可能包含messages对话消息列表、intermediate_steps中间步骤结果、sender发送者信息等键值对。关键在于LangGraph中的图是“有状态”的。这意味着每次图执行run时都会携带并更新这个状态对象。记忆管理的本质就是如何设计这个状态的结构以及如何在不同次执行之间保存和恢复它。LangGraph没有规定你必须用什么它提供了高度的灵活性。最简单的记忆就是利用状态本身来存储本次运行周期内的信息。例如在一个客服Agent中你可以把用户本次会话中提供的订单号、问题描述都暂存在状态里供后续节点查询使用。注意状态设计是记忆系统的蓝图。一开始就要想清楚哪些信息需要被记忆是仅限本次运行还是需要跨会话持久化。一个常见的错误是把所有东西都塞进状态导致状态臃肿影响性能和清晰度。好的做法是区分“运行时临时变量”和“需要记忆的核心信息”。2.2 记忆的三大层次与实现载体根据记忆的寿命和用途我们可以将其分为三个层次LangGraph为每一层都提供了相应的实现思路和工具。第一层短期/工作记忆Working Memory这相当于Agent的“脑内存”完全由当次运行的State对象承载。它存储的是处理当前任务所必需的即时信息。例如在一个检索增强生成RAG流程中从向量数据库查到的相关文档片段就会作为工作记忆暂存在状态里供LLM生成答案时参考。工作记忆的特点是“随用随弃”图运行结束如果没有特意保存这部分记忆就消失了。它的实现直接依赖于你如何在节点函数中读写State。第二层会话/上下文记忆Conversation Memory这是克服“金鱼脑”的关键。它的目标是让Agent能记住在同一会话可能包含多次图的run调用中的历史对话。LangGraph通过与LangChain的深度集成提供了多种开箱即用的“记忆后端”Memory Backend。最常用的是ConversationBufferMemory和ConversationSummaryMemory。ConversationBufferMemory简单粗暴地把所有历史对话消息都保存下来并在每次调用时作为上下文完整地传递给LLM。优点是信息无损缺点是上下文长度增长很快容易触及模型令牌Token限制。ConversationSummaryMemory更智能的选择。它会定期或按策略让LLM对之前的对话历史进行总结只保留总结摘要和最近的几条原始消息。这样既能保留关键信息又能极大地节省令牌数适合长对话。在LangGraph中你可以将这类Memory对象作为图的一个属性或者在状态中维护一个消息列表并在每次运行时智能地管理其长度如只保留最近N轮。第三层长期记忆Long-term Memory这是让Agent成为“大师”的终极武器。长期记忆的目标是跨会话、甚至永久地记住关于用户或实体的关键信息。例如用户说“我喜欢用Python讨厌Java”或者“我的项目截止日期是下周五”。这类信息不应该在会话结束后就丢弃。实现长期记忆通常需要外部存储如数据库SQLite, PostgreSQL、键值存储Redis或向量数据库Chroma, Weaviate。 其工作流程通常是1在对话中识别需要长期记忆的信息如用户偏好、重要事实2将这些信息以结构化的形式如JSON或嵌入向量Embedding存入外部数据库3在未来的会话中根据当前查询从数据库中检索相关的长期记忆并注入到工作记忆或上下文记忆中。LangGraph本身不捆绑特定的长期记忆实现但它与LangChain的VectorStoreRetrieverMemory等组件无缝结合让你可以轻松构建这套流程。2.3 记忆的流动编码、存储与检索一个完整的记忆系统离不开这三个环节。在LangGraph的上下文中编码决定把什么信息、以什么格式存入记忆。对于上下文记忆编码可能就是原始的对话消息对象。对于长期记忆你可能需要设计一个Memory数据模型包含user_id,memory_type,content,embedding向量,timestamp等字段并使用Pydantic来定义。存储决定存到哪里。上下文记忆可能存储在图的checkpointer中或内存变量里长期记忆则要写入外部数据库。LangGraph的Checkpointer接口非常强大它不仅能保存状态快照还能实现记忆的版本管理和回溯对于复杂Agent的调试和恢复至关重要。检索决定用什么方式把记忆找回来。对于上下文记忆检索就是读取历史消息列表。对于长期记忆则可能涉及相似性搜索通过向量、关键字过滤或基于时间的查询。检索到的记忆需要被巧妙地整合到当前的状态或提示词Prompt中让LLM能够利用它们。3. 实战构建一个拥有三层记忆的智能客服Agent理论说得再多不如动手实现。下面我们一步步构建一个具备完整记忆能力的客服Agent。这个Agent将能1记住当前会话中的问题细节工作记忆2保持连贯的多轮对话上下文记忆3记住用户的偏好和过往工单长期记忆。3.1 环境准备与基础图搭建首先确保你的环境已安装必要库langgraph,langchain-openai,langchain-community。我们将使用OpenAI的GPT模型作为大脑。# 基础导入 from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langchain_community.chat_message_histories import ChatMessageHistory from langchain_core.chat_history import BaseChatMessageHistory # 1. 定义状态State class AgentState(TypedDict): # 当前输入的问题 question: str # 累积的对话消息用于上下文记忆 messages: Annotated[List[BaseMessage], operator.add] # 从长期记忆库检索到的相关信息 retrieved_memories: List[str] # 最终答案 answer: str # 2. 初始化模型和基础组件 llm ChatOpenAI(modelgpt-4o, temperature0) system_prompt 你是一个专业的客服助手请根据用户的提问和已知信息包括对话历史和用户档案进行友好、准确的回答。 # 3. 定义节点函数 def retrieve_memories(state: AgentState): 节点检索长期记忆。这里我们先模拟后续替换为真实数据库查询。 user_id user_123 # 假设从状态或上下文中获取真实用户ID # 模拟从数据库检索到该用户的偏好和过往问题 fake_memories [ 用户偏好喜欢通过邮件接收详细报告不喜欢电话沟通。, 历史工单上周曾反馈过‘登录页面加载慢’的问题已解决。 ] return {retrieved_memories: fake_memories} def generate_answer(state: AgentState): 节点生成回答整合所有记忆。 # 构建提示词整合系统指令、长期记忆、上下文历史和当前问题 context_messages state[messages][-5:] # 取最近5条作为上下文防止过长 memory_context \n.join(state[retrieved_memories]) prompt f {system_prompt} 【关于该用户的长期记忆】 {memory_context} 【最近的对话历史】 {context_messages} 用户当前问题{state[question]} 请生成回答 response llm.invoke(prompt) return {answer: response.content, messages: [AIMessage(contentresponse.content)]} # 4. 构建图 workflow StateGraph(AgentState) workflow.add_node(retrieve, retrieve_memories) workflow.add_node(respond, generate_answer) workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, respond) workflow.add_edge(respond, END) app workflow.compile()这是一个最简单的两节点图检索记忆 - 生成回答。但目前我们的上下文记忆state[“messages”]还只是当次运行有效且长期记忆是模拟的。3.2 集成持久化上下文记忆为了让上下文记忆在多次app.invoke()调用间持续我们需要引入ChatMessageHistory和一个存储区如内存字典来管理不同会话的历史。from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver import asyncio # 使用LangGraph的Checkpointer来持久化整个状态包含messages # 这将自动为我们管理会话级别的状态持久化 memory AsyncSqliteSaver.from_conn_string(:memory:) # 使用内存SQLite生产环境可换为文件或PostgreSQL # 重新编译图并传入checkpointer app_with_memory workflow.compile(checkpointermemory) # 异步调用示例 async def run_conversation(): config {configurable: {thread_id: thread_1}} # thread_id 标识唯一会话 # 第一轮 result1 await app_with_memory.ainvoke( {question: 我的登录页面又打不开了和上周的问题一样吗, messages: []}, configconfig ) print(Answer 1:, result1[answer]) # 第二轮状态包括之前的messages会自动从checkpointer恢复 result2 await app_with_memory.ainvoke( {question: 能把解决方案发我邮箱吗, messages: []}, # 这里传入的messages为空但checkpointer会提供完整历史 configconfig ) print(Answer 2:, result2[answer]) # 检查状态中的消息历史会发现包含了第一轮和第二轮的所有消息 print(Full message history length:, len(result2[messages])) # asyncio.run(run_conversation())通过AsyncSqliteSaver作为checkpointer我们实现了两个目标1整个AgentState包括messages在每次调用后都被自动保存2通过相同的thread_id我们可以恢复特定会话的完整状态。这样Agent在第二轮就能“记得”第一轮的对话内容。checkpointer是LangGraph管理有状态工作流的首选方式它比手动管理ChatMessageHistory更强大和统一。3.3 实现真正的长期记忆存储与检索现在我们来替换掉模拟的长期记忆连接一个真实的向量数据库这里以Chroma为例用于存储和检索用户的长期偏好和事实。from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain_core.documents import Document from pydantic import BaseModel from datetime import datetime # 定义长期记忆的数据模型 class LongTermMemory(BaseModel): user_id: str content: str # 记忆内容 memory_type: str preference # 类型如 preference, fact, ticket embedding: List[float] None # 向量嵌入 timestamp: datetime datetime.now() # 初始化向量存储和嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore Chroma(embedding_functionembeddings, persist_directory./chroma_db) def store_long_term_memory(user_id: str, content: str, memory_type: str): 函数将一条信息存储为长期记忆 memory LongTermMemory(user_iduser_id, contentcontent, memory_typememory_type) # 生成向量 doc Document(page_contentmemory.content, metadata{user_id: user_id, type: memory_type, timestamp: memory.timestamp.isoformat()}) vectorstore.add_documents([doc]) print(f已存储长期记忆{content}) def retrieve_long_term_memories(user_id: str, query: str, k3) - List[str]: 函数根据查询检索相关长期记忆 # 可以添加基于user_id的元数据过滤 docs vectorstore.similarity_search(query, kk, filter{user_id: user_id}) return [doc.page_content for doc in docs] # 修改retrieve_memories节点使其调用真实的检索函数 def retrieve_memories_enhanced(state: AgentState): 增强版记忆检索节点 user_id user_123 current_question state[question] # 1. 检索相关长期记忆 long_term_mems retrieve_long_term_memories(user_id, current_question, k2) # 2. 可选也可以在这里触发记忆存储逻辑。 # 例如如果检测到用户表达了明确的偏好如“以后都发邮件”可以调用store_long_term_memory if 邮箱 in current_question or 邮件 in current_question: # 这里需要更精细的NLP判断此处仅作示例 store_long_term_memory(user_id, 用户明确要求通过邮件接收资料和报告。, preference) return {retrieved_memories: long_term_mems} # 更新图 workflow.update_node(retrieve, retrieve_memories_enhanced) app_enhanced workflow.compile(checkpointermemory)现在我们的Agent具备了真正的长期记忆能力。当用户再次提到“邮箱”时检索到的记忆里可能就包含了之前存储的“喜欢邮件沟通”的偏好从而让回答更具个性化。记忆的存储可以在对话中被动触发如本示例也可以设计一个独立的“记忆提炼”节点在对话结束后分析整个会话提取关键信息存入长期记忆。3.4 记忆的整合与提示工程记忆检索回来之后如何有效地呈现给LLM至关重要。直接堆砌字符串可能效果不佳。我们需要进行提示工程。def generate_answer_with_integrated_memory(state: AgentState): 改进的回答生成节点更好地整合记忆 # 获取最近N条对话作为上下文 recent_dialogue state[messages][-6:] # 最近3轮对话假设每轮一问一答 # 格式化长期记忆 long_term_mem_formatted if state[retrieved_memories]: long_term_mem_formatted 以下是与当前对话相关的用户背景信息\n \n- .join([] state[retrieved_memories]) # 构建结构化的提示词 messages_for_llm [ SystemMessage(contentf{system_prompt} {long_term_mem_formatted} 请参考上述背景信息并基于下面的对话历史进行回复。), *recent_dialogue, # 注入历史消息 HumanMessage(contentstate[question]) ] response llm.invoke(messages_for_llm) return {answer: response.content, messages: [AIMessage(contentresponse.content)]}将系统指令、长期记忆、对话历史分层、清晰地组织在提示词中能帮助LLM更好地理解和利用这些信息。对于更复杂的场景你还可以使用ChatPromptTemplate来更灵活地构建提示。4. 高级记忆模式与优化策略4.1 记忆的摘要与压缩对于超长对话即使有checkpointer将全部历史扔给LLM也是不现实的。我们需要摘要压缩策略。ConversationSummaryMemory如前所述这是LangChain内置的解决方案。你可以在状态中维护一个summary字段每隔几轮对话或当令牌数接近上限时调用LLM对之前的历史生成摘要然后用摘要替代部分旧历史。滑动窗口Sliding Window最简单的方法只保留最近N条消息。我们的示例中state[“messages”][-5:]就是滑动窗口的实现。关键信息提取不摘要整个对话而是只提取关键实体、决策、用户声明等结构化信息存入长期记忆或一个独立的“关键点”列表。这需要更复杂的设计但记忆效率最高。4.2 记忆的更新、遗忘与冲突解决记忆不是只增不减的。用户的偏好会改变事实会被修正。更新为长期记忆设计版本机制或支持更新操作。当检测到用户说“我换邮箱了”应能更新之前存储的邮箱偏好。遗忘可以基于时间自动删除旧记忆、相关性低频使用的记忆降权或主动指令“忘记我刚才说的密码”来实现遗忘。冲突解决当检索到两条矛盾的记忆时如用户曾说“喜欢A”后来说“喜欢B”需要在提示词中让LLM知晓这种冲突或设计规则如“以最新为准”在检索层处理。4.3 基于图的记忆路由LangGraph的图结构允许我们实现更精巧的记忆路由。例如可以设计一个专门的“记忆管理”子图或节点。记忆写入路由根据对话内容决定信息是存入短期状态、会话缓存还是长期数据库。例如普通聊天内容不进长期记忆只有明确的用户声明“我住在北京”才触发长期存储。记忆读取路由根据当前查询的类型决定从哪种记忆库检索。例如处理具体技术问题时优先从向量库检索知识文档处理用户个性化问题时优先从用户长期记忆库检索。# 伪代码示例一个简单的记忆路由节点 def memory_router(state: AgentState): query state[question] memories_to_use [] # 规则1如果问题涉及用户个人情况检索长期记忆 if is_personal_question(query): memories_to_use.extend(retrieve_long_term_memories(state[user_id], query)) # 规则2总是加入最近3轮对话作为上下文 memories_to_use.extend(format_recent_messages(state[messages][-6:])) # 规则3如果问题是关于操作步骤从知识库检索 if is_procedural_question(query): memories_to_use.extend(retrieve_knowledge_base(query)) return {retrieved_memories: memories_to_use}5. 常见问题、调试技巧与性能考量5.1 实战中遇到的典型问题记忆混淆Memory Bleeding不同会话thread_id的记忆串了。确保你的checkpointer配置和retriever的过滤条件正确使用了会话ID或用户ID作为隔离键。令牌超限上下文记忆过长导致API调用失败。必须实施摘要、滑动窗口或选择性载入策略。监控state[“messages”]的总令牌数可用tiktoken库估算。记忆检索不准长期记忆检索返回无关内容。优化向量嵌入模型升级到text-embedding-3-large调整检索的相似度阈值或在元数据中添加更丰富的过滤字段如topic,priority。状态污染某个节点错误地修改了不应修改的状态字段。严格定义State的TypedDict结构并在节点函数中只返回需要更新的字段避免直接修改传入的state。5.2 调试与监控技巧可视化状态流使用LangGraph的内置功能或手动打印在每次图执行后输出state的内容观察记忆是如何被累积和使用的。检查点Checkpoint调试利用checkpointer的list和get方法查看历史状态快照这对于调试复杂、多步骤的Agent工作流尤其有用。记忆检索日志记录每次长期记忆检索的查询词和返回结果分析检索相关性。提示词快照在调用LLM前将构建好的完整提示词记录下来这能帮你直观判断记忆是否被正确整合。5.3 性能与成本优化异步操作对于耗时的记忆存储/检索如向量数据库查询使用异步节点async def和异步检查点保存器避免阻塞整个图执行。缓存策略对频繁读取的长期记忆如用户基本资料进行内存缓存减少数据库查询。分层加载不要一次性加载所有可能相关的记忆。采用“召回-精排”两步走先快速检索出候选记忆如基于关键词再让一个小模型或规则对候选进行精排只将最相关的几条注入上下文。记忆重要性评分为每条长期记忆附加一个重要性权重检索时按权重排序。权重可以根据记忆类型、创建时间、使用频率动态调整。记忆管理是构建强大、实用AI Agent的核心。从简单的状态管理到利用checkpointer实现会话持久化再到连接向量数据库构建长期记忆每一步都在为你的Agent注入更多的“灵魂”。没有一劳永逸的方案最好的记忆系统是根据你的具体应用场景、用户需求和成本预算精心设计和调优出来的。我个人的经验是先从实现一个可靠的、基于checkpointer的上下文记忆开始确保多轮对话的连贯性然后逐步引入长期记忆从存储明确的用户声明开始再扩展到更自动化的信息提炼。在调试时多观察、多记录把记忆的输入输出都可视化出来你会更快地发现逻辑中的问题。最后别忘了性能尤其是在记忆检索和提示词长度方面提前做好规划和监控。