开头前100字里直接点出“Agent”和“记忆系统”这两个核心关键词用工程师聊天的口气展开。做Agent开发做到“记忆系统”这一层基本上就进入了真正的工程阶段。前面几篇我们聊过Agent的任务拆解、工具调用和编排框架那些解决的是“让Agent能干活”的问题而记忆系统解决的是“让Agent越干越像个人”的问题。这一篇我会从数据模型、读写策略、存储选型和并发处理几个角度把我在实际项目里积累的Agent记忆系统落地经验拆开讲清楚适合正在从Demo走向生产的大模型开发工程师也适合刚入行、想系统理解Agent记忆机制的同学。1. 先搞清楚Agent记忆系统到底在解决什么问题1.1 没有记忆的Agent本质上是一条金鱼很多刚接触Agent开发的同学会有一个错觉LLM本身不是能记住上下文吗为什么还要单独做一个记忆系统这个问题的答案要从LLM的工作方式说起。模型本身是无状态的——它不保存任何历史信息所谓上下文只是每一轮请求里你塞给它的那一堆文本。你这次调用传了多少内容它就只看得到多少内容下一次调用又是从零开始。拿我最早做的客服Agent举例。用户说“我要退掉昨天买的那个红色耳机”当时对话里确实有“昨天买了一个红色耳机”这样一条信息模型能理解。但如果用户第二天又来说“昨天说的那个耳机我不想要了”模型根本不知道“那个耳机”指的是什么。它没有跨轮次记忆更谈不上跨会话记忆。这就是记忆系统的第一个价值跨会话的状态保持。没有记忆Agent就是个失忆症患者每一次对话都像是第一次见面。第二个价值是减少重复提问、降低Token消耗。如果用户已经在上一轮对话里告诉过你“我在上海工作通勤单程40分钟”那么再做通勤建议类任务时记忆系统可以直接把这条偏好注入Prompt而不是让用户重新说一遍也不用每次把完整的历史对话都塞给模型。1.2 记忆系统的三层结构对应你钱包里的三种钱工程上做记忆系统我习惯把它分成三个层次别想复杂了层次对应概念生命周期成本特征工作记忆当前Prompt里的上下文窗口当轮请求结束即失效贵每轮都要付钱短期记忆一次会话内的多轮上下文会话结束可以考虑清理中等需要滑动窗口管理长期记忆跨会话持久化存数据库/向量库按遗忘策略淘汰便宜但需要工程开销这三个层次不是孤立存在的它们之间有一个降级与提升的过程对话过程中产生的短期记忆里那些高频出现、具有长期价值的信息会被提取出来写入长期记忆下一轮对话开始时长期记忆里相关的部分会被检索出来重新注入到工作记忆也就是Prompt里。这里有一个设计原则值得反复强调能放长期记忆的就尽量别放短期记忆能放短期记忆的就别硬塞进每一轮的Prompt里。因为Prompt越长响应越慢、成本越高而且模型对超长上下文的注意力会明显衰减——这不是玄学是实测结论。注意力在长上下文里分布很不均匀中段信息最容易丢。把重要信息放在头部系统提示和尾部最近几轮是上下文工程的基本功。1.3 一个记忆系统的完整工作链路我用一张流程来说明记忆系统在整个Agent里的位置一次完整对话经历四个阶段——感知、理解、决策、行动。记忆系统主要介入两个时间点对话进行中每轮判断“这段内容要不要写入记忆”对话结束后或后台异步执行“记忆提取、归纳、归档”的任务。下一次对话开始时先做“检索、筛选、注入”把相关记忆拼到Prompt里。整个链路里最容易做崩的地方有三个写什么、怎么取、忘不了——分别对应写入策略、检索策略和遗忘策略。后面三个章节我逐层拆解。2. 记忆的数据模型把“记忆”变成可存储、可检索、可遗忘的结构2.1 记忆不是一段文本而是一个有Schema的对象很多第一次做记忆系统的开发者会踩同一个坑把历史对话直接存进一个JSON数组或者干脆存成一大段Markdown文本用的时候整段塞进Prompt。这个方案在小规模验证时能跑通但一旦用户量上来、对话轮次变多问题会集中爆发你没法精确查“这个用户上次提到的收货地址是什么”只能把一整段历史全拉出来让模型自己找你没法单独删除某条过期信息比如用户已经改了手机号但旧的手机号还在记忆里模型每次都优先看到旧号码你也没法量化“哪条记忆更重要”所有历史地位平等结果就是重要的信息淹没在噪声里。所以记忆系统的第一性原理是把记忆当成数据建模而不是当成文字处理。一条记忆应该是一个结构化的对象包含核心字段namespace记忆的领域分类比如用户偏好、业务状态、任务进度entity_id记忆属于哪个主体比如用户ID、订单IDcontent记忆的核心内容可以是JSON结构也可以是文本片段importance重要性权重0到1之间created_at / updated_at时间戳expires_at过期时间长期记忆可以设置为nullsource记忆来源比如“对话记录#第3轮”memory_type事实型记忆用户的姓名、偏好型记忆喜欢简洁回复、状态型记忆退款流程进行中我用的一个典型长期记忆条目长这样{ namespace: user_preference, entity_id: user_1024, memory_type: preference, content: { timezone: Asia/Shanghai, response_style: 简洁直接, shipping_address: 上海市浦东新区XX路100号 }, importance: 0.85, created_at: 1734576000, updated_at: 1734576000, expires_at: null, source: conversation#9217 }2.2 记忆的四种类型决定了不同的写入和淘汰策略工程上我把记忆分成四类每一类的读写法都不一样第一类是事实型记忆用户的姓名、公司、地理位置、业务往来记录。这类记忆通常由信息提取模型从对话中抽取写入后基本稳定不变适合长期保存更新条件是“出现了新的冲突事实”。第二类是偏好型记忆用户喜欢什么样的回答风格、更关心什么指标、对什么敏感。这类记忆的提取难度高需要从多轮对话里总结归纳而且会随着用户行为变化而变化需要定期重估。第三类是状态型记忆当前正在进行的任务进行到哪一步了比如购物流程、审批流程、售后流程。这类记忆要求实时性和强一致性属于短期记忆为主任务结束后可以归档或清除。第四类是能力型记忆Agent从历史任务中学到的技能、工具使用经验。这类记忆目前业界做得还不多但它是“经验积累”这个方向的核心载体通常以“经验片段”的形式存储配合复用触发条件。这个分类的意义在于不同种类的记忆生命周期和更新机制天然不同。偏好型记忆需要归纳和定期刷新状态型记忆需要精确到字段级别更新能力型记忆则需要嵌入到决策流程里。一个统一的“存文本、取文本”方案根本无法承载这些差异。2.3 为什么我不建议用纯向量存储搞定一切长期记忆的检索很多人第一反应就是“上向量数据库”。这没问题但只靠向量记忆系统做不扎实。向量检索擅长解决语义相似的问题比如“用户上次提到养了一只柯基我不记得名字了搜‘狗’也能搜出来”。但记忆里很多查询是精确匹配比如“用户当前处于哪个审批环节”、“用户的会员等级是多少”。这些用向量检索反而效率低、准确率不稳定。我的建议是混合检索事实型、状态型记忆用结构化存储如SQLite、PostgreSQL、Redis跑精确查询偏好型、经验型记忆用向量存储跑语义相似度检索最后把两类结果统一做重排注入Prompt。后面第四个章节会讲一个具体的工程实现。3. 记忆读写策略什么时候写、怎么写、怎么取、怎么忘3.1 写入策略不是每句话都值得记记忆写入最常见的错误是事无巨细地记录。对话里用户说了一句“今天天气不错”也要存进长期记忆吗当然不。长期记忆是有存储成本和检索噪声的记了太多低价值信息检索的时候高价值信息反而被淹没。我给项目定的写入规则是三元组判断内容是否与用户或任务的长期属性相关。用户换了城市生活、变更了收货地址、改变了对回复风格的要求这些值得记。内容是否对未来的交互有复用价值。用户说“我每天晚上都要开会到八点”这就可以记因为后续安排任务时间时会用到。内容是否与正在进行的关键状态变化有关。退款订单的状态变了、审批通过了一笔款项、任务切换到了新的阶段这些必须记。满足其中一条就可能写入一条不满足那这轮对话就让它过去。具体落地时我通常安排一个独立的LLM调用做记忆提取单独写Prompt在每一轮对话结束后异步执行prompt 你是一名记忆提取器。请从下面的对话中提取值得长期保存的信息。 只提取与用户属性、任务状态、业务流程强相关的信息忽略寒暄问候。 输出一个JSON数组每个元素包含 - namespace: user_preference | business_state | task_progress - memory_type: fact | preference | state - content: 结构化的信息内容 - importance: 0到1之间的小数 对话内容 {conversation_text} 这个提取过程最怕两件事提取过泛导致噪声过多以及提取过窄导致关键信息漏提。我的经验是给模型一个“重要性”的判断标准并且每一类namespace有一个专属的content格式约定提取质量会比自由发挥稳定得多。3.2 检索策略相关度、时间衰减、重要性三者加权记忆写入后怎么在需要时把最合适的那几条捞出来是决定用户体验的关键环节。我在几个Agent项目里沉淀下来的一套排序方案核心公式是这样[ score \alpha \times similarity \beta \times importance \gamma \times recency_decay ]similarity查询与记忆的语义相似度由向量检索得到importance记忆本身的重要程度写入时就已经确定recency_decay时间衰减因子比如对最近7天内的记忆给满权重超过30天按月线性衰减α、β、γ三个超参数我常用的是0.5、0.3、0.2具体按业务调优排序选出Top-K条通常在5到15条之间再拼成“记忆上下文”注入Prompt。这里有一个检索范围设计的关键点namespace必须作为硬过滤条件。比如用户当前在咨询退款问题那么检索应该优先限定在business_state和task_progress这两个namespace里user_preference里那些内容可以少取甚至不取。否则就会出现用户明明是在问订单号多少记忆却优先给他的偏好排序喂一屏“用户喜欢什么文风的回复”这就是检索精准度崩坏。检索另一个常见问题是注入成本。每条记忆都是Token5条短文记忆还好15条长文记忆可能就额外产生1500到2000个Token的消耗。所以在注入之前我会先做一个裁剪只注入和当前问题实体相关的记忆字段比如用户地址类的记忆只取content.shipping_address这个字符串而不是把整个JSON序列化塞进去。3.3 遗忘与更新记忆系统必须会“忘”记忆系统设计和人类记忆有个共同点不会遗忘的记忆系统很快就会被历史垃圾淹没。遗忘策略在工程上有三种主流的做法第一种是预算制。给每个用户或每个任务设定一个记忆条目数上限比如1000条。超限时按score从低到高淘汰。这个方案的优点是实现简单、运行稳定缺点是淘汰时可能会丢掉尚未充分复用的新记忆。第二种是过期制。每条记忆写清楚expires_at到期自动归档或删除。状态型记忆尤其适合用过期制退款流程三个月后基本不可能还有用直接清掉。第三种是重要性淘汰。和预算制类似但淘汰优先级以importance为主保留高价值记忆。这三种策略不是互斥的我在生产环境是组合使用状态型记忆走过期制偏好型和事实型记忆走预算制重要性淘汰特殊的高价值记忆打标签永不淘汰。更新策略同样关键。用户三月份在深圳生活六月份搬到了杭州记忆里如果还是“用户在深圳工作”就会持续产生错误的信息注入。所以做记忆更新时要优先处理冲突新写入的事实如果和旧记忆指向同一个entity_id namespace content字段那么覆盖旧值并更新时间戳而不是新增一条。否则检索时可能出现两个互相矛盾的记忆同时出现模型一犹豫结果就飘了。4. 记忆系统的工程化落地存储选型、并发与一个可跑的架构4.1 选型对比别一上来就上重型向量库记忆系统的存储选型应该是从业务规模倒推的而不是从技术热度正推的。如果你的Agent还是一个Demo、内部工具或者千级用户以内的应用用SQLite加sqlite-vec插件或者直接用PostgreSQL的pgvector扩展就足够了。这类方案的好处是运维成本极低、事务能力强还能和业务表放同一个库里容易保证一致性。我见过很多团队一上来就部署Milvus集群结果用户量三位数向量集合几千条纯属给自己找运维负担。当数据规模到了千万级向量、QPS到了几百以上再考虑独立的向量数据库比如Qdrant、Milvus或者云上的向量服务。这时候你还要考虑配套的索引类型HNSW还是IVF、分片策略这些问题复杂度一下子上去。我整理了一份选型参照表纯属个人实践结论存储方案适用规模检索能力运维成本我的建议SQLite sqlite-vec万级以内精确查询强极低原型和小规模首选PostgreSQL pgvector百万级以内混合查询强低业务库顺手用推荐Redis 向量插件十万级内极快低高频、短期记忆场景Qdrant千万级专精向量中长期记忆独立服务Milvus亿级专精向量高大规模平台化慎选这个表的核心结论是大多数Agent项目根本用不到独立的向量数据库。你的长期记忆规模在相当长一段时间里是几千到几万条用PostgreSQL加pgvector完全够跑还能白嫖事务和备份。4.2 并发与一致性多实例写入时的三个坑Agent系统一旦上生产就不会只有单实例。用户请求会负载均衡到多个服务节点每个节点都可能异步写记忆这时候并发问题就来了。第一个坑是重复写入。两个并发的记忆提取任务可能同时检测到“用户换城市了”这一事实于是同时插入两条几乎一样的记忆。解决方式是在数据库层面做唯一约束key是entity_id namespace content_hash插入时用ON CONFLICT DO UPDATE而不是先查后插。第二个坑是覆盖更新丢失。用户先发起一个会话修改了手机号另一个会话同时也在改回旧手机号两个写操作先后到达时间戳旧的那次覆盖了新的。解决方式有些团队用Redis分布式锁串行化同实体的记忆更新有些团队依赖数据库的乐观锁CAS。我推荐优先用乐观锁——更新时带上updated_at条件更新不成功就重试。第三个坑是读多写少场景的缓存问题。记忆通常读的频率远高于写的频率所以缓存是必要的。但缓存和库里数据不一致会造成用户上一秒改了偏好下一秒Agent还在按旧偏好在回答。我的方案是记忆写入时同步失效缓存下一次读取重新查询入库。短期记忆和高频检索的偏好记忆可以放在Redis里长期低频繁记忆直接查库。4.3 一个可跑通的工程实现示例我用一个简化但完整的Python示例演示记忆系统在工程上的最小闭环。这套结构我在多个项目里用过逻辑上可以直接复制去改。import json import sqlite3 import hashlib from typing import Optional class MemoryStore: def __init__(self, db_path: str): self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, entity_id TEXT NOT NULL, namespace TEXT NOT NULL, memory_type TEXT NOT NULL, content TEXT NOT NULL, importance REAL DEFAULT 0.5, created_at INTEGER, updated_at INTEGER, expires_at INTEGER, content_hash TEXT, UNIQUE(entity_id, namespace, content_hash) ) ) self.conn.commit() def write_memory(self, entity_id: str, namespace: str, memory_type: str, content: dict, importance: float, expires_at: Optional[int] None): raw json.dumps(content, ensure_asciiFalse, sort_keysTrue) content_hash hashlib.md5(raw.encode(utf-8)).hexdigest() now int(time.time()) self.conn.execute( INSERT INTO memories (entity_id, namespace, memory_type, content, importance, created_at, updated_at, expires_at, content_hash) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(entity_id, namespace, content_hash) DO UPDATE SET content excluded.content, importance excluded.importance, updated_at excluded.updated_at, expires_at excluded.expires_at , ( entity_id, namespace, memory_type, raw, importance, now, now, expires_at, content_hash )) self.conn.commit() def read_memories(self, entity_id: str, namespace: Optional[str] None, limit: int 10) - list[dict]: sql SELECT * FROM memories WHERE entity_id ? params [entity_id] if namespace: sql AND namespace ? params.append(namespace) sql ORDER BY importance DESC, updated_at DESC LIMIT ? params.append(limit) rows self.conn.execute(sql, params).fetchall() return [self._row_to_dict(row) for row in rows] def forget_expired(self): now int(time.time()) self.conn.execute( DELETE FROM memories WHERE expires_at IS NOT NULL AND expires_at ?, (now,) ) self.conn.commit()这个实现里有几个工程细节值得单独强调ON CONFLICT DO UPDATE处理了重复写入content_hash做了去重去冲突read_memories用importance DESC, updated_at DESC作为默认排序这能在没有向量检索的情况下先跑通业务——很多场景下单纯按重要性和时间排序就已经能覆盖大部分需求向量检索是加分项而不是必需品。真正的检索模块是在这个基础上叠加向量化。生产项目里我通常用两条腿走路精确查询直接SQL语义查询先向量化再做向量检索最后合并排序。5. 常见问题与排查实录5.1 记忆污染模型被错误的旧记忆带偏这是记忆系统上线后最容易被用户感知的问题。用户已经在对话里明确说“我的手机号改成138了”但Agent还是按照旧号码在操作或者用户已经很烦躁地说了三遍“我是要退货不是要换货”但状态型记忆一直写着“用户申请换货”。排查方向有两个一是写入链路确认新事实是否真的被提取并覆盖了旧记忆重点看更新的条件有没有生效二是检索链路确认相同实体下是否长期保存了互相矛盾的记忆且旧记忆并没有因为冲突更新而被覆盖。我的习惯是给记忆条目加一个updated_at的字段排查时按时间倒序看矛盾记忆的先后顺序一目了然。另外一个更彻底的方案是在注入Prompt前加一道LLM校验把检索出来的记忆和当前这一轮的用户问题放在一起让模型判断“这些记忆是否和当前对话冲突”冲突的记忆直接丢弃不参与注入。这个校验会把记忆检索延迟多几百毫秒但换来的是回答质量的稳定性值得。5.2 Token爆炸记忆越攒越多Prompt越来越长用户量大了以后每个用户的长期记忆条数会持续增长如果检索Top-K设定得太高或者每条记忆的content太长光是记忆部分的Token可能就超过3000再加上系统指令、工具定义、近期对话一次请求的输入直接顶到上万Token。排查的顺序先看检索的limit值是否合理很多问题就出在“为了让模型理解更充分我多取了几条”这种心态上再看注入格式是把整个JSON注入还是只注入了相关内容字段最后检查过期清理任务有没有在跑如果清理逻辑半年没触发过那大概率存了一堆已经失效的历史状态。我在项目里通常把记忆注入预算和对话输出预算拆开管理比如总Prompt预算8K Token其中记忆最多占2K超了就要压缩或者优化格式。这不是硬性的数值但预算意识必须一开始就有否则上线三个月后你会被账单和延迟双重教育。5.3 检索不命中看起来明明有这条记忆但搜不到这个问题的根源通常不在向量检索本身而在写入阶段。写入时如果只存了原始对话片段比如“用户上周问我怎么配置Nginx”当用户这周问“服务器反代怎么搞”时语义上确实应该能关联上但如果向量化模型本身对短口语文本的语义泛化能力一般距离就会偏大。我的排查习惯是三步走第一步看这条记忆到底写没写进去排除写入链路问题第二步手工把查询向量化和记忆向量算余弦相似度看距离到底多大排除查询文本的问题第三步如果距离确实偏大就考虑把查询做扩展比如用LLM生成几个同义改写再分别查询取分数最高的结果。这个方法在名字难匹配、语境差异大的场景里效果好但成本会增加。还有一种“不命中”特别隐蔽检索命中了但排序把它排在Top-K之外。这类问题的调试方式是对检索接口输出分数debug日志看这条记忆的真实得分再做对应调整。5.4 多实例并发写冲突记忆被后写入的旧数据覆盖这个在前面已经提到了具体方案但我想特别强调排查现象。典型表现是同一个实体的记忆更新时间在回退明明五分钟前更新为“杭州”日志里却显示三分钟前又写回了“深圳”。这种回退事故大概率是并发更新没有用乐观锁。解决方式是更新语句里带上前一次读取的version或者updated_at作为条件更新不生效就说明被别人改了重新读取再合并提交。这个方案在数据库层面几行SQL就能解决千万不要为了省事用“先删后插”的无脑覆盖式更新。另外如果业务对状态型记忆的一致性要求极高比如涉及资金、审批这类流转建议在架构层面加一个内存队列把同一实体的记忆写操作串行化。虽然损失一点吞吐但换来的是一致性。实操总结几个支撑我做完整个系统的核心体会最后聊点掏心窝的话。整个Agent记忆系统做下来我最大的感受是记忆系统的高下不在技术选型的豪华程度而在设计者对信息价值的判断能力。你知道什么是重要信息系统才知道该记什么你会给记忆分门别类系统才知道该忘什么你理解了LLM上下文的物理限制系统才知道该存什么。不要迷信BigModel能解决一切更不要迷信单个向量库能扛住所有语义检索沉下心来把数据模型设计干净、把读写策略写清楚这个系统才能在上线之后经得住流量和业务的反复捶打。最后分享一个小技巧给每条记忆在写入时增加一个trace_id关联上这一条记忆是由哪一轮对话提取产生的。排查“模型怎么知道这个信息的”这类问题时这一个小字段能帮你省掉至少一个小时的追查时间。这个习惯我到现在还在用也希望读过这篇的你在自己工程里试一次。