生产级Agent记忆系统设计：AgentScope架构与SSE流式实践

1. 为什么记忆才是生产级 Agent 和玩具 Demo 的分水岭大多数人第一次搭 AI Agent跑通的流程都差不多用户提问模型思考调用一两个工具返回答案结束。这个链路在演示场景里看起来很惊艳但只要把它放到真实业务里连续跑上三天问题就会集中爆发——昨天用户明确说过的偏好今天全忘了上周处理过的同类工单又要从头问一遍多轮对话超过十几轮之后模型开始胡言乱语因为它根本不知道自己之前做过什么。这就是玩具 Demo 和生产级 Agent 之间最本质的差距记忆。不是那种把历史消息一股脑塞进 context 的伪记忆而是有分层、有生命周期、有检索策略、有遗忘机制的真记忆系统。AgentScope 这个项目之所以值得单独拿出来讲就是因为它把记忆当作一等公民来设计而不是事后打补丁。我接触 AgentScope 是从它早期版本开始的当时最直观的感受是它没有急着堆功能而是先把 Agent 的运行时骨架、消息模型、记忆抽象这几件事想清楚了。到了 2.0 版本RAG as a Service、SSE 流式输出、HITLHuman-in-the-Loop这些能力陆续补齐整个项目才真正具备生产级的形态。这篇文章面向三类人一是已经会写简单 Agent、但被记忆和状态管理折磨过的开发者二是准备从零搭一套 Agent 中台、需要选型参考的架构同学三是想通过一个完整项目把 DDD、SSE、HITL 这些工程概念串起来的学习者。我会从架构设计讲到代码落地把踩过的坑和想明白的道理都摊开说。提示本文讨论的记忆特指 Agent 运行时的上下文与长期知识管理不涉及任何模型训练层面的参数记忆。2. AgentScope 的整体架构DDD 分层到底解决了什么问题2.1 从一个文件写完到领域分层的必然演进我见过太多 Agent 项目的代码组织方式一个agent.py里塞进模型调用、工具注册、记忆读写、HTTP 接口两千行起步。这种写法在原型阶段效率极高但一旦要加一个新记忆后端、换一个模型供应商、或者接入人工审核流程改动就会像多米诺骨牌一样牵连全身。AgentScope 采用的是典型的 DDD领域驱动设计分层思路把系统切成几块职责清晰的区域分层职责典型模块领域层Agent、Message、Memory 等核心概念与规则Agent 抽象、消息模型、记忆接口应用层编排用例协调领域对象完成一次对话对话服务、工具调度、HITL 流程基础设施层模型客户端、存储、SSE 传输等具体实现LLM 适配器、向量库、消息队列接口层对外暴露的 API 与流式通道HTTP 接口、SSE 端点这么分的好处不是看起来专业而是依赖方向被固定住了领域层不依赖任何具体实现基础设施层可以随时替换。举个实际例子我一开始用内存字典存短期记忆后来要换成 Redis 做多实例共享因为记忆接口是领域层定义的抽象我只写了一个新的实现类业务代码一行没动。2.2 消息模型Agent 世界的通用语言AgentScope 里最基础也最重要的领域对象是 Message。它不只是一段文本而是一个带角色、带元数据、带工具调用信息的结构化载体。我把它理解成 Agent 世界的通用语言——模型、工具、记忆、人工审核节点之间传递的都是它。一条消息通常包含这些字段角色system / user / assistant / tool、内容、时间戳、会话 ID、以及可选的工具调用记录。为什么要把工具调用也塞进消息里因为当 Agent 调用工具后工具返回的结果必须作为一条独立消息回灌给模型模型才能基于结果继续推理。如果这一步处理不当模型就会忘记自己刚调用过工具陷入重复调用的死循环。注意工具返回消息的角色一定要和普通用户消息区分开否则模型会把工具输出误当成用户输入导致推理方向跑偏。这是新手最容易踩的坑之一。2.3 记忆抽象短期、长期、工作记忆的三层设计AgentScope 的记忆设计我拆成三层来理解这个分层不是官方强制的而是从它的接口设计里能读出来的意图短期记忆Short-term Memory当前会话的完整消息序列本质是一个有上限的滑动窗口。它的作用是维持对话连贯性超出窗口的旧消息会被压缩或丢弃。长期记忆Long-term Memory跨会话持久化的知识通常以向量形式存储通过语义检索召回。用户偏好、历史结论、领域知识都属于这一层。工作记忆Working Memory当前任务执行过程中的临时状态比如正在处理的工单 ID已经尝试过的方案。任务结束就清理。这三层的生命周期完全不同混在一起管理必然出问题。我早期就犯过一个错把用户偏好直接写进短期记忆结果会话一结束偏好就丢了下次用户还得重新说一遍。后来把偏好抽到长期记忆用向量检索召回体验立刻不一样。3. 记忆系统的落地从存储选型到检索策略3.1 短期记忆的窗口管理不是简单截断短期记忆最朴素的做法是保留最近 N 条消息但这样会丢掉早期的重要信息。比如用户在第一轮说了我对花生过敏聊到第二十轮问推荐个零食如果这条过敏信息被截掉了Agent 就可能推荐含花生的产品这是灾难性的。AgentScope 的思路是摘要 窗口结合当消息数超过阈值时把最早的一批消息交给模型做摘要摘要结果作为一条特殊的 system 消息保留在窗口头部原始消息则归档到长期记忆。这样既控制了 token 消耗又保住了关键信息。具体阈值怎么定我的经验是按 token 而不是按条数来算。一条消息可能只有十个字也可能有几千字按条数截断非常不靠谱。一般我会把窗口控制在模型上下文长度的 60% 左右留出 40% 给工具返回和模型输出。# 短期记忆窗口管理的简化示意 class ShortTermMemory: def __init__(self, max_tokens8000, summarize_threshold0.8): self.messages [] self.max_tokens max_tokens self.summarize_threshold summarize_threshold def add(self, message): self.messages.append(message) if self._current_tokens() self.max_tokens * self.summarize_threshold: self._compress_oldest() def _compress_oldest(self): # 取最早的一批消息做摘要替换为一条 system 消息 old_batch self.messages[:len(self.messages) // 2] summary self._summarize(old_batch) self.messages [summary] self.messages[len(old_batch):]这段代码是示意真实实现里摘要调用本身也要考虑失败重试和成本控制。摘要不是每轮都做而是攒够一批再做否则模型调用次数会爆炸。3.2 长期记忆的向量化什么时候写、什么时候读长期记忆的核心是写入和召回两个动作。写入时机很关键我的做法是异步写入对话结束后后台任务把这一轮的关键信息抽取出来向量化后存入向量库。为什么不实时写因为实时写会拖慢响应而且很多对话内容其实不值得长期保存需要先过滤。抽取什么内容进长期记忆我总结了几类高价值信息用户的显式偏好我喜欢简洁的回答事实性结论这个项目的部署环境是 K8s未完成的待办下周要跟进这个工单反复出现的实体人名、项目名、产品名召回策略上纯向量检索有个明显问题它擅长语义相似但不擅长精确匹配。用户问上次那个叫张三的客户向量检索可能召回一堆客户相关的内容但漏掉精确的张三。所以我在实际项目里用的是混合检索向量召回 关键词召回两路结果合并去重后再排序。检索方式优势劣势适用场景向量检索语义理解强精确匹配弱模糊意图、概念查询关键词检索精确命中无法理解同义实体名、编号、专有名词混合检索兼顾两者实现复杂度高生产环境推荐3.3 记忆的遗忘机制不删反而更贵很多人只想着怎么记住却忽略了怎么忘记。长期记忆无限增长会带来两个问题检索噪声越来越大召回质量下降存储成本持续上升。我的做法是给每条记忆打上访问计数和最后访问时间定期做衰减清理。长期没被召回、且访问计数低的记忆降权或归档。这不是简单地删数据而是让记忆系统有新陈代谢的能力。就像人脑一样不常用的记忆会逐渐模糊但不会瞬间消失。提示遗忘策略一定要可配置、可回滚。我见过有团队直接硬删除旧记忆结果用户投诉你之前明明记得的排查时发现数据已经没了非常被动。4. SSE 流式输出让 Agent 的思考过程看得见4.1 为什么 Agent 场景必须用 SSE 而不是普通请求普通 HTTP 请求是一问一答客户端发请求服务端算完返回。但 Agent 的一次响应可能要几秒甚至几十秒中间还要调用多个工具。如果用户盯着一个转圈的加载图标等半分钟体验是灾难性的。SSEServer-Sent Events解决的就是这个问题服务端可以持续向客户端推送数据片段用户能看到文字一个个流出来甚至能看到正在调用搜索工具正在整理结果这样的中间状态。这种过程可见性对 Agent 尤其重要因为它让用户知道系统在干活而不是卡死了。AgentScope 的流式设计把 Agent 的执行过程拆成多个事件类型思考中、工具调用、工具返回、最终回答。前端根据事件类型渲染不同的 UI体验非常自然。4.2 SSE 连接的生命周期与超时陷阱SSE 用起来简单但生产环境里坑不少。最常见的就是空闲超时连接建立后如果一段时间没有数据推送中间的网络设备负载均衡、反向代理会主动断开连接客户端收到类似 stream disconnected before completion: idle timeout waiting for sse 的错误。解决办法有两个方向心跳保活服务端每隔 15-30 秒推送一个注释行以冒号开头保持连接活跃。这个数据前端不解析纯粹用来占线。调整超时配置把反向代理的读超时调大但这不是根本办法因为 Agent 思考时间不可控。我一般两个都做心跳保活是标配超时配置作为兜底。心跳间隔不要设太短否则无谓地消耗连接资源也不要太长超过大多数设备的默认超时就没意义了。20 秒左右是个比较稳的经验值。// 前端消费 SSE 的典型写法 const eventSource new EventSource(/api/agent/stream?sessionIdxxx); eventSource.onmessage (event) { const data JSON.parse(event.data); switch (data.type) { case thinking: renderThinking(data.content); break; case tool_call: renderToolCall(data.tool, data.args); break; case answer: appendAnswer(data.content); break; case done: eventSource.close(); break; } }; eventSource.onerror (err) { // 断线重连逻辑注意带上 lastEventId 做断点续传 console.error(SSE error, err); };4.3 断线重连与断点续传别让用户从头再来SSE 断线是常态不是异常。移动网络切换、页面切后台、代理超时都会导致断线。如果每次断线都让用户从头开始体验会很差。标准 SSE 协议支持Last-Event-ID头客户端重连时会带上最后收到的事件 ID服务端据此从断点继续推送。要实现这个服务端必须给每个事件编号并且能根据 ID 找到后续事件。我的做法是把 Agent 的输出事件按序号缓存在 Redis 里保留最近一段时间重连时按 ID 续传。这里有个细节Agent 的执行是一次性的如果客户端断线时 Agent 还在跑重连后不能重新触发一次执行否则会重复调用工具、重复扣费。所以执行和推送要解耦——Agent 把事件写入一个流式缓冲区推送层从缓冲区读取并发送客户端重连只是换了个读取位置。5. HITL 人工介入Agent 不是全自动才叫高级5.1 什么场景必须让人来兜底纯自动的 Agent 听起来很酷但在生产环境里有些决策就是不能交给模型。比如涉及资金操作、对外发送正式通知、删除数据这类高风险动作必须有人确认。HITLHuman-in-the-Loop就是为这种场景设计的。AgentScope 的 HITL 机制我理解成暂停点Agent 执行到某个需要确认的步骤时主动挂起把当前状态和待确认内容推给人工人工审核通过后 Agent 继续拒绝则走另一条分支。这个暂停点可以是工具调用前也可以是最终输出前。我实际项目里设了三类暂停点高风险工具调用前比如发送邮件、修改订单状态置信度低于阈值时模型自己都不确定就别硬答涉及敏感信息时需要人工确认是否脱敏5.2 挂起与恢复的状态管理HITL 最难的不是暂停而是恢复。Agent 挂起时它的完整状态——对话历史、已调用工具、中间变量——都必须被序列化保存。人工审核可能几分钟也可能几小时这期间服务可能重启、实例可能扩容恢复时必须能找到原来的状态。我的做法是把 Agent 的执行状态存成一个可序列化的快照用会话 ID 步骤 ID 作为键。恢复时根据这两个 ID 加载快照从暂停点继续执行。这里要注意快照里不能存不可序列化的对象比如数据库连接、文件句柄所有外部依赖都要在恢复时重新获取。# HITL 暂停点的简化示意 class HumanApprovalGate: def __init__(self, state_store): self.state_store state_store async def request_approval(self, session_id, step_id, action, context): # 保存当前执行快照 snapshot build_snapshot(session_id, step_id, context) self.state_store.save(session_id, step_id, snapshot) # 推送待审核任务给人工 await notify_reviewer(session_id, step_id, action) return {status: pending, step_id: step_id} async def resume(self, session_id, step_id, approved): snapshot self.state_store.load(session_id, step_id) if approved: return await continue_execution(snapshot) else: return await handle_rejection(snapshot)5.3 人工审核的体验设计别让审核员抓狂HITL 不只是技术问题也是产品问题。审核员面对的是一个待确认的动作如果只给他一句是否批准他根本没法判断。所以推送的内容必须包含足够的上下文Agent 为什么要做这个动作、依据是什么、影响范围是什么。我在实际项目里给审核界面加了三个要素原始用户请求、Agent 的推理摘要、动作的预期影响。审核员看完这三样基本能在几秒内做出判断。另外审核操作要支持批准拒绝修改后批准三种因为很多时候 Agent 的方向是对的只是细节需要调整。6. 从零搭建的实操路径与踩坑清单6.1 环境准备与依赖选择搭 AgentScope 项目第一件事是把依赖理清楚。核心依赖分几块模型客户端对接各家 LLM、向量库长期记忆、缓存短期记忆和状态、Web 框架对外接口和 SSE。我的选型经验是能用成熟组件就别自己造。向量库用现成的缓存用 RedisWeb 框架用生态成熟的。唯一需要自己花心思的是模型客户端抽象层因为不同供应商的接口差异大而且经常要切换。组件推荐选择理由模型客户端统一抽象 多实现便于切换供应商和做降级向量库成熟向量数据库检索性能和维护成本可控缓存Redis短期记忆、状态快照、事件缓冲都能用Web 框架支持异步和 SSE 的框架流式输出是刚需6.2 分阶段搭建先跑通再优化我强烈建议不要一上来就追求生产级完整架构。正确的路径是分阶段第一阶段单 Agent 内存记忆 普通 HTTP 接口。目标是跑通提问-思考-回答闭环。第二阶段接入工具调用 短期记忆窗口管理。目标是让 Agent 能干活、能记住当前会话。第三阶段接入长期记忆 向量检索。目标是跨会话记忆。第四阶段SSE 流式输出 断线重连。目标是体验。第五阶段HITL 状态持久化。目标是安全可控。每个阶段都能独立运行、独立验证。我见过有人一上来就搭全套结果出了问题根本不知道是哪一层导致的排查成本极高。6.3 我踩过的几个典型坑坑一工具返回结果太长撑爆上下文。有一次接了个搜索工具返回了几万字的网页内容直接塞给模型token 瞬间超限。后来加了工具结果截断和摘要只把最相关的部分回灌。坑二记忆检索召回了错误会话的内容。早期没做会话隔离向量检索把别的用户的记忆也召回了差点造成信息泄露。后来在检索时强制加上用户 ID 过滤这是底线。坑三SSE 事件顺序错乱。并发推送时事件到达顺序和生成顺序不一致前端渲染出来的文字是乱的。解决办法是给事件编号前端按编号排序后再渲染。坑四HITL 恢复后重复执行。恢复逻辑没做幂等人工批准后 Agent 把之前的工具又调了一遍。后来在快照里记录了已完成的步骤恢复时跳过。注意这几个坑有一个共同点——都是看起来能跑但边界情况下出问题。生产级和 Demo 的差距往往就藏在这些边界里。6.4 学习路径建议如果你是从零开始学 AgentScope我的建议是先读它的消息模型和记忆接口定义这两个是理解整个项目的钥匙。然后跑通官方的最小示例再逐步往里加东西。不要一上来就啃全部源码会迷失在细节里。对于想通过这个项目练手的朋友我推荐一个渐进式任务清单先实现一个只会聊天的 Agent再给它加一个计算器工具再加短期记忆再加长期记忆最后加 SSE。每加一个能力都写测试验证边界情况。这样走一遍你对 Agent 系统的理解会比看十篇文章都深。7. 关于 Agent 中台与工程化的一些个人体会把单个 Agent 跑通只是起点真正难的是把它做成中台支撑多个业务线、多种 Agent 形态。我在实际推进中最大的体会是中台的价值不在于功能多而在于约束清晰。什么该由中台统一提供记忆、模型接入、流式传输、审核流程什么该由业务方自己决定提示词、工具集、业务逻辑这条线必须划清楚否则中台会变成什么都管、什么都管不好的大泥球。另一个体会是关于可观测性。Agent 的执行链路比普通服务复杂得多一次请求可能涉及多次模型调用、多个工具、若干次记忆检索。如果没有完整的链路追踪出了问题根本无从下手。我从项目早期就坚持给每个步骤打点记录耗时、token 消耗、工具调用结果这些数据后来成了优化的重要依据。最后说一句关于记忆的认知记忆不是越多越好而是越准越好。一个能精准召回三条相关记忆的 Agent比一个召回三十条噪声的 Agent 好用得多。检索质量、遗忘策略、写入过滤这三件事值得反复打磨。我在实际项目里花在记忆调优上的时间比花在模型选型上的时间多得多但回报也最明显——用户能明显感觉到这个 Agent 记得我。