ARTICLE DETAIL

资讯详情

深耕网站视觉设计与运营推广的一线实战洞察。

AI Agent记忆架构实战:LangGraph分层记忆与存储选型指南

AI Agent记忆架构实战:LangGraph分层记忆与存储选型指南 你大概率遇到过这种情况头一天还在跟某个AI助手详细聊你负责的项目背景、技术栈和团队分工第二天再来问它一个跟进问题它一脸茫然仿佛你是第一次来。这其实就是“无记忆Agent”的通病。我前两篇讲了Agent的整体框架和编排方式本篇专门聊记忆怎么让Agent记住你并且记住的是“有用的事”而不是一股脑把所有对话流水账全存下来。这一篇不堆概念会直接给出一套可以落到代码里的记忆架构方案结合LangGraph的实现思路覆盖短期记忆、长期记忆、工作记忆三个层次以及存储选型、检索策略、遗忘机制这些工程上绕不开的问题。做AI Agent开发的、做AI产品设计的朋友看完应该能把“记住用户”这件事从玄学变成明确的技术方案。1. “记住你”这件事为什么比想象中难得多先说一个反直觉的结论让Agent记住用户难点根本不在存储上而在“该记什么”和“该忘什么”这两个问题上。存储顶多算个工程活但“从一段混乱对话里识别出哪些信息值得永久保存、哪些只是临时上下文”这件事直到今天也没有完美的自动解法。1.1 大模型本身只拥有“一次性记忆”很多人第一次接触大模型API时会有个错觉模型是不是自带记忆当然不是。GPT、Claude、通义千问这些模型本身是“无状态”的。你用API发一段对话过去它根据这段对话生成回复之后这段对话在模型侧就“消失”了。下一次调用模型面对的是一个全新的开始。之所以你会觉得某些聊天工具有记忆是因为产品层在偷偷做拼接——把你和它的历史对话、用户画像、知识库摘要一起塞进Prompt里再发给模型。所以“Agent记住你”的本质不是模型能力而是工程能力的体现谁来存、存什么、怎么取、什么时候取这些全都需要你亲自设计。这个前提一定要先想清楚否则后面设计记忆系统时容易被误导成“多开点context窗口就能解决”。1.2 想象中的记忆 vs 现实中的记忆做一个对比你会更清楚目标是什么维度想象中理想状态现实中要做到的状态记住范围几个月甚至永久的对话都能记住关键事实长期保留琐碎细节自动淡化获取速度每次回复前把所有记忆全部扫描一遍按需检索只把相关记忆注入当前Prompt一致性永远不会记错、不会矛盾允许有过期信息但要有纠错和更新机制成本无所谓全量塞给模型严格控制Token消耗不能无限制堆积跨会话换设备、换对话窗口都能记住要有全局存储层而不是局部变量这不是给你泼冷水而是提前帮你校准目标。如果一开始就把“记忆”定义为“全量对话永不遗忘”那后面做的所有设计都会跑偏。现实工程里好的记忆系统最重要的能力是一个字挑。挑出该记的、挑出该取的、挑出该忘的。1.3 记忆失效的常见形态从我测试过的开源Agent项目来看记忆问题翻车通常有四种形态第一种是“该记的没记”。用户明确说“我住在杭州平时通勤用地铁”结果下次问生活建议时系统完全没利用这个信息。原因多半是记忆写入逻辑太粗糙只把整段对话塞进向量库检索时又没召回这段内容。第二种是“不该记的乱记”。用户随口说了句“今天天气好热”系统就把这句话当成饮食偏好存进长期记忆之后每次推荐菜系都往清淡方向跑。这就是缺少记忆筛选策略的典型症状。第三种是“记忆冲突没人管”。用户第一次说“我养了一只猫”过了三个月说“我家猫去世了”。旧的记忆还在检索结果里模型同时看到两者回复就会非常拧巴。第四种是“上下文爆炸”。把过去所有对话全部拼进Prompt导致Token开销巨大、模型注意力被稀释反而回答质量直线下降。这四种问题在我下面要讲的方案里都有对应的处理手段。不卖关子直接进核心。2. 记忆架构怎么分层不是把所有会话记录堆在一起单靠一个向量数据库解决不了记忆问题这是我做多个Agent项目后最深的体会。原因很简单不同记忆的生命周期完全不同访问频率也完全不同。把三个月前的偏好和当前对话里的临时状态放在同一个存储方案里一定会顾此失彼。所以工程上更稳妥的方案是分层记忆架构。这里我把它拆成三层外加一个服务层。2.1 短期记忆上下文窗口内的“工作台”短期记忆指的就是当前会话里模型上下文窗口中的那部分内容。它解决的问题是同一轮对话里你说了前一句模型得知道后一句该怎么接才不跑题。配置应该是这样的存储形式不走外部存储直接使用对话消息数组。生命周期一个会话结束或超过窗口长度后就被截断或压缩。管理策略窗口长度不够时做摘要压缩或移除最早的消息。这里有一个工程细节容易被忽略短期记忆不一定要“完整保存”。假设用户和Agent聊了三个小时窗口早超了这时候你要是直接把最早的消息丢掉前面确认过的信息就白聊了。更好的做法是在上下文快满时触发一次总结节点让模型把当前对话里已经确认的事实、用户表达过的偏好提取成摘要然后用摘要替换掉旧消息继续给新的对话腾空间。LangGraph里实现这个逻辑很顺。核心是给对话加一个“状态管理”节点在每次调用模型之前判断当前消息数组的长度超过阈值就先跑一次summarize。代码如下逻辑并不复杂def maybe_summarize(state): messages state[messages] if num_tokens(messages) 6000: return {messages: messages} summary llm.invoke( 请将以下对话压缩为简洁摘要 保留所有事实性信息和用户偏好。 f对话内容{messages} ) system_msg {role: system, content: f以下是之前对话的摘要{summary}} return {messages: [system_msg, messages[-4:]]}我建议压缩的触发阈值不要上得太激进。比如窗口上限是8000 token那6000 token左右就该做压缩了不要让真实请求触顶。留出缓冲区域否则摘要节点本身也要消耗Token很容易在业务高峰期压爆成本。2.2 长期记忆跨会话持久化的“笔记本”长期记忆解决的是“下次聊天你还记得我”的问题。它的特征是写入频率低、读取频率低、信息价值高、生命周期长。什么内容适合进长期记忆呢我列一个接地气的清单用户的基础画像职业、城市、家庭情况、作息习惯用户表达过的稳定偏好口味、沟通风格、对某些话题的态度用户主动要求记住的事项会议时间、截止日期、给过的重要背景项目相关的长期事实正在做的事情、阶段目标、技术栈、约束条件长期记忆的存储介质通常是向量数据库或者“键值存储向量检索”的结合。后面第四章细讲选型这里先聚焦架构。LangGraph里长期记忆的标准接入方式是把向量存储封装成一个检索器节点在Agent开始工作前先执行一步retrieve把跟当前问题相关的记忆注入System Prompt。2.3 工作记忆当前任务状态下的“便利贴”工作记忆这个提法在Agent开发里相对少但它很关键。它的含义是当前正在处理的这个任务里哪些中间状态是不能丢的。举一个具体的例子。假设你让Agent写一篇行业分析报告。报告分三步收集素材、列大纲、写正文。第一步收集素材的阶段Agent已经确认了三个核心观点这些观点在第一阶段结束时若不做持久化第二阶段开始就丢了——虽然从对话记录里还能翻到但那是纯文本模型没法高效利用。工作记忆的工程实现方式通常是给Agent定义一份结构化的状态schema里面专门放当前任务的关键中间产物。在LangGraph里这就是StateGraph的State。它的生命周期短但状态结构必须是结构化的不能只是消息列表。我的实践经验是一定要给工作记忆定义明确的字段比如“已完成步骤”“当前假设”“待确认问题”否则模型很容易在该落状态时只是随口聊聊最后什么都没记下来。2.4 三者的协作关系与流转三层记忆不是各管各的它们之间应该有明确的流转路径对话产生的内容先进短期记忆上下文窗口。上下文快满时旧内容做摘要压缩压缩出的“关键事实”候选进入长期记忆写入队列。当前任务涉及的临时状态放工作记忆任务结束后有价值的部分沉淀进长期记忆。每次新会话启动时先从长期记忆检索与当前问题相关的部分注入短期记忆上下文。这个流转顺序你可以理解为工作记忆是大脑当前正在处理的任务缓存短期记忆是桌面上的文件长期记忆是档案室。每次开工前先去档案室查资料再放到桌面上这就是一次标准的知识激活过程。3. 落地实操用LangGraph搭一个能记住用户的Agent前面原理讲了半天这一章直接动手。我把这套记忆架构落到LangGraph代码上做一个“用户偏好记忆Agent”功能设定如下用户首次告知偏好后Agent能存入长期记忆用户下次来问时Agent能自动回忆起相关偏好用户主动更正偏好时旧信息能被替换整个记忆过程不依赖外部服务便于本地跑通3.1 记忆状态的结构设计LangGraph的核心是状态图。我们首先定义记忆Agent的状态结构。重点是不能只存messages还必须给记忆操作留出接口字段。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] # 当前问题对应的相关记忆 relevant_memories: list # 本轮对话里要写入长期记忆的候选 memory_candidates: list # 用户主动要求修正的记忆 memory_updates: list四个字段里messages是常规对话流relevant_memories是检索结果memory_candidates和memory_updates是记忆写入和更新动作的载体。为什么单独搞两个“动作字段”因为LangGraph的节点是执行单元节点之间要通信通过state传参是最自然的方式。记忆节点做的判断结果后续节点直接用不用重复调用模型。3.2 记忆检索节点怎么把“相关记忆”找出来检索节点的职责是根据用户当前这轮问题从长期记忆库里召回最相关的记忆片段。我用一种最接地气的方式实现先用关键词和向量双重召回再做一次重排。完整代码如下import chromadb from chromadb.utils import embedding_functions client chromadb.PersistentClient(path./memory_db) col client.get_or_create_collection( user_preferences, embedding_functionembedding_functions.DefaultEmbeddingFunction() ) def retrieve_memories(state: AgentState) - AgentState: if not state[messages]: return state query state[messages][-1].content results col.query(query_texts[query], n_results5) memories [ doc for doc in results[documents][0] if doc is not None ] return {relevant_memories: memories}这里用到了ChromaDB的默认embedding函数本地直接能跑不用申请向量化服务的Key。生产环境我后面会建议你换专用embedding模型。关键点是n_results召回数量不要设太高5到8条足够。记忆检索不是越多越好给模型喂一堆弱相关记忆效果和没有记忆几乎一样还白白浪费Token。3.3 记忆写入节点从对话里“提炼”该记的东西这是整套机制里最核心的一个节点也是最容易被低估难度的节点。简单把所有对话全部存进向量库会带来两个问题一是垃圾信息进库以后检索幺蛾子多二是没有结构化的提炼同一个用户信息可能被存了十几种不同的说法互相打架。所以写入节点必须做一件事让大模型判断“这段对话里有哪些值得长期记忆的信息”然后用固定模板输出。做法如下from pydantic import BaseModel class MemoryExtraction(BaseModel): memories: list[str] updates: list[str] def extract_and_store(state: AgentState) - AgentState: last_round state[messages][-2:] prompt f 分析下面这段对话提取值得长期记忆的事实或偏好。 要求 1. 只提取稳定的事实和偏好不提取一次性情绪、临时状态 2. 如果用户明确在更正/推翻之前说过的话放入updates字段 3. 输出格式为JSON 对话内容 {last_round} response llm.with_structured_output(MemoryExtraction).invoke(prompt) for mem in response.memories: col.add( ids[hash(mem)], documents[mem], metadatas[{created_at: time.time()}] # 注意这里 ) return { memory_candidates: response.memories, memory_updates: response.updates, }这里有一个非常容易踩的坑向量数据库的id如果直接用“内容hash”那用户更正记忆时新内容hash变了旧条目不会自动被替换结果新旧两条同时存在。我的建议是存元数据时加一个memory_key比如这条记忆是“用户的居住城市”key就设为“user.city”。更新时按key删除旧数据再写入新数据保证同一类事实只有一个版本。下面第四章也会讲更新策略。3.4 状态更新与Graph串联把四个节点串成完整流程现在把检索、生成、提炼、存储这几个节点串成一张图。完整代码如下from langgraph.graph import StateGraph, END def should_continue(state: AgentState): # 简单判断用户消息里是否含有“记住”等写入信号 if any(k in state[messages][-1].content for k in [记住, 我的, 我喜欢, 我在]): return memory return respond def build_memory_graph(): g StateGraph(AgentState) g.add_node(retrieve, retrieve_memories) g.add_node(agent, agent_node) g.add_node(memory, extract_and_store) g.set_entry_point(retrieve) g.add_conditional_edges( retrieve, should_continue, {memory: memory, respond: agent} ) g.add_edge(agent, memory) g.add_edge(memory, END) return g.compile()实际使用的时候要注意这个conditional edge的逻辑比较粗糙只是关键词匹配。真实环境我会建议用一个小模型分类器来判断当前轮是否包含值得记忆的信息或者直接让记忆写入变成“每轮异步跑”而不是“靠规则触发”。规则触发的缺点是用户用了“我平时通勤靠地铁大概四十分钟”这样不带“我在”字样的句子就不会触发记忆白白丢掉一条关键画像。更稳的办法是把extract_and_store节点放在agent节点之后无条件执行。代价是多一次小模型调用但换来的是“全量提炼该存的全存”。成本可控收益更高。4. 记忆存储选型向量库不是唯一正确答案很多Agent开发教程会把向量数据库捧成记忆系统的唯一解但实际工程里“记忆存储到底用啥”完全取决于你的使用场景。我把主流方案对比一下你就知道该怎么选了。存储方案适合场景优点缺点典型产品向量数据库语义检索、相似度召回召回灵活、支持模糊匹配精确匹配弱、需要embedding成本ChromaDB、Milvus、Pinecone键值存储用户画像、结构化偏好读写快、精确覆盖更新无法语义检索Redis、Memcached关系型数据库有强结构的记忆实体事务一致、便于管理对语义检索不友好PostgreSQL、MySQL本地文件JSONL单机原型、调试阶段零依赖、可读性高检索性能差、规模上不去无这里我重点展开两类。4.1 什么时候必须用向量库什么时候用不上如果你的Agent主要是“知识问答型”用户会问“我之前是不是提到过我出差去深圳的事”那必须靠向量库做语义召回因为用户不会用你当初写入的原文来查询你需要的是语义匹配能力。如果你的Agent是“事务处理型”比如订机票、记日程、管任务那用户的记忆其实是高度结构化的日期、地点、参与人、事项。这种场景下纯向量库反而是绕远路。直接用键值存储比如Redis Hash把用户ID映射到一个JSON对象里读写干脆利落更新还不会产生语义残留。我可以直接给一个代码片段import redis, json r redis.Redis(hostlocalhost, port6379, db0) def save_preference(user_id: str, key: str, value: str): r.hset(fuser:{user_id}, key, json.dumps(value)) def get_preference(user_id: str, key: str): data r.hget(fuser:{user_id}, key) return None if data is None else json.loads(data)这套方案对需要精确覆盖更新的记忆项如“当前所在城市”“当前职位”非常友好。注意不要覆盖已经存过的历史轨迹把当前值存一份加一个字段做历史表以后做用户分析会很有用。4.2 embedding模型怎么选性能和成本的平衡向量库本身不产生向量你得用embedding模型把文本变成向量。这里的选型直接影响记忆检索效果。追求零成本本地跑用ChromaDB默认的all-MiniLM-L6-v2大约80MB模型文件CPU跑得动中文效果凑合。中文场景入门用BAAI/bge-small-zh-v1.5本地可跑中文效果比通用英文模型好不少。生产级中文场景用bge-large-zh-v1.5或通义千问的text-embedding-v3效果好但需要显卡或API调用成本。多模态场景如果要存图片相关记忆需要用CLIP类模型把图文映射到同一向量空间。有一个经验值得分享embedding模型的维度直接影响存储和检索成本。512维和1024维的向量库在数据量大时资源和速度差距非常明显。如果你的场景是轻量级记忆512维足够盲目上大模型只是给自己添负担。embedding模型一旦选定尽量不要中途换否则旧记忆向量和新模型生成的向量不在同一空间语义对齐会出问题。真到必须换的那天老老实实做一次全量迁移重建。4.3 记忆的时间衰减与优先级设计记忆系统不能只有存取还得有“老化”逻辑。长期不用的记忆检索权重应当逐渐降低最近确认过的记忆权重应该提高。实现上不用太复杂给每条记忆打两个标签就够created_at是创建时间last_accessed_at是最后命中时间。检索时可以给结果加一个简单的重排分数score 向量相似度 α * recency_bonus其中α根据你的业务调。这样既保持了语义相关性又让最近说过的话更容易被想起非常实用。为什么必须做时间衰减如果没有这一层可能会出现一种尴尬情况用户两年前提到自己喜欢玩某款游戏现在早就不玩了但每次对话系统都在提这个老梗。召回不是越多越好让旧记忆慢慢沉底本身就是一种隐性的遗忘机制。5. 记忆的更新与遗忘会正确忘记的Agent才是好Agent真正落地记忆系统后你会发现“写入什么”是第一个坎“怎么更新旧的”是第二个坎“怎么主动遗忘”是第三个坎。前两个不解决记忆系统跑三个月后就会变成一个自相矛盾的大杂烩。5.1 用户主动纠错的更新链路用户说“我之前说我在北京其实我现在搬到上海了”。如果你的系统没有更新机制向量库里可能同时存在“用户在北京”和“用户在上海”两条记录模型检索时两条都会命中回答就非常拧巴。解决方案是给长期记忆引入“memory_key”机制。每条记忆写入时都会关联一个语义上的key比如“user.city”。更新时先用key删除旧记录再写入新记录。关键代码def upsert_memory(user_id, key, new_fact, new_embedding): # 1. 按key删除旧记忆 old_items col.query( where{user_id: user_id, memory_key: key} ) for item in old_items[ids]: col.delete(ids[item]) # 2. 写入新记忆 col.add( ids[f{user_id}:{key}:{uuid4()}], documents[new_fact], metadatas[{user_id: user_id, memory_key: key}] )注意删除旧记忆时用的是结构化字段不是靠内容相似度。用相似度删除非常危险很容易误删不该删的内容。这也就是为什么我坚持每条记忆写入时必须带结构化元数据——没有这个字段更新与纠错就是一笔糊涂账。5.2 冲突检测什么时候该问用户而不是自作主张有一类冲突是机器判断不了的。比如用户第一次说“我女朋友喜欢喝咖啡”第二次说“我男朋友也喜欢喝咖啡”。这两个信息不一定矛盾可能是两任对象也可能用户表述有变化。遇到这种情况更安全的做法不是直接覆盖而是把冲突写入待确认队列在对话中自然发问“你之前提到的是另一位还是更新了偏好”我习惯在记忆写入节点里增加一个conflict check步骤提取新记忆后用关键词和语义检索找到同类历史记忆做一个一致性判断。如果模型判断为高度冲突就把问题返回给对话节点让Agent主动向用户求证。一句话总结记忆系统在拿不准的时候最有价值的动作是提问而不是猜。5.3 主动遗忘的工程实现遗忘机制常被忽略但一个好的记忆系统必须包含主动遗忘。遗忘的触发条件一般有三类第一类是“过期性事实”。比如用户的临时日程“下周去深圳开会”——会议一开完这条就没用了定时任务每周清理过期日程类记忆即可。第二类是“负反馈信号”。用户明确说过“不要再提这个话题了”那相关记忆要么删除要么打上suppress标签检索时直接过滤掉。第三类是“长期未命中”。比如超过90天没有被检索到的记忆降至冷存储或直接删除。别心疼这些内容——用户如果真需要重新说一次的成本远低于系统每天在这些低质量记忆上耗费的检索精度。实现上我通常给每条长期记忆加一个状态字段简单三段式就够了状态含义处理方式hot近期或高频命中的核心记忆全量参与检索warm偶尔用到的一般记忆参与检索但权重降级cold长期未命中的边缘记忆不进主检索仅存档定期跑一个离线任务把超过30天未命中的记忆从hot降为warm90天未命中降为cold。所谓“记住了”其实是对用户当前最有用的信息触手可及。6. 带记忆Agent容易翻车的四个隐蔽坑最后这部分是我在实际测试各种开源Agent项目时的经验沉淀。前五章讲的是“怎么搭”这章讲讲“怎么避免搭完之后难用”。6.1 记忆检索的上下文污染问题最典型的翻车现场是用户问“帮我推荐一部今天能看的电影”这时候记忆系统把用户三个月前说过的“我不喜欢科幻片”也取出来了模型回复就在那纠结半天“虽然你之前不喜欢科幻但今天推荐的这部不错”。其实用户自己都忘了自己说过这话被系统一提反而觉得被冒犯。解决办法是给记忆检索加“时效性加权”近期记忆权重高远期记忆权重低。或者更直接一点在注入Prompt时加一句指令“以下记忆可能过时如果与当前对话冲突以当前对话为准。”别小看这句话它能抵消掉不少记忆污染带来的副作用。6.2 摘要压缩过程中的信息失真短期记忆的摘要压缩如果做得太粗暴会把关键信息丢掉。典型例子用户和Agent讨论项目细节说“数据库连接串在.env里端口是3306用户名是root密码在团队密码本里”。摘要压缩时模型可能只保留“数据库配置信息”几个字变量全丢了。下次用户问“数据库密码怎么拿”系统完全答不上来。我的建议是摘要节点使用带结构的Prompt要求“保留所有专有名词、数字、API接口名称、用户明确表达的偏好原句”。甚至可以给摘要设计一个固定模板关键事实列表、明确指令列表、待办事项列表。这样至少能保证压缩后骨架还在。6.3 个人记忆与共享记忆的隔离问题如果Agent服务面向多个用户记忆必须做严格隔离。别觉得这是句废话——很多人在多用户Agent上翻车就在这一步。因为向量数据库的query如果不带where过滤条件它默认是全库检索。一旦用户A的对话文本和用户B的有相似性A的检索结果里就可能出现B的记忆这是严重的隐私事故。所以我建议所有Embedding存储操作在写入时强制带user_id元数据查询时强制带user_id过滤并且这个过滤条件要写在代码里而不是依赖调用方传参。同时提供一个外部偏好开关让用户查看系统记得自己什么、手动删除某条记忆这个功能既是隐私合规需要也是产品体验上加分的小细节。6.4 记忆系统的可观测性记忆系统是个“黑盒”很容易让人抓狂模型突然说“我记得你之前说过……”的时候开发人员根本不知道这个结论是从哪条记忆来的。我的做法是所有注入上下文窗口的记忆统一加上引用来源标记。比如改成“根据记忆如果记忆是‘用户喜欢靠窗座位’那最终注入的内容是“[记忆#322024-06-20记录] 用户喜欢靠窗座位”。这样一来产品阶段用户发现问题时你能快速定位到具体是哪条记忆导致的。排查成本从小时级降到分钟级。建议日志里至少记录每次检索的query、召回的记忆id、注入Prompt的记忆文本、最终模型的回答。这四样配齐记忆系统就具备基本的可追溯性了。回到最初的问题——让Agent记住你到底难不难如果只是把对话录像存起来再检索三天就能做出来。但如果要做到“该记的能记、该忘的能忘、说错了能改、各用户之间不串味”就需要先梳理清记忆分层架构再把LangGraph、向量库、键值存储这些工具按正确的姿势组合起来。我个人在做这套方案的第二个版本时最大的感悟是记忆系统的复杂度和Agent业务本身的复杂度是同比例增长的一开始就留出更新机制和隔离边界后面会省掉大量改数据的力气。如果你的Agent也已经到了“上下文总是不够用、用户老觉得你没记性”的阶段不妨按这篇的思路重构一遍结果会很不一样。
返回列表