记忆型AI Agent生产级落地：DDD架构、SSE流式输出与HITL实战

记忆型 AI Agent 这个词这两年几乎被用烂了但真正落到生产环境里能扛住多轮对话、能记住上下文、能在关键节点让人介入的其实没几个。我最近花了不少时间在 AgentScope 这套体系上做落地从最开始的能跑起来就行到后来被记忆丢失、SSE 断流、多 Agent 调度混乱这些问题反复教育才算摸到一点生产级的门槛。这篇就把我踩过的坑、验证过的方案、以及那些文档里不会写的细节完整地摊开讲一遍。如果你正在用 Java 技术栈做 AI Agent或者想搞清楚 DDD 架构怎么和 Agent 结合、SSE 流式输出怎么做到实时渲染还能中断、HITLHuman-in-the-Loop到底该在哪些环节插入那这篇内容应该能帮你省下不少试错时间。我会从架构分层讲到记忆机制从流式输出讲到多 Agent 协作尽量把每个为什么这么设计都说清楚。1. 先把概念理清楚Agent、LLM、AI 模型到底谁是谁很多人一上来就混淆这几个词导致后面架构设计的时候边界完全乱掉。我在带团队的时候发现只要这个概念没对齐后面写出来的代码一定是耦合的。1.1 LLM 是能力底座不是 Agent 本身LLM大语言模型本质上是一个输入文本、输出文本的概率模型。你给它一段 prompt它给你一段补全。它没有记忆、没有工具调用能力、没有状态管理甚至不知道自己上一轮说了什么。DeepSeek、GPT 系列、Claude 这些都属于 LLM 的范畴它们是模型本身。我经常用这样一个类比LLM 就像一个知识渊博但失忆的专家你每次找他咨询他都得重新听你讲一遍背景讲完他给你一个回答然后你们之间的对话就结束了下次再找他他完全不记得你。1.2 Agent 是在 LLM 之上加的一层操作系统Agent 的核心价值在于它给 LLM 装上了记忆、工具、规划能力和执行循环。一个完整的 Agent 至少包含这几个部分模型层底层调用的 LLM可以是 DeepSeek、也可以是其他模型记忆层短期对话记忆 长期知识记忆工具层让 Agent 能调用外部 API、查数据库、执行代码规划层把复杂任务拆解成可执行的步骤执行循环观察 → 思考 → 行动 → 再观察的闭环所以当有人问DeepSeek 属于哪个的时候答案很明确DeepSeek 是 LLM是模型。而 Agent 是你在 DeepSeek 外面包的那一层调度逻辑。1.3 记忆型 Agent 和普通 Agent 的分水岭普通 Agent 只维护当前会话的上下文窗口一旦超出 token 限制或者会话结束记忆就没了。记忆型 Agent 则不同它会把关键信息持久化下来在后续会话中能够召回。这个区别在生产环境里是致命的。举个实际场景用户第一天跟客服 Agent 说我对花生过敏第二天再来咨询如果 Agent 不记得这件事推荐了含花生的产品这就是事故。记忆型 Agent 要解决的就是这类跨会话、跨时间的信息保持问题。维度普通 Agent记忆型 Agent上下文范围单次会话跨会话持久化存储方式内存/上下文窗口向量库 关系库召回机制全量塞入 prompt相似度检索 摘要成本低但易丢信息高但可靠适用场景一次性问答长期陪伴、客服、助手理解了这三层关系后面讲 AgentScope 的架构设计就不会迷路。2. AgentScope 的分层设计为什么 DDD 在这里不是过度设计一开始我也觉得一个 Agent 项目搞什么 DDD是不是有点杀鸡用牛刀。但当我真正开始处理多 Agent 协作、记忆持久化、工具注册这些需求的时候才发现如果没有清晰的领域边界代码会迅速腐化成一大坨。2.1 领域层Agent 的核心行为建模在 DDD 的视角下Agent 本身就是一个聚合根。它有自己的生命周期创建、运行、暂停、销毁有自己的行为思考、行动、记忆写入也有自己的不变量比如不能在没有记忆的情况下声称记得某事。我在实际项目里把领域层拆成了几个核心实体Agent 实体持有 Agent 的身份、配置、当前状态Memory 值对象封装记忆条目包含内容、时间戳、重要性权重Tool 实体描述一个可调用的工具包含名称、参数 schema、执行逻辑Conversation 聚合管理一次完整对话的上下文这样拆的好处是当业务需求变化时比如要加一个记忆衰减机制我只需要改 Memory 这个值对象的逻辑不会波及到 Agent 的执行流程。2.2 应用层编排 Agent 的执行流程应用层负责的是用例也就是用户发起一次对话这个动作背后需要协调哪些领域对象。这里我通常会放一个 AgentOrchestrator它的职责是接收用户输入从记忆层召回相关上下文组装 prompt调用模型层处理流式输出决定是否写入新记忆判断是否需要人工介入这一层不包含业务规则只负责流程编排。业务规则都在领域层基础设施细节都在基础设施层。2.3 基础设施层模型调用、存储、SSE 的具体实现基础设施层是最脏的一层所有跟外部系统打交道的东西都在这。比如调用 DeepSeek 或其他模型的 HTTP 客户端向量数据库的读写关系型数据库的持久化SSE 连接的建立和维护工具的具体实现比如查天气、查订单我习惯在这一层做接口抽象领域层只依赖接口不依赖具体实现。这样当你要从 DeepSeek 换到别的模型时只需要换一个实现类领域层代码一行不用动。2.4 接口层对外的 API 和 SSE 端点接口层就是 Controller 那一层负责接收 HTTP 请求、返回响应。对于流式输出这里会暴露一个 SSE 端点把模型生成的内容一块一块推给前端。这四层各司其职看起来多但当你项目规模上去之后这种分层带来的可维护性是实打实的。我见过太多一开始图快、所有逻辑塞一个 Service 里的项目最后改一个功能要动十几个文件。3. 记忆机制怎么落地从上下文窗口到向量召回记忆是记忆型 Agent 的灵魂但也是最容易做砸的地方。我见过不少项目号称有记忆实际上就是把历史对话全塞进 prompttoken 一超就截断这根本不叫记忆叫临时缓存。3.1 短期记忆和长期记忆的分工短期记忆就是当前会话的上下文通常放在内存里随会话结束而销毁。长期记忆则需要持久化跨会话存在。我的做法是短期记忆维护最近 N 轮对话N 根据模型上下文窗口动态调整长期记忆把每轮对话的关键信息抽取出来存到向量库和关系库关键信息怎么抽取我一般用两种策略结合规则抽取比如用户明确说记住XXX直接标记为重要记忆模型抽取让 LLM 判断这轮对话里有没有值得长期保留的信息3.2 向量召回的实际参数调优向量召回听起来简单但参数没调好召回质量会差很多。我踩过的坑主要集中在几个地方分块大小太大会导致召回内容冗余太小会丢失上下文。我实测下来中文场景下 200-500 字一块比较合适具体要看内容密度。相似度阈值设太高会召回不到设太低会召回一堆无关内容。我一般从 0.7 开始试根据实际效果调整。召回数量不是越多越好。召回太多会挤占 prompt 空间还会引入噪声。我通常控制在 3-5 条。下面是一个我常用的召回逻辑伪代码public ListMemory recall(String query, int topK, double threshold) { float[] queryVector embeddingModel.embed(query); ListMemory candidates vectorStore.search(queryVector, topK * 2); return candidates.stream() .filter(m - m.getScore() threshold) .sorted(Comparator.comparing(Memory::getScore).reversed()) .limit(topK) .collect(Collectors.toList()); }注意这里我先召回 topK*2 再过滤是为了保证过滤后还能有足够的候选。3.3 记忆的写入时机和去重记忆写入不是每轮对话都写那样会产生大量冗余。我的策略是用户明确要求记住的立即写入模型判断有价值的异步写入重复内容做去重用向量相似度判断去重这一步很关键。我遇到过用户反复问同一个问题结果记忆库里存了十几条几乎一样的内容召回的时候全是重复的浪费 token 还降低质量。提示记忆写入建议走异步队列不要阻塞主对话流程。用户感知不到写入延迟但你能避免因为写入失败导致对话中断。3.4 记忆的衰减和更新记忆不是存进去就一劳永逸的。有些信息会过时比如用户说我下周要去北京下周过了这条记忆就没意义了。我一般会给记忆加一个时间衰减因子召回时综合考虑相似度和新鲜度。对于明确过期的记忆定期清理或者标记为失效。4. SSE 流式输出实时渲染背后的工程细节SSEServer-Sent Events是流式输出的主流方案但用好不容易。我在这上面踩的坑比记忆机制还多。4.1 为什么选 SSE 而不是 WebSocket很多人第一反应是用 WebSocket毕竟它是全双工的。但对于大模型对话这种场景SSE 其实更合适SSE 基于 HTTP不需要额外的协议升级天然支持断线重连服务端推送模型简单不需要维护双向状态浏览器原生支持 EventSourceWebSocket 的优势在于双向通信但对话场景里客户端主要就是发一次请求然后接收流式响应SSE 完全够用。4.2 流式输出的完整链路一次流式输出的链路是这样的前端发起请求建立 SSE 连接后端调用模型 API开启流式模式模型每生成一个 token后端就通过 SSE 推给前端前端收到后立即渲染生成结束后后端发送结束事件关闭连接看起来简单但每一步都有坑。4.3 那个让人头疼的 idle timeout我遇到过最典型的问题就是stream disconnected before completion: idle timeout waiting for sse。这个报错的意思是SSE 连接在完成之前断开了因为等待超时。原因通常有几个模型生成太慢超过了网关或客户端的超时时间中间有代理层代理层有自己的超时设置后端没有及时发送心跳连接被判定为空闲我的解决方案是发送心跳即使模型还没生成内容也定期发送注释行: heartbeat保持连接活跃调整超时把网关、代理、客户端的超时都调大确保覆盖最长的生成时间分块发送不要等一大段生成完再发每个 token 都发心跳这块特别重要很多超时问题都是因为连接长时间没有数据传输被判定为空闲。// 心跳发送示例 ScheduledExecutorService scheduler Executors.newSingleThreadScheduledExecutor(); scheduler.scheduleAtFixedRate(() - { if (connection.isOpen()) { connection.sendComment(heartbeat); } }, 0, 15, TimeUnit.SECONDS);4.4 配合 abort 实现中断用户点了停止生成你得真的停下来不能只是前端不显示了后端还在烧 token。实现中断的关键是前端发送一个中断信号可以是另一个 HTTP 请求后端维护一个连接 ID 到执行线程的映射收到中断信号后找到对应线程并中断这里要注意中断模型调用不一定能立即生效因为模型 API 可能不支持中断。我的做法是中断后端的处理线程让它不再往 SSE 推送内容同时关闭连接。模型那边即使还在生成结果也会被丢弃。public void abort(String connectionId) { Future? future runningTasks.get(connectionId); if (future ! null) { future.cancel(true); runningTasks.remove(connectionId); } sseConnections.get(connectionId).close(); }4.5 前端渲染的细节前端收到 SSE 消息后不要每次都重新渲染整个消息那样会闪烁。正确的做法是追加内容。另外要注意处理特殊字符和换行模型输出的内容里可能有 markdown 格式需要正确解析。5. 多 Agent 协作调度、通信和冲突处理单 Agent 能解决的问题有限复杂任务往往需要多个 Agent 协作。但多 Agent 一上复杂度是指数级上升的。5.1 多 Agent 的几种协作模式我实践下来常见的协作模式有这几种主从模式一个主 Agent 负责拆解任务分发给子 Agent 执行流水线模式多个 Agent 串行处理前一个的输出是后一个的输入辩论模式多个 Agent 对同一问题给出方案然后投票或由裁判 Agent 裁决黑板模式所有 Agent 共享一块黑板各自读写选哪种模式取决于任务类型。任务拆解清晰的用主从需要多视角的用辩论需要迭代优化的用流水线。5.2 Agent 之间的通信协议Agent 之间怎么传消息这是个设计难点。我的做法是定义一个统一的消息格式public class AgentMessage { private String from; private String to; private String type; // request, response, event private String content; private MapString, Object metadata; private long timestamp; }所有 Agent 都按这个格式收发消息这样解耦得比较彻底。metadata 里可以放各种上下文信息比如任务 ID、优先级等。5.3 调度策略和资源竞争多个 Agent 同时跑资源竞争是必然的。比如同时调用模型 API可能触发限流。我的处理方式令牌桶限流控制单位时间内的模型调用次数优先级队列重要任务优先调度超时熔断某个 Agent 卡住太久直接终止调度这块我建议一开始就做好不要等到出问题了再补。我见过因为没做限流高峰期直接把模型 API 配额打满整个系统瘫痪的案例。5.4 冲突检测和解决多 Agent 写同一份记忆的时候可能产生冲突。比如 Agent A 写用户喜欢红色Agent B 写用户喜欢蓝色。我的解决方案是引入版本号和冲突检测每条记忆带一个版本号写入时检查版本号不一致就触发冲突处理冲突处理可以是覆盖、合并或者人工介入对于关键信息我倾向于人工介入避免自动合并产生错误。6. HITL人工介入该插在哪些环节HITLHuman-in-the-Loop是生产级 Agent 的必备能力。全自动的 Agent 在关键决策上不可靠必须留人工介入的口子。6.1 哪些场景需要人工介入不是所有环节都需要人工那样效率太低。我总结了几类必须介入的场景高风险操作比如涉及资金、删除数据低置信度决策模型自己都不确定的时候合规要求某些行业规定必须人工审核异常情况Agent 陷入循环或者无法处理6.2 介入的时机设计介入时机有两种同步介入Agent 执行到某一步暂停等待人工确认异步介入Agent 先执行人工事后审核有问题再回滚同步介入适合高风险操作异步介入适合批量处理。我一般会根据操作的风险等级来选择。6.3 介入界面的设计要点人工介入的界面要让人能快速理解上下文并做出决策。我的经验是展示 Agent 的完整思考过程不只是最终结论提供足够的上下文让审核者能判断操作要简单一键通过或拒绝记录审核日志便于追溯6.4 介入后的状态恢复人工介入后Agent 需要从暂停状态恢复。这里要注意恢复时要重新加载上下文因为可能已经过了很久人工的决策要写入记忆让 Agent 学习如果人工修改了 Agent 的输出要记录差异7. 从零搭建的实操路径和避坑清单讲了这么多原理最后落到实操。如果你要从零搭一个记忆型 Agent我建议按这个顺序来。7.1 环境准备和依赖选型Java 技术栈的话我推荐框架Spring Boot 3.x Spring AI或者直接用 AgentScope Java模型调用HTTP 客户端用 WebClient支持流式向量库Milvus 或者 pgvector看团队熟悉度关系库PostgreSQL缓存Redis用于会话状态和限流依赖不要贪多够用就行。我见过引入一堆库结果版本冲突搞半天的。7.2 分阶段实现路线我的建议是分四个阶段第一阶段跑通单 Agent 对话能调用模型能返回结果第二阶段加上 SSE 流式输出实现实时渲染第三阶段接入记忆机制实现跨会话召回第四阶段多 Agent 协作 HITL每个阶段都要能独立运行和验证不要想着一步到位。7.3 我踩过的那些坑坑一上下文窗口计算错误。不同模型的 token 计算方式不一样中文和英文的 token 比例也不同。我一开始按字符数估算结果经常超限。后来改用模型自带的 tokenizer 精确计算。坑二SSE 连接泄漏。连接建立后如果异常没处理好连接不会关闭时间长了会耗尽资源。一定要在 finally 里关闭连接。坑三记忆召回污染。早期没做去重召回的内容大量重复导致模型回答质量下降。加了去重和相关性过滤后才好转。坑四多 Agent 死锁。两个 Agent 互相等待对方的消息结果都卡住。后来加了超时机制才解决。坑五HITL 状态丢失。人工介入后如果服务重启暂停的任务就丢了。后来把暂停状态持久化到数据库才解决。7.4 性能优化的几个方向模型调用缓存相同的问题缓存结果减少调用记忆召回优化加缓存避免每次都查向量库并发控制合理设置线程池大小避免资源耗尽流式输出批处理小 token 合并发送减少网络开销7.5 监控和可观测性生产环境一定要有监控。我关注的指标包括模型调用延迟和成功率SSE 连接数和断连率记忆召回命中率Agent 任务完成率和平均耗时HITL 介入频率这些指标能帮你快速定位问题。我遇到过 SSE 断连率突然升高查下来是网关配置被改了。8. 一些容易被忽略的工程细节最后补充几个细节都是我在实际项目里踩过或者见别人踩过的。8.1 模型输出的内容安全过滤模型可能输出不合适的内容生产环境一定要做过滤。我一般会在流式输出的每个 chunk 上做检测发现违规立即中断。8.2 会话隔离不同用户的会话必须严格隔离记忆不能串。我见过因为缓存 key 设计不当导致 A 用户看到了 B 用户记忆的事故。8.3 版本兼容模型 API 会升级接口可能变化。要做好版本管理避免升级导致线上故障。8.4 成本控制模型调用是花钱的要做好成本监控。我一般会设置每日预算超了就降级或者限流。8.5 测试策略Agent 的测试比普通应用难因为输出不确定。我的做法是单元测试测确定性逻辑集成测试用固定输入验证流程端到端测试用真实场景验证效果定期做回归测试确保模型升级后行为一致这套东西搭下来工作量不小但每一步都是必要的。我见过太多项目前期图快后期维护成本高到想重写。记忆型 Agent 尤其如此记忆机制一旦设计错了后面改起来牵一发动全身。如果你也在做类似的项目建议先把记忆和流式输出这两块吃透这两块是地基。多 Agent 和 HITL 可以后面再加但地基没打好上面盖什么都是危房。