做 AI Agent 做了快两年我越来越确信一件事没有记忆的 Agent 只能叫会话机器人有了记忆的 Agent 才谈得上智能体。最开始我做的聊天助手用户五分钟前刚说完的偏好换个会话就又得重新问一遍上周处理过的历史任务这周再去问它完全想不起来。说白了没有记忆的 Agent 只是一段无状态的函数离生产可用的 AI Agent 还差着十万八千里。这篇文章想聊的是我最近围绕 AgentScope 这个框架做的一个完整项目从零构建一个生产级记忆型 AI Agent。AgentScope 是阿里开源的多智能体开发框架用消息驱动的思路把 Agent 之间的通信、编排、记忆都统一了起来。我会按照自己的真实实践讲清楚记忆系统该怎么分层设计、长期记忆如何向量化落库、以及从 Demo 到生产环境要补哪些稳定性功课。如果你正在选型 Agent 框架或者想把已有的 Agent 升级成带记忆的版本这篇文章应该能给你不少可以直接抄作业的参考。1. AgentScope 到底解决了什么问题记忆型 Agent 的技术选型逻辑1.1 为什么记忆是生产级 Agent 的硬门槛先说我判断一个 Agent 是玩具还是生产级的标准它是否记得住东西。业务上Agent 要记住的东西其实就三类用户偏好比如汇报用中文简洁风格、历史任务比如昨天给运营部生成的周报模板是这个、领域事实比如公司内部系统的字段含义。任何一个缺失都会直接导致体验崩坏。拿我做的一个内部客服 Agent 举例。最初版本没有记忆用户说再帮我查一下刚才那个订单系统完全不知道刚才那个订单是哪个订单因为第二次调用是新的会话上下文清零。用户只能重新描述一整串订单号这样的 Agent 上线之后只会有一种效果被业务方退货。后来我统计过线上日志这类需要跨会话记忆的请求约占 18%而且随着使用时间推移只增不减。所以记忆不是加分项是生产级 AI Agent 的硬门槛。但记忆又是一个很容易被低估的技术域它涉及消息历史管理、长期存储、召回、冲突更新、并发隔离还要考虑隐私合规。很多人把记忆简单等同于把对话历史存下来等真正上了生产就发现根本不是那么回事。1.2 AgentScope 的核心设计理念与 2.0 的变化在这个项目里我选型的是 AgentScope。为什么没有自己手写消息管道因为 AgentScope 的底层抽象和记忆系统的天然形态非常契合它把 Agent 之间一切交互都收敛成 Message包括用户请求、助手回复、工具调用、系统指令全部是统一消息结构。这套设计对记忆型 Agent 太友好了。你可以把记忆本身就看成一串历史消息的集合短期记忆是会话内的消息序列长期记忆是经过筛选、向量化、持久化的历史消息的子集。记忆的读写本质上就是消息的新增和召回不需要为记忆单独发明一套数据模型。AgentScope 2.0 之后变化挺大。它开始支持 Java 和 Python 双语言Agent 可以以 Service 的形式独立部署并通过服务注册发现互相调用还顺势把 RAG 做成了标准服务能力RAG as Service。这对生产环境意味着长期记忆模块可以独立成一个服务不再和 Agent 进程强耦合Java 技术栈的兄弟团队也能直接复用。对我的项目来说这解决了一个很实际的架构问题——记忆库不应该是某个 Agent 实例私有的它应当是团队共享的基础设施。1.3 选型时我对比过的方案在定下来之前我也对比了另外两条路线。一条是用 LangChain 那套记忆模块快速搭另一条是自己从零写消息和记忆管理。这里不展开拉踩只说最终选择的理由。对比维度自研消息管道LangChain 记忆链AgentScope消息统一性需要自己设计基于链式调用消息形态偏弱Message 一等公民多智能体协作工作量大支持但较绕MsgHub 原生支持分布式服务化完全自研较弱2.0 原生支持RAG 能力自己接向量库需组装组件RAG as Service上手成本高低中对比结果已经很直接我需要的是一个消息驱动 多智能体 服务化的底座AgentScope 在三个维度都正中靶心。特别是它把 RAG 做成服务这个能力让长期记忆模块可以一个人独立维护不用拖拽整个 Agent 系统。当然选型不是看宣传语真正让我放心用的是它消息机制的简洁——我在周五下午花两小时跑通了第一个带记忆的 Demo比预期快很多这个体验让我确定选型没有选错。需要说明的是不同版本的 API 细节会有些差异这里分享的是实现思路和工程结构具体方法名以你当前使用的官方文档为准。2. 记忆系统分层设计短期、工作、长期记忆的边界划分2.1 三层记忆的职责边界很多教程教人做记忆上来就推向量库我的建议反过来先分清记忆的层次否则只顾长期记忆短期上下文又会乱套。我把 Agent 的记忆拆成三层——短期记忆、工作记忆、长期记忆。短期记忆是当前会话内的消息序列作用范围最小通常是模型上下文窗口内能容纳的最近 N 轮。工作记忆是这次任务进行中的状态比如已经调用了哪些工具、中间结果是什么、当前正在决策什么它不等同于对话文本更像是 Agent 的草稿纸。长期记忆是跨会话持久化的信息包括用户画像、历史事实、关键决策这一层才需要向量库。记忆层存储位置生命周期典型数据访问方式短期记忆进程内列表会话结束即清最近 N 轮对话顺序读取工作记忆进程内状态对象任务结束即清工具结果、待办状态结构化读写长期记忆外部向量库跨会话持久化用户偏好、历史任务向量召回这个划分不是过度设计。我见过一个反面案例有人把用户所有历史对话一股脑全部塞进向量库然后每次请求召回 20 条最相似的旧消息塞给模型。结果是模型经常被过时信息干扰——用户上个月说我喜欢周末处理报表这月改成都让 Agent 自动跑了旧记忆却一直在召回Agent 就反复建议周末人工处理。三层的边界清楚了处理策略就自然清晰短期记忆负责记住刚才说了什么工作记忆负责记住现在做到哪长期记忆负责记住这个用户是谁、过往发生过什么。2.2 记忆的消息格式与生命周期管理在 AgentScope 里我强烈建议你做一件事让一切记忆都以 Message 形态存在但给 Message 加上足够的路由元数据。一条记忆消息至少要有这些字段来源会话 session_id区分消息属于哪次对话时间戳 timestamps支持时间衰减和时效判断消息类型 msg_type是用户陈述、助手决策、工具事实还是系统指令主题标签 tags用于后续路由召回重要度 importance决定是否值得进入长期记忆不要小看这个 schema 设计。我第一次做的时候完全没加 session_id 和 tags结果线上多人共用记忆池A 用户说自己喜欢图表多被 B 用户召回出去Agent 对 B 也疯狂输出图表场面一度失控。生命周期管理这块每个会话结束时我会执行一个记忆整理流程把会话内的消息分成三类值得长期记忆的例如用户明确表达偏好、完成的关键任务、只保留短期的闲聊类、直接丢弃的噪音。这个筛选不是靠正则而是用一个独立的记忆提炼模型调用让它根据 schema 抽取记忆点。成本并不高但质量比规则方案好一个量级。2.3 模型侧上下文窗口的约束如何处理有了三层记忆之后每次请求向模型灌什么内容就需要精打细算。模型上下文窗口是硬约束不能无限塞。我采用的上下文拼装顺序是系统提示词 长期记忆召回片段 工作记忆摘要 短期历史消息。这里容易踩坑长期记忆片段不是越多越好。我的经验是长期记忆召回控制在 3 到 5 条、每条压缩到 100 到 200 个 token占整体预算不超过四分之一。工作记忆必须做摘要化不能让多个工具返回的大 JSON 直接堆进上下文。短期历史消息则按 token 预算滚动裁剪超过部分转成上一轮对话摘要。这套分层和预算策略让我在模型上下文只有 8k 到 16k 的时候也能稳定跑长时间会话不会因为历史太长而崩溃。等到后续换更长的上下文模型这套结构依然是安全的因为每一层都有自己的边界。3. 从零搭建记忆型 Agent项目初始化到主流程打通3.1 环境准备与依赖安装开始动手之前先把环境捋清楚。我用的是 Python 3.10装 agentscope 主包再加向量库的客户端。向量库选了哪个放到下一节说这里先跑通不带长期记忆的主流程。pip install agentscope然后初始化模型配置。AgentScope 支持多种模型后端统一走 OpenAI 兼容协议这样不管接哪家的模型代码层面基本不用动。下面的例子用的是国内大模型平台的兼容接口主要是为了说明配置方式。import agentscope agentscope.init( model_configs{ config_name: llm_main, model_type: openai, model_name: qwen-plus, api_key: 你的密钥, base_url: https://dashscope.aliyuncs.com/compatible-mode/v1, } )一点实操提示模型配置建议单独放到配置文件里不要硬编码进代码。我见过有人把 key 写死在仓库里然后提交结果就是被安全扫描盯上连夜改配置。这种低级错误在生产环境一次都不能犯。3.2 定义消息协议与记忆接口接下来我会先定义一套统一的消息封装。AgentScope 的 Message 本身自带 name、role、content我会在此基础上再加扩展字段用来给记忆模块做路由。from dataclasses import dataclass, field from typing import Optional dataclass class MemoryMsg: session_id: str role: str content: str msg_type: str # user_state / task_result / assistant_decision / system_note tags: list field(default_factorylist) importance: float 0.5 created_at: Optional[str] None同时抽象一个记忆接口这样短期记忆和长期记忆可以各自实现而 Agent 主流程不关心具体实现。class MemoryStore: def add(self, msg: MemoryMsg) - None: ... def recall(self, query: str, top_k: int 5) - list[MemoryMsg]: ... def clear_session(self, session_id: str) - None: ...这个抽象越早做越省事。我第一版图快直接在 Agent 里写死列表存储后期换向量库时改到怀疑人生。先定义接口再填实现后续想换存储引擎只动一个类。3.3 构建 Agent 主流程主流程我用的是 AgentScope 的 ReActAgent 思路也就是模型先思考再行动循环执行直到完成任务。为了结合自定义记忆我没有直接用封装好的 Agent而是基于它做了一层包装这样记忆的读写在每个决策循环里都是显式的。from agentscope.agent import ReActAgent from agentscope.message import Msg class MemoryAgent: def __init__(self, name: str, sys_prompt: str, model: str, memory: MemoryStore): self.name name self.memory memory self.react_agent ReActAgent( namename, sys_promptsys_prompt, modelmodel, tools[...], # 业务工具列表 ) def step(self, user_msg: dict, session_id: str) - dict: # 1. 短期记忆加载 history self.react_agent.memory.get_memory() # 2. 长期记忆召回 recalled self.memory.recall(user_msg[content], top_k5) # 3. 组装上下文 prompt self._build_prompt(history, recalled) # 4. 交给 ReAct Agent 执行 reply self.react_agent(Msg(nameuser, contentprompt)) # 5. 写入记忆 self._persist(session_id, user_msg, reply) return {reply: reply, recalled_context: recalled}跑通这一步之后你就拥有了一个听话的骨架能对话、能调工具、能记住当次会话里说了什么。接下来所有复杂度都集中在长期记忆怎么落库、怎么召回得准。主流程打通有一个容易被忽略的点每步都要给记忆写入留出显式的入口不要等会话结束再统一写。因为一个任务可能执行很长时间中途崩溃会导致整段记忆丢失逐步写崩溃后至少保留了已经发生的部分。4. 长期记忆落库向量化写入、召回与更新策略4.1 向量库与嵌入模型的选型长期记忆要能被想起来就不能老老实实按关键词匹配得走语义检索。我对比过几款向量库Milvus 功能全、能扛大规模数据但运维成本高适合多人用的生产环境Chroma 轻量、起步快适合小团队和原型Qdrant 是 Rust 写的性能好部署也简单单机即可。这个项目最终用的是 Qdrant因为它的并发和过滤能力足够撑住线上服务又不需要专门养一个运维团队。嵌入模型选的是 bge-m3一个多语言向量模型中文召回效果在同级别里表现不错且对长文本友好。选它还有一个分层的理由向量检索只是第一道粗筛后续我还会让模型自己读一遍召回内容所以嵌入模型不需要是天花板级召回率够用、响应够快更重要。4.2 记忆写入的完整实现长期记忆写入不能是把每条消息都塞进向量库。我设计了两道过滤第一道由记忆提炼模型判断这条信息是否值得长期保存输出一个结构化记忆点第二道做去重相同的记忆点不能无限叠加。def persist_long_term(session_id: str, conversation: list[MemoryMsg]) - None: # 1. 用模型提炼值得长期保存的记忆点 memory_points summarize_to_memory_points(conversation) for point in memory_points: # 2. 检查相似记忆是否已存在避免重复写入 dup vector_store.check_duplicate(point.content, threshold0.9) if dup: # 3. 已有则更新重要度、合并标签 vector_store.update(dup.id, mergepoint) else: vector_store.upsert( vectorembed(point.content), payload{ session_id: session_id, content: point.content, msg_type: point.msg_type, tags: point.tags, importance: point.importance, created_at: now(), }, )写入的关键是 payload 要带全元数据。很多教程只教 vector 和 id忽略了 payload等到要做时间衰减、按用户隔离、按主题过滤时就傻了只能全量拉出来在内存里筛既慢又费钱。4.3 召回优化相关度阈值、时间衰减与消息路由召回做得准长期记忆才真正有价值。我的召回管道是三段式。第一段是硬过滤按 session_id 或用户维度过滤绝对不允许跨用户召回记忆这是生产级合规的底线。第二段是向量召回用当前请求的 query 向量去检索 TopK先取相关性 score 高于阈值的结果。第三段是排序调整在向量得分的基础上叠加时间衰减因子和重要度权重。def recall(query: str, user_id: str, top_k: int 5) - list[MemoryMsg]: results vector_store.search( vectorembed(query), query_filter{user_id: user_id}, # 硬过滤 limittop_k * 3, ) ranked [] for hit in results: if hit.score 0.35: # 相关度阈值 continue age_weight decay(hit.created_at) # 时间衰减 importance_weight hit.importance ranked.append((hit, hit.score * age_weight * importance_weight)) ranked.sort(keylambda x: x[1], reverseTrue) return [hit for hit, _ in ranked[:top_k]]为什么时间衰减是必要的记忆和知识有一个共同点近期发生的、频繁确认的信息权重更高很久以前的偏好如果长时间没被提及就应当慢慢淡出召回结果。我用的衰减函数很简单按天指数衰减30 天为一个半衰期。够用且容易调参。4.4 记忆更新的冲突处理长期记忆还有一个绕不开的问题用户改主意了。用户上周说报告用 PDF 格式今天明确说以后全部用 Excel。如果旧记忆还在召回Agent 就会给出互相矛盾的答案。我的处理方案是给记忆加版本状态。当新的记忆点和旧记忆冲突时旧记忆不是被硬删而是被标记为 superseded已取代新记忆置为 active。召回时只返回 active 记忆。这样既保证了当前行为遵循最新偏好又保留了历史变更轨迹方便排查问题。def resolve_conflict(old: MemoryPoint, new: MemoryPoint) - None: if old is None: vector_store.upsert(new) return if conflict_score(old, new) 0.8: vector_store.update(old.id, payload{status: superseded}) vector_store.upsert(new)用户显式纠正的记忆点我也会给它一个很高的重要度这样它在未来很长一段时间内都会稳定压制旧偏好。这套冲突处理上线后客服侧用户为什么最近不要求 PDF 了的疑惑减少了非常多。5. 生产级改造稳定性、可观测性与并发隔离5.1 失败重试、超时与模型异常兜底Demo 能跑和线上能跑完全是两回事。第一个要解决的是模型调用的稳定性。模型 API 不像本地函数它可能超时、限流、返回 5xx。我在模型调用层包了一个统一的调用函数带指数退避重试和阶梯超时控制。async def call_llm_with_retry(agent, msg, max_retries3): for attempt in range(max_retries): try: return await asyncio.wait_for(agent.reply(msg), timeout30) except asyncio.TimeoutError: logger.warning(LLM timeout, attempt%s, attempt 1) except RateLimitError: sleep 2 ** attempt * 0.5 await asyncio.sleep(sleep) raise LLMUnavailableError()这里要特别注意重试只对幂等的请求安全。像生成订单这类有副作用的工具调用一旦超时不能盲目重试必须先查询工具执行状态否则会出现重复下单。记忆写入本身是相对幂等的但工具调用不是这个边界一定要分清楚。5.2 日志与链路追踪生产级记忆 Agent 最怕的就是出了问题不知道怎么查。Agent 每次都涉及多层调用用户请求、模型推理、记忆召回、工具执行、记忆写入。任何一层出错没有链路信息就只能逐层猜。我在入口生成一个 request_id贯穿整个调用链。然后所有关键节点都打结构化日志模型输入输出各打一条摘要、记忆召回打 rec_id 和 score、工具调用打参数和执行时长。日志统一进集中采集用 request_id 就能把一次完整的 Agent 决策过程拉出来看。记忆读写的审计日志我还会单独保留一份。这不仅是技术需要也是安全需要——当用户质疑你为什么把我这个隐私信息记下来了你能拿出完整的写入链路来核对和澄清。5.3 并发与会话隔离多用户上线之后并发隔离是最大的一场硬仗。最典型的问题就是不同用户的会话如果共用一个 Memory 实例A 用户的短期历史会串到 B 用户那里Agent 会拿 A 的偏好去回答 B 的问题。我的做法是严格按 session_id 和 user_id 双重隔离短期记忆与长期记忆。短期记忆在 Agent 实例里按 session_id 分桶每个桶独立维护历史消息长期记忆在向量检索时用 user_id 做硬过滤。Agent 实例本身无状态具体会话状态全部放到外部存储里这样 Agent 可以水平扩容。凡是把会话状态放在 Agent 实例内存里的架构在线上的并发场景一定会翻车。这个原则我踩了很多坑才真正落地Agent 可以是无状态的计算单元记忆必须是有状态的外部资产。5.4 记忆内容的隐私过滤与权限控制长期记忆有一个天然风险它会持续囤积用户的隐私信息。上线之前我花了很大精力做内容过滤。写入侧记忆提炼模型输出的记忆点要过一道 PII 检测手机号、身份证号、银行卡等明文信息直接打码或丢弃召回侧涉及隐私类型的记忆点需要额外的授权校验比如用户没有授权就不能用他的详细信息作为 Agent 的回答依据。同时必须有删除记忆的能力。用户一旦提出删除不只是删向量库里的记录还要让相关记忆从备份和文本索引中同步失效。这需要我在设计 payload 时就考虑好删除策略不能等被问到再补。生产级 Agent 的记忆能力从来不是能记就够了而是记了能管、能以受控的方式被使用。6. 实测踩坑记录那些文档里不会写的细节6.1 异步模型调用导致的事件顺序错乱第一个坑来自我图省事做异步化。为了让用户感知更快我把多个工具结果并行送给模型但异步回调里消息 append 的顺序乱了模型看到的工具结果编号和实际执行顺序对不上导致 Agent 答非所问。排查了一晚上最后发现不是模型问题是我自己的消息列表被并发写坏。解决方式是在工作记忆里显式维护工具执行结果的序号和状态把消息到达顺序和业务执行顺序解耦。教训很直接异步可以带来性能但消息排序必须自己负责不能让回调顺序决定逻辑顺序。6.2 Memory 复用导致的串话第二个坑是测试阶段就埋下的。我当时为了省资源让多个会话共用一个短期 Memory结果两个测试用户几乎同时发消息Agent 把 A 用户的问题和 B 用户的历史混在一起推理回答彻底串话。这迫使我提前把会话隔离从规划升级成强制。之后我写了内存检查脚本专门扫描有没有跨 session_id 的消息混用并且在开发环境每次跑通自动化用例时都会核对记忆归属。这个习惯一直保留到现在。6.3 向量召回在冷启动阶段的表现新用户刚接入时长期记忆库是空的向量召回什么都召不回来。如果代码写得不够健壮这个阶段 Agent 会反复把空结果拼进上下文浪费 token 还影响回复质量。后来我加了召回结果非空判断并对冷启动状态单独设置提示词模板如果对用户过往情况没有充分信息应明确询问而不是臆测。冷启动还让我意识到不能只依赖向量召回这一条路径。用户明确提出过的偏好我会同步建立一组高优先级标签索引直接走关系型查询不经过向量检索。两条召回路径合并的结果比单靠向量检索稳定得多。6.4 模型输出格式不稳定的兜底解析最后一个坑也是所有 Agent 项目都会遇到的模型偶尔不按约定的 JSON 格式输出。ReAct 流程要求模型输出思考步骤、工具调用参数一旦格式炸了整个流程就断掉。我写了解析兜底层先按严格 JSON 解析失败后做括号补全、截断处理还不行就匹配工具参数的关键字段最后兜底是让模型重新生成一次并附带上次输出格式不合法的提示。这套兜底把格式失败率从百分之几压到了千分之一以内对生产稳定性提升非常明显。现在回头看这个项目我最深的体会是做记忆型 AI Agent真正难的地方不在模型选择也不在框架 API而在记忆系统的工程可靠性。记忆是一个需要写入过滤、存储分层、召回优化、冲突更新、隐私管控五件事同时做好的系统工程每一件单独看都不难难的是把它们串进一个完整链路里。最后分享一个我坚持到现在的操作习惯在动手写 Agent 之前先把记忆 Schema 定下来再考虑 Agent 编排。Schema 是记忆系统的地基地基歪了上面再漂亮的 Agent 都会在线上垮掉。如果你也在做类似的项目欢迎带着具体场景来交流我很乐意把更细的排查过程和参数调优思路摊开来聊。