1. 为什么我要从零手搓一个记忆型 AI Agent市面上开箱即用的 Agent 框架已经多到挑花眼但我还是决定从零构建一个生产级记忆型 AI Agent。原因很直接大部分教程里的 Agent 只能叫“演示级”一旦放到真实业务里多轮对话记不住上下文、并发一上来就雪崩、流式输出断断续续、人工介入没有入口这些问题不解决Agent 永远只是个玩具。这次我选的技术底座是AgentScope配合DDD 架构做工程分层用SSE做流式推送把HITLHuman-in-the-Loop作为一等公民设计进去。整套东西的目标很明确让 AI Agent 真的能“下地干活”而不是停留在 notebook 里自嗨。这篇文章适合三类人看。第一类是想从 0 到 1 搭建 AI Agent 的开发者你不需要有很深的框架经验但至少要写过 Python 或 Java 的 Web 服务。第二类是被并发和流式输出折磨过的后端SSE 断连、idle timeout、消息乱序这些坑我会一个个拆开讲。第三类是正在做 AI Agent 中台或二次开发的同学DDD 分层和记忆模块的设计思路可以直接抄作业。先把结论摆出来一个生产级记忆型 Agent 的核心不是模型多强而是记忆怎么存、上下文怎么拼、流怎么稳、人怎么插手。这四件事做好了模型换个便宜的也能跑得不错这四件事做不好用再贵的模型也是一地鸡毛。2. 整体架构设计与技术选型拆解2.1 为什么是 AgentScope 而不是自己造轮子我一开始也想过纯手写毕竟 Agent 的逻辑看起来不复杂接收消息、拼上下文、调模型、返回结果。但真动手就会发现消息抽象、工具调用、多 Agent 协作、记忆管理这些如果全自己写光是把不同模型的返回格式统一就是个大坑。AgentScope 吸引我的点在于它的消息抽象足够干净Msg对象把 role、content、工具调用结果都统一了换模型的时候业务代码基本不用动。另外它对多 Agent 对话的原生支持让我后面扩展“规划 Agent 执行 Agent”这种模式时省了很多事。不过 AgentScope 不是银弹。它的默认记忆实现偏简单直接拿来做生产级记忆是不够的所以我在它上面套了一层自己的记忆管理层。这也是我建议的做法框架负责消息流转和模型适配记忆和业务状态自己管边界清晰后面出问题也好排查。2.2 DDD 分层别把 Agent 写成一个大泥球很多人搭 Agent 的习惯是全部塞进一个agent.py几百行下去改一个 prompt 都要小心翼翼。我用 DDD 的思路把它拆成四层这里说的 DDD 不是要你搞一堆领域事件、聚合根那么重而是借用它的分层思想。层级职责典型内容接口层对外暴露 HTTP/SSE 接口Controller、SSE Emitter、鉴权应用层编排用例协调领域对象AgentService、会话管理、HITL 调度领域层核心业务逻辑记忆策略、上下文组装、工具定义基础设施层技术实现模型客户端、向量库、Redis、DB这么分的好处是当我想把记忆从“滑动窗口”换成“向量召回 摘要”时只动领域层接口层和应用层完全不用改。反过来当我要把 SSE 换成 WebSocket 时也只动接口层。变化被隔离在单层内这是生产级代码和演示级代码最大的区别。2.3 SSE 还是 WebSocket流式输出的选型逻辑流式输出这块我纠结过 SSE 和 WebSocket。最后选 SSE理由有三条。第一Agent 的输出本质是单向流服务端推、客户端收SSE 天然契合WebSocket 的双向能力用不上还增加复杂度。第二SSE 基于 HTTP走标准端口网关、负载均衡、鉴权中间件都能直接复用WebSocket 经常要在网关层做额外配置。第三SSE 的自动重连机制浏览器原生支持客户端代码简单。但 SSE 有个绕不开的坑idle timeout。就是那个经典的stream disconnected before completion: idle timeout waiting for SSE。原因是 Agent 在思考或者调工具的时候可能几十秒不吐一个字中间的反向代理或者网关就认为连接死了直接掐断。解决办法后面会详细讲核心就是心跳保活 服务端超时配置对齐。2.4 HITL让 AI 在关键节点停下来等人HITL 是我认为最容易被忽略但最重要的设计。纯自动的 Agent 在演示时很酷但生产环境里涉及资金、删除、对外发送这类操作你敢让它全自动吗我不敢。所以我在架构里把 HITL 做成一个可插拔的拦截点。Agent 在执行工具调用前先判断这个工具是否标记为“需要人工确认”。如果是就暂停执行通过 SSE 推一个确认请求给前端等用户点了确认或拒绝再继续。这个暂停不是阻塞线程而是把会话状态存下来等确认回来再恢复。这样即使确认要等几分钟也不会占着连接和线程。3. 记忆模块的核心设计与实操要点3.1 记忆不是简单的对话历史堆叠新手最容易犯的错就是把所有历史消息一股脑塞进 context。短对话没问题一旦聊到几十轮token 直接爆炸而且模型还会被无关信息干扰回答质量反而下降。我的记忆模块分三层这也是目前比较主流的生产级做法。短期记忆是最近 N 轮对话保证连贯性直接放 context。长期记忆是把历史对话做摘要或者向量化存起来需要的时候召回。工作记忆是当前任务相关的临时状态比如正在处理的订单号、用户刚提到的偏好任务结束就清掉。三层各司其职context 里永远只放“当前最相关”的内容而不是“所有内容”。3.2 上下文组装token 预算怎么算上下文组装的核心是token 预算管理。我一般这么分配系统提示词占 10%长期记忆召回占 20%短期记忆占 40%当前用户输入和工具结果占 30%。这个比例不是死的但要有意识地去控制。具体操作上我会先算当前模型的最大 context 长度比如 128k然后按比例切分。短期记忆从最近往远取取到预算用完为止。长期记忆用向量检索取 top-k 相关片段。如果加起来超了就触发摘要压缩把更早的对话压成一段摘要。注意token 计算不要用字符数除以 2 这种土办法不同模型的分词器差异很大。用对应模型的 tokenizer 算误差能控制在几个 token 内。3.3 记忆持久化Redis 加向量库的组合拳短期记忆我放 Redis因为读写快而且可以设 TTL 自动过期。长期记忆放向量库我用的是轻量级的方案存摘要和对应的向量。工作记忆也放 Redis但用单独的 key 前缀任务结束主动删。这里有个实操心得Redis 的 key 设计要带会话 ID 和用户 ID比如mem:short:{user_id}:{session_id}。这样查的时候一次命中也方便做多租户隔离。我见过有人把所有会话塞一个 list 里查的时候全量扫并发一上来直接拖垮 Redis。3.4 记忆召回的相关性判断向量召回不是召回了就完事还要做相关性过滤。我的做法是设一个相似度阈值低于阈值的直接丢掉宁可少召回也不要召回噪音。另外召回的内容要带上时间戳因为用户上周说的偏好和今天说的可能冲突时间新的优先。还有个细节召回的记忆要重新格式化再放进 context不能直接把数据库里的原始记录塞进去。我会把它包装成“根据历史对话用户曾经提到……”这种自然语言形式模型理解起来更顺。4. SSE 流式接口的完整实现与踩坑记录4.1 SSE 服务端实现的关键参数SSE 服务端的核心是保持连接、按格式推数据、正确处理断开。以 Python 的 FastAPI 为例返回一个StreamingResponsemedia_type 设成text/event-stream。from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app FastAPI() async def event_generator(session_id: str): # 先推一个心跳告诉客户端连接建立 yield event: connected\ndata: {\status\:\ok\}\n\n try: async for chunk in agent_stream(session_id): # 每条消息按 SSE 格式封装 yield fevent: message\ndata: {chunk}\n\n # 每推一条就检查是否需要心跳 except asyncio.CancelledError: # 客户端断开清理资源 cleanup(session_id) raise app.get(/agent/stream) async def stream(session_id: str): return StreamingResponse( event_generator(session_id), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 关键禁用 Nginx 缓冲 }, )这里有几个参数是血泪教训换来的。X-Accel-Buffering: no必须加否则 Nginx 会缓冲你的流客户端要等一大坨才收到流式就失去意义了。Cache-Control: no-cache防止中间层缓存。Connection: keep-alive保持长连接。4.2 心跳保活解决 idle timeout 的正解stream disconnected before completion: idle timeout waiting for SSE这个报错本质是连接空闲太久被中间层掐了。解决办法是定期发心跳。我的做法是起一个后台任务每 15 秒检查一次如果距离上次推数据超过 15 秒就推一个注释行: heartbeat\n\n。SSE 规范里以冒号开头的行是注释客户端会忽略但能保持连接活跃。async def keepalive(emitter, interval15): while True: await asyncio.sleep(interval) if emitter.idle_seconds() interval: await emitter.send_comment(heartbeat)同时服务端、网关、负载均衡的超时时间要对齐。我一般设成服务端 300 秒网关 300 秒负载均衡 300 秒。如果网关是 60 秒而服务端是 300 秒那 60 秒一到照样断。这个对齐工作不做光加心跳也没用。4.3 前端消费 SSE 的正确姿势前端用EventSource消费 SSE但要注意它只支持 GET 请求而且不能自定义 header。如果鉴权需要 token要么放 query 参数要么用 fetch ReadableStream 自己解析。const es new EventSource(/agent/stream?session_id${sid}token${token}); es.addEventListener(message, (e) { const data JSON.parse(e.data); appendToChat(data.content); }); es.addEventListener(error, (e) { // EventSource 会自动重连但要处理重连后的状态同步 console.warn(SSE error, will reconnect, e); // 重连后需要拉取断连期间的消息 syncMissedMessages(sid); }); es.addEventListener(done, () { es.close(); });这里有个坑EventSource断线重连后断连期间的消息会丢。所以我在服务端给每条消息带一个递增的 seq前端重连时带上最后收到的 seq服务端把之后的消息补推。这个机制不做用户就会遇到“回答突然少了一段”的诡异问题。4.4 并发场景下的 SSE 连接管理“AI Agent 怎么扛并发”是热词里高频出现的问题。SSE 是长连接一个用户一条连接1 万用户就是 1 万条连接。这时候连接管理就很重要。我的做法是连接和 Agent 执行解耦。Agent 的执行放到独立的 worker 里结果写到一个消息队列或者 Redis 的 stream 里。SSE 连接只负责从队列里读消息推给客户端。这样即使客户端断开Agent 的执行也不受影响重连后还能接着推。另外要限制单用户的连接数防止有人开一堆标签页把连接占满。我一般限制单用户最多 3 条 SSE 连接超了就踢掉最老的。5. HITL 人工介入的落地实现5.1 什么操作该触发人工确认不是所有操作都要人工确认那样用户体验会很差。我的判断标准是不可逆 高影响。比如删除数据、发起支付、对外发送消息这些必须确认。查询类、计算类操作直接放行。实现上我在工具定义里加一个requires_confirmation标记Agent 调用工具前先检查这个标记。这样新增工具时只要打个标不用改调度逻辑。5.2 暂停与恢复的状态管理HITL 最难的是状态管理。Agent 执行到一半要暂停这时候上下文、已执行的工具结果、待确认的操作都要存下来。我用一个PendingAction对象存这些状态序列化后放 Rediskey 是hitl:{session_id}:{action_id}。用户确认后根据 action_id 把状态捞出来继续执行。这里要注意幂等性用户可能重复点确认或者网络重试导致确认请求发两次。我的做法是确认操作带一个唯一 ID服务端处理前先检查这个 ID 是否已处理过。5.3 确认请求的推送与超时处理确认请求通过 SSE 推给前端前端弹窗让用户选择。这里要设超时比如 5 分钟没确认就自动取消并推一条消息告诉用户“操作已超时取消”。超时时间不能太长否则状态一直挂着占资源也不能太短用户可能正在看别的。5 分钟是我实测下来比较平衡的值。提示HITL 的确认请求要带上足够的上下文让用户知道自己在确认什么。只显示“是否确认执行 delete_user”是不够的要显示“即将删除用户 张三ID: 12345该操作不可恢复”。6. 常见问题排查与避坑速查6.1 SSE 相关高频问题问题现象根本原因解决办法流式输出一次性全出来Nginx 缓冲加X-Accel-Buffering: no几十秒后连接断开idle timeout心跳保活 超时对齐重连后消息丢失无 seq 机制消息带 seq重连补推中文乱码编码未指定响应头加charsetutf-8连接数暴涨未限制单用户连接限制单用户连接数6.2 记忆模块常见坑第一个坑是记忆无限增长。短期记忆如果不设上限聊得越久 context 越大最后直接超模型限制。一定要设滑动窗口大小比如最近 20 轮。第二个坑是摘要丢失关键信息。做长期记忆摘要时如果摘要太粗关键信息就丢了。我的做法是摘要时保留实体人名、订单号、金额这些是后续召回的关键。第三个坑是多会话串味。用户开了两个会话结果 A 会话的记忆跑到 B 会话里去了。这是 key 设计问题一定要用user_id session_id做隔离。6.3 并发与性能问题Agent 执行是 IO 密集型等模型返回的时候线程是空闲的。所以要用异步别用同步阻塞。Python 里用asyncioJava 里用CompletableFuture或者响应式。模型调用要做超时和重试。模型服务偶尔抽风很正常设个 30 秒超时失败重试 2 次还失败就降级返回一个兜底回复。别让一个模型调用卡死整个会话。还有个容易被忽略的点工具调用的并发。如果 Agent 一次要调多个独立工具可以并发调别串行等。我实测下来3 个工具并发调用比串行快 2 倍多。6.4 我踩过的三个真实坑第一个坑早期我没做 SSE 的 seq 机制测试时网络一抖用户就反馈“回答少了一段”。排查了半天才定位到是重连丢消息。加上 seq 后彻底解决。第二个坑HITL 的确认状态我一开始放内存单机测试没问题一上多实例就出问题——确认请求打到 A 实例但状态存在 B 实例。后来改成 Redis 共享状态才解决。任何要跨请求的状态都别放进程内存。第三个坑记忆召回我一开始没做相关性阈值结果召回一堆无关内容模型被带偏回答质量反而比不召回还差。加上阈值过滤后召回质量明显提升。召回不是越多越好是越准越好。7. 从练手项目到生产级的扩展思路如果你只是想练手把前面说的记忆三层、SSE 流式、HITL 拦截跑通就已经比 90% 的教程项目完整了。但要做成生产级还有几件事要补。可观测性。Agent 的执行链路要能追踪每次模型调用、工具调用、记忆召回都要打点。出了问题能快速定位是哪一环。我用的是 OpenTelemetry 那套trace 一拉整个链路清清楚楚。成本控制。模型调用是花钱的要做 token 用量统计和限额。单用户单日 token 超了就限流防止有人恶意刷。这个不做账单会教你做人。多模型路由。简单问题用便宜模型复杂问题用强模型。我做了个简单的路由规则根据输入长度和关键词判断能省不少成本。灰度与回滚。Prompt 改了、记忆策略改了不能直接全量上。要有灰度机制先放 5% 流量观察指标没问题再全量。出问题能一键回滚到上个版本。这套东西搭下来你会发现 Agent 的难点从来不在模型而在工程。记忆怎么管、流怎么稳、人怎么插手、并发怎么扛这些才是决定一个 Agent 能不能上生产的关键。AgentScope 给了你一个好的起点但剩下的路还得自己一步步走。
