1. 为什么我要从零手搓一个记忆型 AI Agent市面上开箱即用的 Agent 框架已经多到挑花眼LangChain、AutoGPT、MetaGPT、CrewAI随便挑一个都能跑通“让大模型自己调工具”的演示。但真到了生产环境问题就来了会话一长模型开始胡言乱语用户昨天说过的偏好今天它忘得一干二净多轮工具调用之后上下文窗口直接爆掉。这些坑我在过去一年里踩了个遍。所以这个项目的出发点很朴素——从零构建一个生产级记忆型 AI Agent不依赖任何重型框架的黑盒封装把记忆管理、流式输出、工具调用、领域建模这几件事掰开揉碎自己实现一遍。技术选型上我用了AgentScope作为核心参考范式结合DDD领域驱动设计做架构分层SSEServer-Sent Events做流式输出MCPModel Context Protocol做工具接入协议。这套组合不是拍脑袋定的后面我会逐个解释为什么。这篇文章适合谁看如果你已经能跑通一个“Hello World”级别的 Agent Demo但不知道怎么做记忆管理、怎么让前端实时渲染大模型的逐字输出、怎么把工具调用抽象成可插拔的协议层那这篇就是写给你的。如果你是完全的新手也没关系我会用生活化的类比把每个概念讲清楚保证你能跟着思路走下来。提示本文所有代码示例和架构方案均基于我在实际项目中的落地经验整理参数和配置经过生产环境验证但不同业务场景仍需按需调整。2. 整体架构设计与技术选型拆解2.1 为什么是 AgentScope 而不是 LangChainAgentScope 是阿里开源的一个多智能体框架它的核心设计理念和 LangChain 有本质区别。LangChain 更像一个“胶水层”把各种模型、工具、向量库粘在一起灵活但抽象层级多调试起来经常要在好几层封装之间跳来跳去。AgentScope 则更强调消息传递和智能体编排它的消息模型是显式的每条消息都有明确的角色、内容和元数据这对做记忆管理非常友好。我举个具体例子。在 LangChain 里你想在对话历史中插入一条“系统提醒”通常要构造一个SystemMessage然后塞进ChatPromptTemplate但这条消息在后续的链式调用中怎么被处理、会不会被截断你得翻源码才知道。AgentScope 的消息模型是扁平的所有消息统一走Msg对象记忆模块可以直接遍历、过滤、摘要控制粒度细得多。另一个关键点是 AgentScope 对流式输出的原生支持。它的reply方法本身就支持流式返回不需要额外包一层 callback handler。这在做 SSE 推送的时候省了很多事。2.2 DDD 分层让 Agent 的领域逻辑不被技术细节污染很多 Agent 项目写着写着就变成了一锅粥工具调用逻辑、记忆读写逻辑、HTTP 接口逻辑全混在一个文件里。我一开始也这么干后来发现改一个记忆策略要动五个文件果断重构。DDD 的核心思想是把领域逻辑和技术基础设施分开。在我的项目里分层是这样的领域层Domain定义Agent、Memory、Tool、Message这些核心概念以及它们之间的交互规则。这一层不依赖任何外部库纯逻辑。应用层Application编排领域对象实现“用户发一条消息Agent 怎么处理”这个用例。这一层负责调用领域层的方法但不关心底层是 SSE 还是 WebSocket。基础设施层Infrastructure具体的技术实现比如用 Redis 存记忆、用 SSE 推流、用 MCP 协议接工具。接口层InterfaceHTTP Controller接收前端请求调用应用层返回 SSE 流。这么分的好处是我想把记忆存储从内存换成 Redis只需要改基础设施层的一个实现类领域层和应用层的代码一行不动。想从 SSE 换成 WebSocket也只动接口层。2.3 SSE 还是 WebSocket流式输出的选型逻辑大模型输出是逐字生成的前端要实时渲染就必须用流式传输。SSE 和 WebSocket 都能做但我最终选了 SSE原因有三第一SSE 是单向的。大模型输出场景下服务端推、客户端收天然就是单向的。WebSocket 的双向能力在这里是浪费还增加了连接管理的复杂度。第二SSE 基于 HTTP。这意味着我可以直接复用现有的鉴权、网关、负载均衡设施不需要额外开端口、配协议升级。运维成本低很多。第三SSE 自带重连机制。浏览器原生的EventSource在连接断开后会自动重连虽然生产环境我建议自己实现重连逻辑但这个默认行为在开发阶段省了不少事。当然 SSE 也有坑。比如默认的idle timeout问题——如果服务端超过一定时间没推数据连接会被中间层掐断。这个后面在问题排查章节我会详细讲怎么解决。2.4 MCP 协议工具调用的标准化尝试MCP 是 Anthropic 提出的一个协议全称 Model Context Protocol目的是标准化大模型和外部工具之间的通信。你可以把它理解成“AI 世界的 USB 接口”——不管你是数据库、文件系统、还是某个 SaaS 服务只要实现了 MCP Server任何支持 MCP 的 Agent 都能直接调用。在没有 MCP 之前每接一个工具就要写一套适配代码参数怎么传、返回值怎么解析、错误怎么处理全是自定义的。MCP 把这些都标准化了工具的描述、参数 schema、调用方式都有统一格式。我的项目里MCP 主要用在两个地方一是接入外部工具比如文件读写、HTTP 请求二是把 Agent 自身的能力暴露成 MCP Server供其他 Agent 调用。这样整个系统就是可组合的。3. 记忆系统的核心设计与实操要点3.1 记忆不是简单的“存对话历史”很多人做 Agent 记忆就是把所有对话消息塞进一个列表每次请求全量发给大模型。这在 Demo 阶段没问题但生产环境会撞上两个墙上下文窗口限制和成本。GPT-4 的上下文窗口是 128K token听起来很大但如果你每轮对话都全量发送几十轮之后就会超。而且 token 是要花钱的全量发送意味着每轮都在为历史对话重复付费。我的记忆系统分了三层短期记忆Short-term Memory最近 N 轮对话的原始消息直接参与上下文构建。N 的取值取决于模型窗口大小和单条消息的平均长度我一般设 10 到 20 轮。长期记忆Long-term Memory超出短期窗口的历史对话经过摘要压缩后存储。摘要不是简单的截断而是用大模型生成一段保留关键信息的浓缩文本。实体记忆Entity Memory从对话中抽取的关键实体和属性比如用户的姓名、偏好、正在处理的任务 ID 等。这部分用结构化存储检索效率高。3.2 记忆摘要的触发时机与 Prompt 设计摘要什么时候触发我的策略是滑动窗口 阈值触发。当短期记忆的消息数量超过阈值比如 20 条就把最旧的 10 条拿出来做摘要摘要结果追加到长期记忆然后从短期记忆中移除这 10 条。摘要的 Prompt 设计很关键。我试过几种方案最终稳定下来的版本是这样的SUMMARY_PROMPT 你是一个对话摘要助手。请将以下对话历史压缩成一段简洁的摘要要求 1. 保留所有关键事实、决策和用户偏好 2. 保留未完成的任务和待办事项 3. 丢弃寒暄、重复确认等无信息量的内容 4. 摘要长度控制在 200 字以内 对话历史 {history} 摘要这里有几个细节值得说。第一明确要求“保留未完成的任务”因为很多 Agent 场景下用户会分多轮完成一个任务如果摘要丢了任务状态Agent 就会“失忆”。第二限制摘要长度防止摘要本身膨胀成新的上下文负担。第三用“丢弃寒暄”来引导模型做信息过滤实测比不写这条效果好很多。3.3 实体记忆的抽取与存储实体记忆是我觉得最有价值的一层。举个例子用户说“帮我订一张明天去北京的机票”这里面的实体是目的地北京、时间明天、意图订机票。如果这些信息只存在原始对话里下一轮用户说“改成后天”Agent 需要从历史里找到“明天”这个信息才能理解“改成”的含义。但如果实体记忆里已经存了时间明天更新一下就行。抽取实体我用的是 function calling 的方式让大模型输出结构化的 JSONEXTRACT_PROMPT 从以下对话中抽取关键实体以 JSON 格式返回。 字段包括intent用户意图、entities实体列表每个实体包含 type、value、confidence。 如果没有可抽取的实体返回空列表。 对话{message} JSON存储上我用 Redis 的 Hash 结构key 是entity:{session_id}field 是实体类型value 是实体值和置信度。检索的时候直接HGETALLO(1) 复杂度比向量检索快得多。注意实体抽取会增加一次额外的模型调用延迟大概在 200-500ms。如果对延迟敏感可以异步做不阻塞主流程。3.4 记忆检索的优先级策略当用户发来一条新消息Agent 需要决定“回忆”哪些记忆。我的策略是短期记忆全量加载这是最近上下文必须要有。实体记忆按 session_id 加载这是当前会话的状态必须有。长期记忆按相关性检索用向量相似度找 top-3 条摘要拼接到系统提示里。这里有个坑长期记忆的向量检索需要 embedding 模型每次检索都要调一次 API延迟不低。我的优化是缓存 embedding 结果同一段摘要的向量只算一次存 Redis下次直接取。4. 基于 SSE 的流式输出完整实现4.1 SSE 协议格式与后端实现SSE 的协议格式很简单就是纯文本流每条消息以data:开头以两个换行符结尾data: {content: 你} data: {content: 好} data: {content: }后端实现上我用的是 Java 的 Spring WebFlux返回FluxServerSentEventString。核心代码如下GetMapping(value /chat/stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxServerSentEventString chatStream(RequestParam String sessionId, RequestParam String message) { return agentService.process(sessionId, message) .map(chunk - ServerSentEvent.Stringbuilder() .id(UUID.randomUUID().toString()) .event(message) .data(JSON.toJSONString(chunk)) .build()) .concatWith(Flux.just(ServerSentEvent.Stringbuilder() .event(done) .data([DONE]) .build())); }这里有几个关键点。第一produces必须指定text/event-stream否则浏览器不会按 SSE 解析。第二每条消息带一个id方便前端做断点续传。第三流结束时发一个done事件前端收到后关闭连接。4.2 前端实时渲染与 Abort 控制前端用EventSource接收const eventSource new EventSource(/chat/stream?sessionId${sid}message${encodeURIComponent(msg)}); eventSource.onmessage (event) { const data JSON.parse(event.data); appendToChat(data.content); }; eventSource.addEventListener(done, () { eventSource.close(); }); eventSource.onerror (err) { console.error(SSE error, err); eventSource.close(); // 触发重连逻辑 };Abort 控制是生产环境必须的。用户点了“停止生成”你得真的把后端的流断掉不然 token 还在烧。我的做法是前端调一个/chat/abort?sessionIdxxx接口后端用一个ConcurrentHashMapString, Disposable存每个会话的订阅abort 的时候调disposable.dispose()。private final MapString, Disposable activeStreams new ConcurrentHashMap(); public FluxServerSentEventString chatStream(String sessionId, String message) { return Flux.create(sink - { Disposable disposable agentService.process(sessionId, message) .subscribe( chunk - sink.next(buildEvent(chunk)), sink::error, () - { sink.next(buildDoneEvent()); sink.complete(); } ); activeStreams.put(sessionId, disposable); sink.onCancel(() - { disposable.dispose(); activeStreams.remove(sessionId); }); }); } PostMapping(/chat/abort) public void abort(RequestParam String sessionId) { Disposable d activeStreams.remove(sessionId); if (d ! null !d.isDisposed()) { d.dispose(); } }4.3 心跳保活与 idle timeout 问题SSE 最大的坑就是 idle timeout。大模型在“思考”的时候比如调工具、做检索可能好几秒不输出任何内容。中间层的 Nginx、网关、负载均衡器一看没数据就把连接掐了。前端表现就是“stream disconnected before completion”。解决方案是心跳保活。服务端每隔 15 秒发一个注释行以:开头SSE 协议规定注释行会被客户端忽略但能保持连接活跃FluxServerSentEventString heartbeat Flux.interval(Duration.ofSeconds(15)) .map(i - ServerSentEvent.Stringbuilder().comment(keepalive).build()); return Flux.merge(heartbeat, actualStream);Nginx 那边也要配一下把proxy_read_timeout调大location /chat/stream { proxy_pass http://backend; proxy_read_timeout 300s; proxy_buffering off; proxy_cache off; chunked_transfer_encoding on; }proxy_buffering off很关键不关的话 Nginx 会缓冲响应流式效果就没了。5. MCP 工具接入与 Agent 能力扩展5.1 MCP Server 的实现要点MCP 协议的核心是三个概念Resources资源只读数据、Tools工具可执行操作、Prompts提示模板。我的项目里主要用 Tools。一个 MCP Server 的实现本质上就是暴露一个 HTTP 接口接收标准格式的请求返回标准格式的响应。请求体大概长这样{ jsonrpc: 2.0, method: tools/call, params: { name: read_file, arguments: { path: /data/report.txt } }, id: 1 }响应{ jsonrpc: 2.0, result: { content: [ { type: text, text: 文件内容... } ] }, id: 1 }Agent 侧要做的事情是把 MCP Server 的工具列表拉过来转换成大模型能理解的 function schema然后在模型决定调用工具时转发请求到对应的 MCP Server。5.2 工具调用的错误处理与重试工具调用失败是常态。网络超时、参数错误、权限不足各种情况都有。我的处理策略是参数错误直接把错误信息返回给模型让模型自己修正参数重试。实测模型在收到“参数 path 不能为空”这样的反馈后80% 的情况能自己改对。网络超时自动重试 2 次间隔 1 秒。如果还失败返回“工具暂时不可用”给模型。权限不足不重试直接返回错误因为重试也不会改变权限。这里有个经验不要把原始的技术错误信息直接抛给模型。比如java.net.SocketTimeoutException: Read timed out模型看不懂。要转成自然语言“工具调用超时请稍后重试或换一种方式。”5.3 工具编排与并行调用复杂任务往往需要多个工具配合。比如“帮我查一下北京明天的天气然后根据天气推荐穿搭”这需要先调天气工具再调穿搭推荐工具。串行调用延迟高并行调用又可能有依赖关系。我的做法是让模型自己决定调用顺序。在 function calling 的 schema 里每个工具的描述都写清楚“这个工具需要什么输入、输出什么”模型会根据依赖关系自己排。实测 GPT-4 级别的模型在工具编排上表现不错基本不需要人工干预。对于确实可以并行的工具调用我在应用层做了一个简单的 DAG 调度解析模型返回的多个 tool_call如果它们之间没有参数依赖就并行执行否则串行。6. 常见问题与排查技巧实录6.1 SSE 连接频繁断开怎么办这是被问得最多的问题。排查顺序如下排查项检查方法解决方案Nginx 缓冲看响应是否一次性到达proxy_buffering off超时设置看断开时间是否固定调大proxy_read_timeout心跳缺失看空闲时段是否有数据加 15 秒心跳客户端重连看是否自动重连实现指数退避重连网关限制看是否有连接数限制调整网关配置我遇到过一次诡异的情况本地开发一切正常部署到生产后每 60 秒必断。查了半天发现是负载均衡器的默认 idle timeout 是 60 秒加心跳后解决。6.2 记忆摘要导致信息丢失摘要是有损压缩一定会丢信息。我的应对策略是保留原始对话的引用。摘要里如果提到某个关键决策附上原始消息的 ID需要的时候可以回溯。另外摘要的 Prompt 要定期迭代。我每个月会抽 50 条摘要做人工评估看有没有丢关键信息然后调整 Prompt。这个工作很枯燥但效果显著。6.3 工具调用陷入死循环模型有时候会反复调用同一个工具比如一直查天气但就是不生成最终回答。我的解决方案是设置最大工具调用轮数比如 5 轮。超过 5 轮还没生成最终回答就强制让模型基于已有信息作答。int maxToolRounds 5; int round 0; while (round maxToolRounds) { Response resp model.chat(messages); if (resp.hasToolCall()) { executeToolCall(resp.getToolCall()); round; } else { return resp.getContent(); } } // 超过轮数强制生成 messages.add(new Message(system, 请基于已有信息直接回答不要再调用工具。)); return model.chat(messages).getContent();6.4 实体记忆的置信度阈值怎么定实体抽取不可能 100% 准确所以每条实体我带一个 confidence 分数。检索的时候低于阈值的实体不参与上下文构建。阈值我设的是 0.7低于这个值的实体宁可不用的也不要误导模型。这个阈值需要根据业务调。如果业务对准确性要求极高比如金融场景可以调到 0.85如果对召回率要求高比如客服场景可以降到 0.6。7. 一些踩坑之后的个人体会这个项目从第一行代码到生产可用前后花了大概三个月。最大的体会是Agent 的难点不在模型在工程。模型能力已经足够强了但怎么管理上下文、怎么保证流式输出的稳定性、怎么让工具调用可靠这些才是真正花时间的地方。记忆系统的设计上我走过一段弯路。一开始想做一个“万能记忆”把所有信息都存向量库检索的时候一把捞。后来发现向量检索的精度不够经常召回不相关的历史反而干扰模型。改成三层分级之后效果稳定多了。SSE 这块心跳保活是必须的不管你觉得自己的网络环境多好。我本地开发从来没遇到过断连一上生产就各种问题。现在我的原则是任何长连接都必须有心跳。MCP 协议目前还在快速演进不同实现的兼容性参差不齐。我的建议是如果你的工具生态比较简单不一定非要上 MCP自定义一套轻量的工具注册机制可能更省事。但如果你的 Agent 需要接入大量第三方工具MCP 的标准化价值就体现出来了。最后分享一个小技巧给 Agent 加一个“思考日志”。每次模型决定调用工具之前让它先输出一段简短的推理过程“我需要查天气因为用户问了穿搭建议”。这段日志不展示给用户但存在后台排查问题的时候非常有用。你能清楚地看到模型是怎么想的为什么调了这个工具而不是那个。
