1. 为什么我要从零手搓一个记忆型 AI Agent2026 年开年到现在我陆续接触了七八个 AI Agent 项目有做客服的、有做代码助手的、也有做企业内部知识问答的。说实话大部分项目在 Demo 阶段都很惊艳一旦往生产环境推问题就全暴露出来了会话一长就失忆、流式输出断断续续、用户想中途打断却停不下来、多轮对话里上下文越堆越乱。这些坑我几乎踩了个遍。后来我把目光锁定在AgentScope这个框架上决定从零构建一个生产级记忆型 AI Agent。所谓“记忆型”不是简单地把历史消息塞进 Prompt 里而是要让 Agent 具备分层记忆、记忆检索、记忆压缩和记忆持久化的能力。所谓“生产级”意味着它得扛得住并发、断线重连、人工介入、流式渲染这些真实场景的考验。这篇文章我会把整个项目的设计思路、技术选型、核心实现、踩坑记录全部摊开讲。涉及到的关键词包括AgentScope、AI Agent、DDD 架构、SSE 流式输出、HITL 人工介入。适合谁看如果你已经写过简单的 LLM 调用想往企业级 Agent 方向走或者你正在用 Java 技术栈做 AI 应用那这篇内容应该能帮你省下不少试错时间。我会尽量说人话把每个“为什么这么设计”讲清楚而不是甩一堆代码让你自己悟。2. 整体架构设计与技术选型思路2.1 为什么选 AgentScope 而不是自己造轮子一开始我也想过完全自研毕竟 Agent 的核心逻辑看起来不复杂接收输入、拼 Prompt、调模型、解析输出、存历史。但真动手写起来你会发现光是多智能体协作这一块就够喝一壶的。AgentScope 在这方面的抽象做得比较到位它把 Agent、Message、Memory、Pipeline 这些概念都做了清晰的建模而且支持分布式部署和异步消息传递。更关键的是AgentScope 2.0 版本对RAG as a Service的支持明显加强了记忆检索可以直接对接向量库不用自己从头写 embedding 和召回逻辑。我对比过 Spring AI 和 LangChain4j前者生态好但 Agent 编排能力偏弱后者 Java 味不够纯正最后综合下来还是选了 AgentScope 作为核心框架外层用 Spring Cloud 做服务治理。注意AgentScope 的 Python 版和 Java 版在 API 设计上有差异Java 版更偏向企业级集成如果你团队是 Java 技术栈直接上 Java 版会省很多事。2.2 DDD 架构怎么落到 Agent 项目里很多人觉得 DDD 是业务系统的专利做 AI Agent 用不上。我一开始也这么想直到项目里出现了“记忆”这个概念才发现 DDD 的限界上下文简直是为 Agent 量身定做的。我把整个系统拆成了四个核心域对话域负责会话生命周期、消息收发、SSE 连接管理记忆域负责短期记忆、长期记忆、记忆检索与压缩智能体域负责 Agent 的编排、工具调用、多智能体协作人工介入域负责 HITL 流程、审批节点、中断恢复每个域有独立的实体、值对象和仓储接口。比如“记忆”这个实体它有memoryId、content、embedding、importanceScore、ttl这些属性仓储接口定义在领域层具体实现放在基础设施层可以灵活切换 Redis、PostgreSQL 或者向量数据库。这样拆的好处是当我想把短期记忆从内存换成 Redis 时只需要改基础设施层的实现领域逻辑一行不用动。实测下来这种架构在后期迭代时优势非常明显。2.3 SSE 还是 WebSocket流式输出的选型对比流式输出是 AI Agent 的刚需用户不可能盯着空白屏幕等十几秒。可选方案就两个SSE 和 WebSocket。我最后选了SSE原因有三点。第一SSE 基于 HTTP天然支持断线重连浏览器端EventSource会自动重试而 WebSocket 断了就得自己写重连逻辑。第二SSE 是单向的服务端推客户端正好匹配大模型流式输出的场景不需要双向通信。第三SSE 在网关和负载均衡上的兼容性更好Nginx 配一下proxy_buffering off就能用WebSocket 还得额外处理升级握手。当然 SSE 也有坑比如默认的 idle timeout 会导致stream disconnected before completion这个后面我会专门讲怎么解决。至于“React SSE/WebSocket 轮询文件变化”这种场景如果是需要客户端频繁发消息的那 WebSocket 更合适但纯 Agent 对话场景SSE 足够了。2.4 HITL 人工介入的设计位置HITL 不是事后补的功能必须在架构设计阶段就考虑进去。我的做法是在 Agent 的执行 Pipeline 里埋“检查点”当 Agent 准备执行高风险操作比如调用外部 API、修改数据库、发送邮件时Pipeline 会暂停把当前状态序列化后推给人工审批队列。审批通过后从检查点恢复执行。这里的关键是状态快照要足够完整包括当前对话上下文、已执行的工具调用、待执行的计划。我用的是事件溯源的方式每个步骤都记事件恢复时重放事件即可。这样即使服务重启人工审批完也能接着跑。3. 记忆系统的核心实现细节3.1 分层记忆模型短期、长期与工作记忆记忆型 Agent 的核心竞争力就在记忆系统。我设计了三层记忆结构记忆类型存储介质生命周期容量用途短期记忆内存/Redis单次会话最近 20 轮维持对话连贯性工作记忆内存单次任务动态存放当前任务的中间结果长期记忆向量库关系库永久无限跨会话知识沉淀短期记忆就是最近几轮对话直接拼进 Prompt。工作记忆是 Agent 在执行复杂任务时临时存放的变量比如“用户要查的订单号是 12345”。长期记忆则是把重要信息做 embedding 后存向量库下次遇到相似问题自动召回。这里有个关键决策什么信息值得进入长期记忆。我的策略是给每条记忆算一个importanceScore由 LLM 打分分数超过阈值的才持久化。否则对话里全是“好的”“谢谢”这种废话存进去只会污染检索结果。3.2 记忆检索的向量化与召回策略长期记忆的检索用的是向量相似度。具体流程是用户输入 → embedding → 向量库 ANN 检索 → 召回 Top-K → 重排序 → 注入 Prompt。embedding 模型我选的是国产的维度 1024实测在中文场景下比 OpenAI 的 ada-002 效果好。向量库用的 Milvus单机版就够用QPS 能到几千。召回策略上我做了两层过滤。第一层是向量相似度取 Top-20。第二层是时间衰减和重要性加权公式是finalScore similarity * 0.7 importanceScore * 0.2 timeDecay * 0.1timeDecay是exp(-λ * daysSinceCreation)λ 取 0.01意思是三个月前的记忆权重会降到 0.4 左右。这样既保证了相关性又不会让陈年旧事霸占上下文。实操心得向量检索的 Top-K 不要设太大K5 到 10 就够了。召回太多反而会稀释关键信息而且 Prompt 长度暴涨成本和延迟都受不了。3.3 记忆压缩让上下文窗口不再爆炸多轮对话最大的问题是上下文越来越长。我的解决方案是滚动压缩当短期记忆超过 15 轮时把最老的 5 轮交给 LLM 做摘要生成一段 200 字以内的压缩记忆替换掉原始对话。压缩后的记忆会标记为compressedtrue检索时优先级降低但不会丢失。这样即使聊了 100 轮实际注入 Prompt 的 token 数也能控制在 4000 以内。压缩的 Prompt 我调了好几版最后稳定用的是“请用第三人称总结以下对话的关键信息保留人名、数字、决策和待办事项去掉寒暄和重复内容。”实测这个 Prompt 出来的摘要质量最稳。3.4 记忆持久化的数据库选型与表结构长期记忆的持久化我用了 PostgreSQL pgvector 的组合。选它而不是纯向量库的原因是记忆除了向量检索还需要按用户 ID、时间范围、标签做过滤这些关系型查询 pgvector 也能胜任省得维护两套存储。核心表结构大概是这样CREATE TABLE agent_memory ( id BIGSERIAL PRIMARY KEY, user_id VARCHAR(64) NOT NULL, session_id VARCHAR(64), content TEXT NOT NULL, embedding vector(1024), importance_score FLOAT DEFAULT 0.5, memory_type VARCHAR(32) DEFAULT long_term, tags JSONB, created_at TIMESTAMP DEFAULT NOW(), expires_at TIMESTAMP ); CREATE INDEX idx_memory_user ON agent_memory(user_id); CREATE INDEX idx_memory_embedding ON agent_memory USING ivfflat (embedding vector_cosine_ops);expires_at用来做 TTL有些记忆比如“用户今天心情不好”这种一周后就该自动清理。tags字段存 JSON方便按主题过滤。4. SSE 流式输出与 HITL 的工程实现4.1 Java 实现 SSE 流式输出的完整链路Java 里实现 SSESpring 提供了SseEmitter但生产环境直接用它会遇到不少问题。我的做法是基于 WebFlux 的FluxServerSentEvent来做背压处理更自然。核心链路是这样的Controller 返回FluxServerSentEventString→ Agent 执行产生 token 流 → 每个 token 包装成 SSE 事件 → 客户端EventSource接收并渲染。关键代码结构GetMapping(value /chat/stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxServerSentEventString streamChat(RequestParam String sessionId, RequestParam String message) { return agentService.execute(sessionId, message) .map(token - ServerSentEvent.builder(token).event(message).build()) .concatWith(Flux.just(ServerSentEvent.builder([DONE]).event(done).build())) .onErrorResume(e - Flux.just(ServerSentEvent.builder(e.getMessage()).event(error).build())); }前端用EventSource接收每收到一个message事件就 append 到界面上收到done事件就关闭连接。这样用户看到的就是逐字输出的效果体验比等完整响应好太多。4.2 解决 idle timeout 导致的流中断stream disconnected before completion: idle timeout waiting for SSE这个报错我遇到过好几次。原因通常是 Agent 在调用工具或者做 RAG 检索时中间有十几秒没有输出 tokenSSE 连接被网关或浏览器判定为空闲而断开。解决方案分三层第一层在 Agent 执行期间定期发送心跳事件。我设的是每 10 秒发一个event: heartbeat内容为空客户端收到后忽略即可但连接保持活跃。第二层Nginx 配置调整proxy_read_timeout 300s; proxy_send_timeout 300s; proxy_buffering off; proxy_cache off;第三层客户端EventSource本身有重连机制但重连后会丢失上下文。我的做法是在服务端维护 session 状态重连时带上lastEventId服务端从断点继续推送。注意心跳间隔不要设太短否则会产生大量无用请求。10 到 15 秒是比较合适的值。4.3 配合 Abort 实现用户主动中断用户等得不耐烦了想中断生成这个需求很常见。实现方式是前端调一个/chat/abort接口服务端根据 sessionId 找到对应的执行线程调用Disposable.dispose()或者设置一个AtomicBoolean标志位Agent 在每次循环时检查这个标志发现中断就停止生成并保存当前状态。这里有个细节中断后已经生成的部分内容要保留不能直接丢弃。我的做法是中断时把已生成的 token 拼成完整消息标记为interruptedtrue存入记忆下次对话时 Agent 知道上次被打断了可以接着聊。4.4 HITL 人工介入的检查点与恢复机制HITL 的实现依赖前面说的检查点机制。具体流程Agent 执行到敏感操作前序列化当前状态对话历史、工具调用栈、计划状态存入pending_approval表同时通过 SSE 推送event: approval_required给前端前端弹出审批界面人工确认或拒绝审批结果通过/chat/approve接口回传服务端从检查点恢复继续执行或终止状态序列化我用的是 JSON因为要跨服务传输。工具调用栈里如果有不可序列化的对象就存引用 ID恢复时重新加载。这套机制在代码助手场景特别有用Agent 想执行git push或者删文件时必须人工点确认避免误操作。5. 常见问题排查与避坑经验实录5.1 记忆检索不准的排查思路记忆检索不准通常有三个原因embedding 质量差、召回策略有问题、或者记忆本身质量低。排查顺序我一般是这样的先看召回的 Top-K 里有没有相关记忆如果没有说明 embedding 或向量库有问题如果有但没被用上说明重排序逻辑有 bug如果召回的记忆本身就是废话那就是 importanceScore 阈值设太低垃圾记忆太多。我踩过的一个坑是早期没做记忆去重同一个信息被存了十几遍检索时全是重复内容。后来加了基于内容 hash 的去重存储前先查一下有没有相似的有就更新而不是新增。5.2 SSE 连接数暴涨的处理生产环境跑了一段时间后发现 SSE 连接数只增不减最后把文件描述符耗尽了。原因是客户端异常关闭时服务端没有及时感知连接一直挂着。解决办法是加连接超时和心跳检测。服务端维护一个连接注册表每个连接记录最后活跃时间后台定时任务扫描超过 5 分钟没活跃的就主动关闭。同时客户端EventSource的onerror里要显式调用close()不能只依赖自动重连。5.3 多智能体协作时的消息顺序问题多智能体场景下多个 Agent 并发执行消息到达顺序可能乱掉。我遇到过一次Agent A 的回复还没推完Agent B 的回复就插进来了前端渲染得乱七八糟。解决方案是给每个 Agent 的消息加sequenceId前端按 sequenceId 排序后再渲染。同时服务端在推送时加一个短暂的缓冲窗口比如 100ms把同一批次的消息合并后按序推送。5.4 常见问题速查表问题现象可能原因排查方法解决方案流式输出中断idle timeout看网关日志加心跳调超时记忆检索不准embedding 差人工看召回结果换模型调阈值连接数暴涨未及时关闭看 fd 数量加超时显式 close消息顺序乱并发推送看 sequenceId加序号缓冲上下文爆炸未压缩看 token 数滚动压缩审批后不恢复状态丢失看检查点表补全序列化5.5 几个我踩过的坑和对应技巧第一个坑早期用内存存短期记忆服务一重启全没了。后来改成 Redis但 Redis 的序列化又出问题复杂对象存不进去。最后用 JSON 序列化 手动反序列化解决。第二个坑Agent 调用工具时超时没处理整个对话卡死。后来给每个工具调用加了超时和重试超时后返回友好提示而不是直接报错。第三个坑HITL 审批界面没做超时审批请求挂了一天没人处理状态一直占着。后来加了审批超时默认 30 分钟超时自动拒绝并通知用户。实操心得生产级 Agent 和 Demo 的最大区别就是异常处理。Demo 只考虑正常流程生产环境要考虑网络抖动、服务重启、用户乱操作、并发冲突。每个环节都要有兜底方案。6. 项目扩展方向与个人体会这套架构跑通之后我陆续做了几个扩展。一个是把记忆系统开放成独立的 RAG as a Service其他项目可以直接调接口存取记忆。另一个是接入了更多工具比如 Jenkins 构建触发、PLC 设备状态查询让 Agent 能真正操作物理世界。代码助手场景也试了让 Agent 协助开发规范检查效果比纯规则引擎好很多因为它能理解上下文。至于“Codex 能不能直接读取其他 AI Agent 会话内容”这种问题我的做法是通过共享记忆库来实现而不是直接读会话这样更安全也更可控。面试题里经常问“AI Agent 有哪些产品”“企业级 Java AI Agent 应用平台怎么设计”我的回答都是产品形态千变万化但底层离不开记忆、编排、流式输出、人工介入这四件事。把这四件事做扎实上层怎么变都不慌。最后分享一个小技巧Agent 的 Prompt 不要写死做成可配置的模板不同场景加载不同模板。我项目里 Prompt 模板存在数据库改完即时生效不用重启服务。这个在调优阶段特别省时间。
