做了几年 AI 应用落地,我最大的感受是:模型能力不是瓶颈,记住事情才是。客户不会因为你的 Agent 会写诗而续约,但会因为 Agent 上一分钟刚聊完的需求,下一分钟就忘得一干二净而退货。所以去年我启动了一个有点“笨”的项目——从零构建一个生产级记忆型 AI Agent,底层基于 AgentScope 框架,把长短期记忆网络、双网络记忆模型、以及“记忆=score+时间半衰期”这套机制全部落到了代码里。
这个项目做下来,我踩了无数坑,也把所有能想到的并发、持久化、记忆污染问题都过了一遍。今天把全景写出来,既是给自己的项目做一次复盘,也想给正在从 0 到 1 搭建 AI Agent 的同学一份可以直接抄作业的参考。文章会偏工程落地,重点不在概念,而在于你怎么把“记忆”这个看似抽象的东西,变成一套能扛住生产流量的系统。
1. 项目整体设计与思路拆解
1.1 没有记忆的 Agent 就像一条金鱼
先说个最直观的问题:为什么记忆型 Agent 在真实业务里几乎是必需品。你看市面上的 AI Agent 产品,凡是给人留下“聪明”印象的,基本都有记忆能力。用户偏好、历史决策、对话上下文、任务进度,这些东西如果每次对话都从零开始,用户就感觉自己在跟一台刚出厂的机器说话。
我接过的很多需求都指向同一个痛点:Agent 能回答问题,但不能连贯地为同一个人服务。比如客服场景,用户上周反馈过网络延迟,这周再问“怎么又卡了”,Agent 如果记得上次的诊断结论,可以直接说“上次我们定位到是您家路由器的 5G 频段问题,这次还有类似现象吗”,客户体验完全不一样。这就是记忆型 Agent 的核心价值——把一次性对话变成可持续的服务关系。
本项目要做的不是一个 Demo,而是一个生产级系统。生产级的含义至少包括:支持多用户多会话并发、记忆数据不丢失、检索延迟可控、系统可观测可回放。这些点和“用一个向量数据库存聊天记录”完全是两个量级的事。
1.2 为什么选择 AgentScope 作为项目骨架
选型阶段我对比过 LangChain、LlamaIndex、CrewAI、AgentScope 等框架。最终选了 AgentScope,主要有三个原因。
第一,AgentScope 是多智能体编排框架,本身内置了 Agent 生命周期管理、消息传递和 Pipeline 机制,省掉了我自己造轮子处理多轮对话状态的问题。第二,AgentScope 在工程化这一侧做得比较规整,对异步、批量调用、资源控制的支持比很多偏学术的框架落地实在。第三,它的设计哲学是“让 Agent 开发像后端开发一样可控”,这点非常对我的胃口——我要的是能上生产的框架,不是一套只能跑 Jupyter Notebook 的玩具。
不过这里要多说一句:框架只是骨架,记忆系统必须独立设计。我们项目的核心资产不在 AgentScope 本身,而是一套挂在 AgentScope 旁边的记忆服务。这个服务负责记忆的写入、检索、衰减、固化和遗忘,Agent 每次对话前从里面拿上下文,对话后把新信息写进去。
1.3 核心架构:双网络记忆模型
项目采用的记忆架构是双网络记忆模型。这里的“双网络”不是指两个神经网络,而是指记忆系统在逻辑上分成两个通道:一个是短期记忆网络,一个是长期记忆网络。
短期记忆网络负责承载当前会话的上下文,包括最近几轮对话、用户当前的任务目标、临时产生的中间结果。它的特点是读写极快,数据量小,跟随会话生命周期。长期记忆网络负责跨会话沉淀用户画像、历史偏好、关键事实和长期任务状态。它的特点是容量大、需要持久化、检索依赖索引。
这两个网络之间有双向同步机制。短期记忆里那些高价值信息会经过“固化流程”进入长期记忆;长期记忆中的相关信息会在对话开始时被检索出来,注入短期记忆作为当前上下文的背景。我在实现里把这两个网络拆成了两个独立的存储引擎,中间通过一个 Memory Sync 服务做异步桥接,避免对话链路被记忆固化的延迟拖住。
这套架构解决了一个很实际的问题:如果你把所有记忆混在一个库里,短期信息会淹没长期信息,检索噪声会指数级上升。分离之后,短期检索跑 Redis 或者内存,长期检索跑向量库,各自优化,互不干扰。
2. 记忆的底层机制:score、半衰期与长短期分工
2.1 记忆=score+时间半衰期:遗忘曲线的工程化
很多团队做记忆型 Agent 时,最大的误区是只做“存储”和“检索”,不做“遗忘”。一个无限膨胀的记忆库,最后必然导致检索命中率下降、token 成本飙升、模型被无关信息带偏。所以我把记忆的底层机制设计成了“记忆=score+时间半衰期”这个公式。
每一条记忆在写入时都会被赋予一个初始分数(score),这个分数代表记忆的重要性。来源不同,分数不同。用户主动强调的信息、与当前任务强相关的信息、检测到情感强度的信息,初始分更高。常规闲聊、重复冗余信息、一次性操作日志,初始分更低。
但光有分数不够,记忆会随时间衰减。我用了指数衰减模型,公式是:
def current_score(initial_score: float, created_at: float, now: float, half_life: float = 604800) -> float: elapsed = now - created_at return initial_score * (0.5 ** (elapsed / half_life))这里的 half_life 默认取 7 天。为什么用半衰期而不是“30 天后直接删除”?因为真实遗忘曲线是指数型的,一条记忆在刚产生时衰减最快,但如果它被反复访问,它的有效强度就会维持住——这正好对应人脑的“间隔重复”效应。工程上,指数衰减可以写成简单的浮点运算,不需要定时清理任务去扫描全表,读取时惰性计算即可,性能开销极低。
检索时最终排序分数是“相关度”和“记忆强度”的融合。我用的权重是 0.4 的向量相似度 + 0.6 的当前有效分数。这个比例是我在测试集上调出来的,不能照搬,后面会说怎么调。
2.2 长短期记忆网络如何分工协作
我项目里的 LTM(长期记忆)和 STM(短期记忆)不是抽象概念,而是两套真实的存储服务。
短期记忆存储我在实现里用了 Redis,数据结构是每个会话一个 List plus Hash。List 保存最近的 10-20 轮对话原始内容,Hash 保存当前会话提炼出的临时关键信息,比如用户偏好、正在处理的任务 ID。短期记忆的 TTL 我一般设成 24 小时,超过会话生命周期的短期记忆没有意义,强行保留只会污染以后的新会话。
长期记忆存储用的是 PostgreSQL 做元数据和关系管理,加上 pgvector 做向量索引。每条长期记忆包含几个关键字段:记忆内容、用户 ID、实体标签、初始分数、创建时间、最后访问时间、访问次数、来源会话 ID。这里有个很重要的设计:长期记忆不只是存一段文本,还要存它的上下文来源。如果某条记忆被后续证据推翻,我们可以溯源并做修正。
长短期记忆网络的关键在于“固化”策略。哪些短期记忆值得进入长期记忆?我们定了几条规则:
- 用户显式表达的偏好(“以后都给我用这种风格”)
- 跨多轮重复出现的信息
- 当前任务被判定为“完成”,其中产生的关键决策
- 有效分数达到阈值的记忆
固化操作是异步的,不阻塞 Agent 的响应链路。这样长短期配合,Agent 既拥有当前会话的灵敏性,又拥有跨会话的稳定性。
2.3 记忆全流程:写入、固化、检索、遗忘
整条记忆流程我按四个阶段来设计。
写入阶段,Agent 的 response 完成后,系统会把本轮交互送入一个预处理管道,做 PII 脱敏、语言标准化、去重判断,然后生成一条或几条候选记忆。这里的去重非常关键,不做去重,同样一条“用户喜欢简约风格”会被写入几百遍,长期记忆库就废了。
固化阶段,这是一个后台任务,周期性扫描短期记忆中分数较高的条目,合并相似项,生成摘要式长期记忆,并建立向量索引。合并不是简单覆盖,而是生成一条新记忆,并保留原始记忆的引用。
检索阶段,用户发起新对话时,系统先取出用户的长期记忆向量,做一次相似度召回,再和 Redis 里的短期记忆合并排序,按融合后的得分取 TopK,拼进 system prompt。这里注意:不是把所有检索结果都塞进去,而是严格设置一个提示词预算,比如最多 1200 token 的记忆注入。
遗忘阶段,当记忆的有效分数衰减到阈值以下,或者长期不被访问,它会被标记为“冷却记忆”,不参与日常检索。冷却记忆不是物理删除,而是归档。归档后如果用户再次提起相关话题,可以重新激活。这个设计比硬删除安全得多,你还保留着纠错和回溯的空间。
2.4 记忆编码与索引设计:别迷信“1到100记忆编码大全”
很多人搜过一个词,叫“1到100记忆编码大全”,以为只要给记忆编号就能管理好。我实测下来,这种纯编号的方式在 AI Agent 场景里行不通。AI 时代需要的是语义索引,不是整数编码。
我给每条记忆设了多层标签结构。第一层是用户 ID,第二层是领域标签,第三层是实体名称,第四层是时间范围。查询时先按这些结构化条件过滤,再走向量检索。比如“本周的咖啡偏好”,向量检索可能把去年的一条“用户喜欢杯测咖啡”也捞出来,但如果加上时间范围过滤,去年的记忆直接被排除。
真正的记忆编码是一套 metadata schema。我给你看看我项目里的核心字段:
{ "memory_id": "mem_xxxx", "user_id": "usr_12345", "content": "用户偏爱极简风格的 UI", "content_hash": "sha256:...", "memory_type": "semantic", "domain": "product_design", "entities": ["UI", "极简风格"], "score": 8.5, "created_at": 1735776000, "last_access_at": 1736064000, "access_count": 3, "half_life": 604800, "source_session_id": "sess_67890" }“编码”要做的是让记忆可以被快速定位、允许被覆盖、防止重复写入,而不是给它编个序号就完事。我把 content_hash 当唯一键之一,写入前先查哈希,重复记忆直接累加 score,而不是新增一条。这样长期记忆库的膨胀速度立刻慢了一个数量级。
3. 从零到生产级:AgentScope 项目落地实操
3.1 项目结构与技术选型
这个项目我分了四个模块,结构大概是:
- agent 层:基于 AgentScope 的 Agent 定义,负责对话、工具调用和记忆读写
- memory 层:独立的记忆服务,暴露两个核心接口——remember() 和 recall()
- worker 层:后台固化任务、归档任务、衰减计算任务
- admin 层:记忆管理后台与可观测性面板
技术栈选型上,没有用特别花哨的东西。AgentScope 负责 Agent 编排,Redis 负责短期记忆和分布式锁,PostgreSQL + pgvector 负责长期记忆和向量检索,RabbitMQ 负责异步固化任务的解耦。为什么不用 MongoDB?因为这里核心数据是强关系型的,用户 ID、实体、时间范围过滤离不开关系查询,PostgreSQL 在这套场景里最省心。
整个服务的开发语言是 Python,AgentScope 原生支持 Python,这个搭配非常顺。而且 Python 生态里做向量检索的库、做异步任务的库都很成熟,团队上手成本低。
3.2 记忆服务核心实现
我先说核心接口。记忆服务对外只暴露两个方法,调用方不需要关心底层是 Redis 还是 PG。
# memory_service.py 核心接口示意 class MemoryService: def remember(self, user_id: str, session_id: str, content: str, score: float = 5.0, metadata: dict = None) -> str: # 1. 预处理:清洗、提取标签 # 2. 使用 memory_id 与 content_hash 执行去重 # 3. 写入短期记忆 # 4. 异步触发固化评估 pass def recall(self, user_id: str, query: str, top_k: int = 5) -> list[MemoryItem]: # 1. 从长期记忆向量库召回候选 # 2. 结合短期记忆 # 3. 按 combined_score = sim * 0.4 + memory_score * 0.6 排序 # 4. 截断指定 token 预算 pass这里有个细节值得单独说:remember() 是同步返回还是异步返回?我选择同步返回 memory_id,但内部的固化评估全部异步。这样调用方只需确认写入成功,固化是否立即完成不影响当前对话。如果选择同步等待固化,对话响应时间会被向量化和聚合计算拖到几百毫秒以上,生产上不可接受。
recall() 里面最重要的是阈值控制。我不允许相似度低于 0.3 的长期记忆进入候选集,这个阈值是我用测试集实测出来的。低于 0.3 的召回结果基本是噪声,强行注入 prompt 只会让 Agent 胡说。合理的阈值应该根据你的向量模型和业务领域去调,不要直接用我这里的数值,但方法是一样的:拿一批真实对话,标注哪些记忆该召回、哪些不该召回,然后画折线找拐点。
3.3 在 AgentScope 里接入记忆
AgentScope 的 Agent 可以用 pipeline 的方式组织。我在每个 Agent 的构造阶段注入 memory_service,并在 prompt 组装前后调用 recall 和 remember。
示意代码大致是这样的:
from agentscope.agent import Agent from memory_service import MemoryService class MemoryAgent(Agent): def __init__(self, name, model_config, memory_service: MemoryService): super().__init__(name, model_config) self.memory_service = memory_service def reply(self, message, user_id=None, session_id=None): # 会话开始前,把长短期记忆召回拼进 system prompt memory = self.memory_service.recall(user_id, message.content) prompt = self.build_prompt(message, memory) # 调用模型 response = super().reply(prompt) # response 完成后,把本轮关键信息写入记忆 self.memory_service.remember( user_id=user_id, session_id=session_id, content=response.content, metadata={"role": "assistant", "session_id": session_id} ) return response我刚踩过的一个坑是:把包括 system prompt、历史对话、召回记忆在内的所有东西一股脑塞进 context。第一个Demo跑通了,一旦上下文超过模型窗口,就开始报错。后来我做了预算管理,记忆注入执行硬限制。比如窗口是 32k,我留给记忆的预算只有 2k。多出来?再重要的记忆也不放,宁愿少一点,也不能把主流程挤爆。
AgentScope 里你会接触到 msg list 的管理,我建议你始终把召回记忆单独放在一个字段里,不要把记忆伪装成历史对话。原因很简单:真正调试的时候,你需要能分开看“模型看到了什么”和“模型曾经说过什么”。混在一起,出问题根本没法排查。
3.4 并发场景下的记忆一致性问题
这个点必须单独拿出来写,因为“AI Agent 怎么扛并发”是我被问得最多的问题之一。记忆系统一旦并发,坑比想象中多得多。
先看短期记忆。多会话并发时,短期记忆按 session_id 隔离,Redis 天然支持,不会串。但同一个 session 内,如果用户连续发两条消息,Agent 进程是多副本的,两个进程同时往同一个 session 的 List 里 append,顺序可能会乱。解决办法是给 session 的消息增加一个自增 sequence,写入时用 Lua 脚本保证原子性。
再看长期记忆。多用户写不同用户的记忆,不存在竞争。但同一个用户在两个设备上同时对话,Agent 可能同时写入两条互相矛盾的长期记忆。我在数据表里加了 version 字段,每次更新 memoir 时带版本号,如果版本不一致就拒绝覆盖,交给后续人工或后台策略判重。
长期记忆检索并发,倒是没什么好怕的,pgvector 走只读副本即可。真正要防的是“固化任务”和“人工修正”同时更新同一条记忆。我的做法是:所有对长期记忆的写操作都走一个 actor 模型,每个 user_id 当作一个 actor,串行处理该用户的写请求。不同用户的写操作天然并行,同一用户的写操作严格串行。这个方案很土,但极其有效,我没在这个项目里引入复杂的分布式事务。
4. 生产环境中的硬仗:性能、持久化与可观测性
4.1 记忆检索延迟优化:读多写少也要斤斤计较
记忆系统的读写比例大概是 10:1,读操作必须控制在 50ms 以内。我第一次压测时发现 recall() 慢得离谱,一查发现每次调用都在实时计算所有候选记忆的当前有效分数,然后全量排序。数据量到一万条就卡了。
优化之后,我把衰减计算改成了“读时惰性计算”,排序时只对候选集的前 50 条计算有效分数,而不是对全库计算。本来一条 SQL 要全表扫描,现在先用向量索引粗选出 50 条,再做内存排序,速度立刻降到 20ms 以内。
第二个优化是记忆聚合。如果同一个语义集群下有超过 20 条类似记忆,后台 worker 自动把它们压缩成一条综合记忆,原始记忆转入冷存储。这样既减少检索噪声,又减轻存储压力。
第三个点是缓存。高频用户的 TopK 记忆结果缓存在 Redis 里,缓存 30 秒。用户连续对话时,这期间记忆几乎不会有大变化,从缓存取即可。只有固化任务完成后才主动失效缓存。
4.2 持久化与灾难恢复
生产级系统最怕数据丢。记忆是 Agent 的核心资产,不能接受 Redis flush 之后用户画像全丢。我做了两级持久化策略:Redis 里的短期记忆开启 AOF 持久化,避免进程重启后丢失;PostgreSQL 里的长期记忆本身就是持久化的。
但 Redis 的 AOF 最多只能保证秒级恢复,万一服务器整个挂掉,依然可能丢最近一秒的短期记忆。我的取舍是:短期记忆允许丢失,因为它的价值密度低,丢了最多失去当前这轮的上下文,用户重新说一遍即可。长期记忆绝不允许丢,所以全部落 PG。
还有更细一层的设计:长期记忆的更新策略是 append-only。不是直接 update 老记录,而是新增一条新版本记忆,老版本标记 superseded。这种做法让记忆可以被回滚。假设某条记忆被错误固化,后来发现不对,系统可以沿着版本链找回原始状态。这在 Agent 生产环境里太重要了,因为模型和规则都可能误判,能回退才算真正“生产级”。
4.3 可观测性:让记忆系统可以被审计
我做过不少 AI 项目,深刻体会到 AI 系统的调试难度比传统后端高一个量级。记忆系统尤其如此,你根本没法用日志简单复现“为什么 Agent 突然记得某个不该记得的事”。所以我在项目里专门做了三块可观测性能力。
第一块是记忆写入审计日志。每一次 remember() 调用都记录完整的调用来源、操作者、原始内容、清洗后内容、最终的记忆 ID。任何一次记忆写入都能溯到源头。
第二块是 prompt 快照。每次 Agent 回复时,把完整的 system prompt(含注入记忆)和模型输出一起存下来。这样当 Agent 出现幻觉,我们能回看模型当时到底看到了哪些记忆,是记忆本身错了,还是它的推理错了。
第三块是评估指标。我设计了三个核心指标:记忆召回准确率(能否在需要时召回到正确记忆)、记忆污染率(注入记忆导致回答错误的比例)、记忆固化有效率(被固化后真正被后续检索命中的比例)。这三个指标出了监控报表之后,你再改半衰期参数、改检索阈值,就有依据了,而不是凭感觉拍脑袋。
4.4 不忘兜底:降级与熔断设计
最后再补一块生产级系统不能少的内容:降级。记忆服务哪怕挂掉,Agent 也不能挂。我在所有记忆调用外围做了一层 fallback,如果 Redis 超时超过 300ms,短期记忆读取直接降级为空;如果 PG 不可用,icall 只走 Redis 短期记忆,不查长期记忆。这样在基础设施抖动时,用户最多觉得“Agent 变笨了”,而不是“Agent 完全不可用”。
这个降级策略还有个额外的好处:上线初期你不需要一步到位把记忆系统做到无损高可用。先用降级策略保底,把核心链路跑顺了,再逐步补齐各个高可用细节,风险小得多。
5. 常见问题与排查技巧实录
5.1 记忆故障速查表
挑几个我实际遇到过,而且大概率你也会踩的问题,按“现象—原因—解法”列出来:
| 现象 | 可能原因 | 排查与解决 |
|---|---|---|
| Agent 回答张冠李戴,把 A 的偏好安到 B 身上 | 检索时没有按 user_id 做隔离过滤 | 检查 recall 查询条件,user_id 必须放进过滤条件而不是只放向量检索 |
| 同一件事用户反复提到,Agent 每次都像第一次听说 | 固化策略太严格,短期记忆没进入长期 | 加大重复提及的次数权重,触发固化门槛调低 |
| 记忆库膨胀极快,检索噪声大 | 缺少 content_hash 去重 | 写入前查询哈希,已有记忆只累加 score 和 access_count |
| 并发下长期记忆被覆盖丢失 | 更新时未做版本校验 | 增加 version 字段,更新时带乐观锁 |
| 召回结果相似度很高但内容陈旧 | 半衰期没生效,或者检索时忘记计算有效分数 | 确认 current_score 在排序卷积中被计算,而不是只拿初始 score |
| 首次使用时体验太差,Agent 什么都不懂 | 冷启动没有注入任何种子记忆 | 给新用户插入基于注册信息的默认记忆或冷启动问题清单 |
| 模型输出被注入记忆带偏,答非所问 | 记忆注入量太大、TopK 太高、阈值过低 | 减小 TopK,提示词预算上限,提升相似度阈值 |
5.2 三个我踩过的深坑
除了上面的速查表,我想把三个印象最深的坑单独展开说。
第一个是相似度阈值调太低的坑。我当时为了追求“高召回”,把向量相似度阈值调到 0.2,结果 Agent 开始把不同用户的喜好混着说,用户明确表示不满。后来我把阈值调到 0.45 并加了 user_id 硬性过滤,问题立刻消失。这个数字不代表普适标准,但“宁缺毋滥”的原则是对的。对于记忆来说,召回不到只是“笨”,召回错了是“病”。
第二个是记忆固化异步任务堆压的坑。最初固化 worker 用一个全局队列,结果一个资源池里的消息互相阻塞。后来改成按 user_id 做 key 的队列,每个用户一个分流,同时消费。瞬间队列积压就解除了。不同用户之间本来就不需要共享顺序,串行阻塞是典型的过早设计。
第三个是 memory 更新引发“雪崩”的坑。有个版本我在写入长期记忆后立即主动刷新所有缓存里的缓存结果,结果一个高活跃用户固化一条记忆,导致后续 30 分钟内所有读取都命中失败。后来我把缓存失效改成延迟 30 秒,或者在缓存里标记脏数据,下次 get 时检查是否过期。这个问题带来的教训是:缓存和记忆的关系,要以延迟一致性为标准,不要追求强一致。
5.3 Agent 记忆评测方法论
最后说说评测。很多团队做完记忆系统不知道怎么评估,只说“效果好像不错”。我觉得记忆型 Agent 的评估至少要三套数据。
第一套是单轮检索评测:给定一个用户问题和期望召回的记忆列表,评测 TopK 里包含了多少条。这套数据用来调相似度阈值和权重。第二套是多轮对话仿真:模拟用户连续几天的对话,检查 Agent 能否在第三天正确引用第一天的信息。第三套是安全与隐私测试:确认记忆系统不会把用户 A 的敏感信息泄露给用户 B,同时校验 PII 脱敏是否正常工作。
这三套数据我建议做成 CI 的一部分,每次改记忆代码都跑一遍。没有评测就去调记忆参数,跟闭眼开车没有区别。我们这个项目能稳定跑在线上,靠的不是某一次灵光一现,而是把评测和回归测试常态化。
最后再分享一点我个人的实操体会
这个项目从头做到现在,我最大的体会是:记忆系统的复杂度,90% 不在算法,而在工程细节。score 和半衰期的设计很简单,难的是去重、并发隔离、降级策略、评测回归这些“脏活累活”。但恰恰是这些脏活,决定了一个 Agent 能不能从“能跑”,变成“能扛事”。
如果你也想从零搭一个生产级记忆型 Agent,我的建议是:不要一上来就搞分布式和高可用,先在一台机器上把记忆闭环跑通——写入、召回、注入 prompt、固化、遗忘,五个环节缺一不可。跑通后再考虑多副本、缓存、异步队列。顺序不要搞反,先做出可用的系统,再让它变得可靠。
后续扩展方面,我会在这套架构上继续做多模态记忆和团队共享记忆。前者让 Agent 能记住图片、音频里的信息,后者让多个 Agent 之间共享一套组织级知识库。这个方向我觉得会是从“个人助手”走向“企业大脑”的关键一步,希望我下次写复盘的时候,已经有了新的实战结果可以分享。