企业官网建设流程全解析

1. 为什么我要从零手搓一个记忆型 AI Agent

先说结论：市面上大部分所谓“AI Agent 框架”，本质上只是把大模型的 API 调用包了一层壳，加上几个工具函数就敢叫自己 Agent 平台。我去年接手一个内部知识助手项目，前前后后试了四五个开源方案，踩坑踩到怀疑人生——要么记忆模块形同虚设，要么工具调用链路一断就整个崩掉，要么 SSE 推送到一半直接 idle timeout。最后我决定基于 AgentScope 从零构建一个真正能上生产的记忆型 Agent，这篇文章就是整个过程的完整复盘。

AgentScope是阿里开源的一个多智能体框架，核心解决的是“如何让多个 Agent 协同工作并且拥有可靠记忆”这个问题。它不像 LangChain 那样什么都想塞进去，也不像 AutoGen 那样偏学术实验，AgentScope 的定位很明确：生产级、可观测、支持分布式。配合DDD（领域驱动设计）的分层思路来组织代码，用SSE（Server-Sent Events）做流式输出，再通过MCP（Model Context Protocol）协议对接外部工具生态，这套组合拳打下来，整个系统的稳定性和可维护性比之前用的方案高出一个量级。

这篇文章适合谁看？如果你已经会用大模型 API 写个简单的对话机器人，但一提到“记忆管理”“工具编排”“流式推送”“多 Agent 协作”就头疼，那这篇就是写给你的。我会把每个技术选型背后的“为什么”讲清楚，把踩过的坑标出来，把能直接抄的代码和配置贴出来。全文基于我实际项目的经验，不是翻译官方文档，也不是跑个 demo 就敢写教程。

提示：本文涉及的代码示例基于 AgentScope 2.0 版本，部分 API 与 1.x 有差异，建议先确认你使用的版本。

2. 整体架构设计与技术选型拆解

2.1 为什么是 AgentScope 而不是其他框架

选框架这件事，我的判断标准就三条：记忆机制是否原生支持、工具调用是否可靠、是否支持分布式部署。LangChain 的记忆模块需要自己拼装，而且版本迭代太快，今天能跑的代码下个月就报错；AutoGen 的多 Agent 对话很优雅，但生产环境的日志和监控几乎空白。AgentScope 在这三点上的表现让我比较满意。

它的记忆模块设计得很务实——不是简单地把对话历史塞进 context window，而是分了短期记忆和长期记忆两层。短期记忆用滑动窗口管理当前会话的上下文，长期记忆通过向量数据库做语义检索。这个设计的好处是，你不需要把所有历史都塞给模型，而是根据当前 query 动态召回最相关的片段。实测下来，同样的问题，带记忆召回的准确率比纯 context 拼接高了将近 30%。

另一个让我决定用它的原因是消息传递机制。AgentScope 内部用了一套基于消息总线的通信模型，Agent 之间不直接调用，而是通过消息队列异步通信。这意味着你可以把不同的 Agent 部署在不同的进程甚至不同的机器上，通过配置消息中间件来串联。对于需要横向扩展的场景，这个设计太重要了。

2.2 DDD 分层：让 Agent 代码不再是一坨意大利面

我见过太多 Agent 项目的代码结构是这样的：一个main.py里塞了 prompt 模板、工具定义、API 调用、数据库操作、日志打印，两千行起步。这种代码跑 demo 没问题，一旦要加功能或者排查问题，基本等于重写。

所以我用DDD（领域驱动设计）的思路把整个系统拆成了四层：

• 接口层（Interface）：负责对外暴露 HTTP 接口和 SSE 流式通道，处理请求参数校验和响应格式化。这一层不包含任何业务逻辑。
• 应用层（Application）：编排用例，协调领域服务和基础设施。比如“用户提问”这个用例，需要先调用记忆服务检索历史，再调用 Agent 服务生成回复，最后通过 SSE 推送结果。
• 领域层（Domain）：核心业务逻辑所在。Agent 的行为定义、记忆的召回策略、工具的选择逻辑都在这一层。这一层不依赖任何外部框架，纯 Python 类和方法。
• 基础设施层（Infrastructure）：数据库、向量存储、消息队列、外部 API 调用的具体实现。这一层通过接口与领域层交互，方便替换。

这么拆的好处是什么？举个例子，我一开始用 Redis 做短期记忆存储，后来发现 Redis 的持久化策略在容器环境下不太可靠，想换成 PostgreSQL。因为领域层只依赖一个MemoryRepository接口，我只需要在基础设施层写一个新的 PostgreSQL 实现，改一行依赖注入的配置就完成了切换。如果代码全糊在一起，这种替换想都不敢想。

2.3 SSE 还是 WebSocket：流式输出的选型逻辑

流式输出是 Agent 体验的关键。用户提问后等 10 秒才看到完整回复，和逐字逐句实时显示，体验差距是数量级的。可选方案有两个：SSE和WebSocket。

我最终选了 SSE，理由如下：

Agent 的流式输出本质上是“服务端不断推、客户端只接收”，不需要双向通信。SSE 基于标准 HTTP，Nginx 和各类网关天然支持，部署时少了很多麻烦。WebSocket 虽然更灵活，但在这个场景下属于杀鸡用牛刀。

不过 SSE 有个坑要注意：默认的 idle timeout。很多网关和负载均衡器会在连接空闲 60 秒后自动断开，导致stream disconnected before completion错误。解决办法是在服务端定期发送心跳注释行（: heartbeat\n\n），保持连接活跃。这个后面会详细讲。

2.4 MCP 协议：工具生态的标准化接入

MCP（Model Context Protocol）是 Anthropic 提出的一个开放协议，目的是标准化大模型与外部工具的交互方式。你可以把它理解成“AI 世界的 USB 接口”——不管你是数据库、浏览器、文件系统还是某个垂直领域的软件，只要实现了 MCP Server，任何支持 MCP 的 Agent 都能直接调用。

在没有 MCP 之前，每接一个新工具就要写一套适配代码：定义函数签名、写参数校验、处理返回值、做错误映射。十个工具就是十套代码，维护成本极高。MCP 把这些标准化了：工具的描述、参数 schema、调用方式、返回格式都有统一的协议规范。

AgentScope 2.0 对 MCP 的支持比较完善，你可以通过配置文件注册多个 MCP Server，Agent 会自动发现可用的工具并根据任务需要选择调用。我目前接入了文件系统 MCP、数据库查询 MCP 和一个内部 API 网关 MCP，整个接入过程基本就是改配置文件，没写什么胶水代码。

3. 记忆系统的核心设计与实操要点

3.1 短期记忆：滑动窗口不是简单截断

短期记忆管理当前会话的上下文，最朴素的做法是“保留最近 N 轮对话”。但这么做有个问题：如果最近 N 轮里有一轮特别长（比如用户粘贴了一大段文档），就会把其他重要信息挤掉。

我的做法是基于 token 数量的动态窗口，而不是固定轮数。具体逻辑是：

1. 从最新一轮对话开始往前遍历
2. 累计 token 数量，直到接近模型 context window 的 70%
3. 保留这个范围内的所有对话
4. 如果单轮对话超过窗口的 30%，对该轮做摘要压缩

为什么留 30% 的余量？因为还要给 system prompt、工具定义、记忆召回结果留空间。实测下来，70% 给历史对话、20% 给 system prompt 和工具、10% 给召回结果，这个分配比较合理。

代码层面，我用 tiktoken 做 token 计数，核心逻辑大概是这样：

import tiktoken class ShortTermMemory: def __init__(self, max_tokens=8000, model="gpt-4"): self.max_tokens = max_tokens self.encoder = tiktoken.encoding_for_model(model) self.buffer = [] def add(self, role, content): self.buffer.append({"role": role, "content": content}) self._trim() def _trim(self): total = 0 keep_from = len(self.buffer) for i in range(len(self.buffer) - 1, -1, -1): tokens = len(self.encoder.encode(self.buffer[i]["content"])) if total + tokens > self.max_tokens * 0.7: break total += tokens keep_from = i self.buffer = self.buffer[keep_from:] def get_context(self): return self.buffer.copy()

注意：tiktoken 的编码器需要和实际使用的模型匹配。如果你用的是国产模型，token 计算方式可能不同，建议用模型厂商提供的 tokenizer 或者做一个粗略估算（中文大约 1 个字 1.5 token）。

3.2 长期记忆：向量检索的召回策略调优

长期记忆解决的是“跨会话知识保留”的问题。用户上周问过的某个技术细节，这周再问相关问题时，Agent 应该能想起来。实现方式是把历史对话做 embedding 后存入向量数据库，查询时做语义检索。

听起来简单，但召回策略的调优空间很大。我踩过的坑包括：

坑一：chunk 粒度太粗。一开始我把每轮完整对话作为一个 chunk 存入，结果检索出来的内容太长，噪音太多。后来改成按语义段落切分，每段 200-500 字，召回准确率明显提升。

坑二：没有做时间衰减。三个月前的对话和昨天的对话，在检索时权重应该不同。我加了一个时间衰减因子，越近的对话得分越高。具体公式是final_score = similarity * (0.9 + 0.1 * recency_factor)，其中recency_factor按天数指数衰减。

坑三：top_k 设太大。一开始我设 top_k=10，想着多召回一些总没错。实际上召回太多反而干扰模型判断，后来降到 top_k=3，配合一个相似度阈值（低于 0.75 的直接丢弃），效果最好。

向量数据库我选的是 Milvus 的单机版，部署简单，性能足够。如果你不想额外维护一个数据库，ChromaDB 也是个不错的选择，可以直接嵌入应用进程。

3.3 记忆的写入时机：不是每句话都值得记

什么时候把对话写入长期记忆？如果每轮对话都写，向量库会迅速膨胀，检索质量下降。我的策略是异步批量写入 + 重要性过滤：

• 每轮对话结束后，先存入一个待处理队列
• 后台任务每隔 5 分钟处理一次队列
• 处理时用一个小模型对对话做重要性评分（1-5 分）
• 只保留 3 分以上的对话写入长期记忆
• 低于 3 分的对话只保留在短期记忆中

重要性评分的 prompt 大概是这样的：

请对以下对话片段的重要性进行评分（1-5分）： 1分：闲聊、问候、无信息量的内容 2分：一般性讨论，没有明确结论 3分：包含具体技术细节或决策 4分：包含重要的业务逻辑或架构决策 5分：包含关键配置、密钥、核心算法等必须记住的信息 对话内容：{conversation} 只输出分数，不要解释。

这个策略实测下来，长期记忆的检索命中率比全量写入高了 40% 以上，而且存储成本降低了一个数量级。

4. SSE 流式推送的完整实现与避坑指南

4.1 服务端 SSE 实现：从 Flask 到 FastAPI

我一开始用 Flask 做 SSE，发现它默认是同步阻塞的，一个 SSE 连接就会占住一个 worker，并发能力极差。后来换成了 FastAPI +StreamingResponse，基于 asyncio 的异步模型，单机可以轻松支撑几百个并发 SSE 连接。

核心实现大概是这样：

from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app = FastAPI() async def event_generator(query: str): # 发送开始事件 yield f"event: start\ndata: {json.dumps({'status': 'processing'})}\n\n" # 模拟 Agent 流式输出 async for token in agent.stream_generate(query): yield f"event: message\ndata: {json.dumps({'token': token})}\n\n" # 发送结束事件 yield f"event: done\ndata: {json.dumps({'status': 'completed'})}\n\n" @app.get("/chat/stream") async def chat_stream(query: str): return StreamingResponse( event_generator(query), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" } )

几个关键点：

• media_type必须是text/event-stream
• X-Accel-Buffering: no是给 Nginx 看的，告诉它不要缓冲 SSE 响应
• 每个事件以\n\n结尾，这是 SSE 协议的格式要求
• 事件类型通过event:字段区分，方便前端做不同处理

4.2 心跳机制：解决 idle timeout 的终极方案

前面提到的stream disconnected before completion: idle timeout waiting for sse是我踩过最深的坑。Agent 在处理复杂任务时，可能 30 秒甚至更久不产生输出（比如在调用工具、检索记忆），这时候网关就会认为连接空闲，直接断开。

解决方案是心跳保活。在事件生成器里加一个定时任务，每隔 15 秒发送一个注释行：

async def event_generator_with_heartbeat(query: str): heartbeat_interval = 15 last_heartbeat = asyncio.get_event_loop().time() async for event in agent.stream_generate(query): current_time = asyncio.get_event_loop().time() if current_time - last_heartbeat > heartbeat_interval: yield ": heartbeat\n\n" last_heartbeat = current_time yield f"event: message\ndata: {json.dumps(event)}\n\n"

SSE 协议规定，以:开头的行是注释，客户端会忽略。所以心跳不会干扰正常的事件处理，但能让网关知道连接还活着。

提示：心跳间隔要小于网关的 idle timeout。常见的网关默认值是 60 秒，保险起见设 15-20 秒比较稳妥。

4.3 前端消费 SSE：EventSource 的正确用法

前端用EventSourceAPI 消费 SSE 流，基本用法很简单：

const eventSource = new EventSource('/chat/stream?query=' + encodeURIComponent(query)); eventSource.addEventListener('message', (e) => { const data = JSON.parse(e.data); appendToChat(data.token); }); eventSource.addEventListener('done', (e) => { eventSource.close(); }); eventSource.onerror = (e) => { console.error('SSE error:', e); eventSource.close(); // 可以实现重连逻辑 };

但有几个坑要注意：

坑一：EventSource 不支持自定义 header。如果你需要在请求头里带认证 token，EventSource 做不到。解决办法是把 token 放在 URL 参数里，或者用fetch+ReadableStream自己实现 SSE 解析。

坑二：浏览器对 SSE 连接数有限制。同域名下最多 6 个并发 SSE 连接（HTTP/1.1）。如果你的应用需要同时开多个流，要么升级到 HTTP/2，要么用fetch方案。

坑三：EventSource 会自动重连。这本来是好事，但如果服务端已经处理完了，客户端还在重连，就会重复请求。解决办法是在收到done事件后主动调用eventSource.close()。

5. MCP 工具接入与多 Agent 协作实战

5.1 MCP Server 的注册与工具发现

AgentScope 2.0 接入 MCP 的方式很直接，在配置文件里声明 MCP Server 的地址和认证信息就行：

mcp_servers: - name: filesystem command: npx args: ["-y", "@modelcontextprotocol/server-filesystem", "/data/workspace"] - name: database url: "http://localhost:8080/mcp" token: "${MCP_DB_TOKEN}" - name: internal_api url: "http://api-gateway:9090/mcp" token: "${MCP_API_TOKEN}"

Agent 启动时会自动连接这些 MCP Server，拉取工具列表和参数 schema。当 Agent 决定调用某个工具时，框架会自动构造符合 MCP 协议的请求并处理响应。

这里有个经验：不要一次性注册太多 MCP Server。每个 Server 的工具列表都会占用 context window，工具太多反而会让模型选择困难。我的做法是按需加载——根据当前会话的领域标签，动态激活相关的 MCP Server。比如用户问的是代码相关的问题，就只激活 filesystem 和 git 相关的 MCP。

5.2 多 Agent 协作：什么时候需要，什么时候不需要

AgentScope 支持多 Agent 协作，但我得说句实话：大部分场景不需要多 Agent。一个配置良好的单 Agent 加上工具调用，能解决 90% 的问题。多 Agent 带来的复杂度是成倍增加的——消息传递、状态同步、冲突解决，每一个都是坑。

我目前只在两个场景用了多 Agent：

场景一：需要不同角色的专业视角。比如代码审查，一个 Agent 负责检查安全性，一个负责检查性能，一个负责检查代码风格，最后汇总。这种场景下，每个 Agent 的 prompt 和工具集不同，拆开确实比一个 Agent 干所有事效果好。

场景二：需要并行处理独立子任务。比如用户要求“分析这份财报并生成图表”，可以拆成“数据提取 Agent”和“图表生成 Agent”并行工作，最后合并结果。

多 Agent 的通信我用的是 AgentScope 内置的消息总线，配置了一个 Redis 作为消息中间件。Agent 之间通过send和receive方法传递消息，框架负责序列化和路由。

5.3 工具调用的错误处理与重试策略

工具调用失败是常态——网络超时、参数错误、权限不足、返回格式不符合预期，各种问题都会遇到。如果每次失败都直接抛给用户，体验会很差。

我的处理策略是分级重试 + 降级方案：

这套策略的核心思想是：能自动恢复的自动恢复，不能自动恢复的给用户明确的指引。最怕的是那种“转圈半天然后报一个看不懂的错误”，用户完全不知道发生了什么。

6. 常见问题与排查技巧实录

6.1 SSE 连接问题速查表

6.2 记忆检索不准确的排查思路

记忆检索效果差，通常不是单一原因，而是多个环节的问题。我的排查顺序是：

1. 先看 embedding 质量：拿几个典型 query 手动算一下 embedding，看看相似度分数是否合理。如果相似度普遍偏低，可能是 embedding 模型不适合你的领域数据。
2. 再看 chunk 切分：把检索到的 chunk 打印出来，看看内容是否完整、是否有意义。如果 chunk 被切得支离破碎，调整切分策略。
3. 然后看 top_k 和阈值：召回太少可能漏掉关键信息，召回太多会引入噪音。建议从 top_k=5、阈值 0.7 开始调。
4. 最后看 prompt：召回的內容怎么塞进 prompt 也很关键。我一般会加一句“以下是与当前问题相关的历史信息，请参考但不限于此”，给模型一个明确的指引。

6.3 工具调用超时的处理经验

工具调用超时是最常见的问题之一。我的经验是给每个工具设置独立的超时时间，而不是用全局统一值。比如：

• 文件读取：5 秒
• 数据库查询：10 秒
• 外部 API 调用：30 秒
• 复杂计算任务：120 秒

超时后的处理也有讲究。如果是查询类工具超时，可以返回“查询超时，请稍后重试”；如果是操作类工具超时，需要确认操作是否已经执行（比如写文件超时，文件可能已经写入了），避免重复操作。

注意：超时时间不是越长越好。用户等待超过 30 秒就会开始焦虑，超过 60 秒大概率会刷新页面。所以对于耗时操作，更好的做法是异步执行 + 轮询结果，而不是让用户干等。

6.4 我踩过的三个印象最深的坑

坑一：向量数据库的维度不匹配。我一开始用 OpenAI 的 embedding（1536 维）建了索引，后来想换成国产模型（1024 维），结果所有检索都报错。教训是：embedding 模型一旦确定，就不要轻易换。如果非要换，必须重建整个索引。

坑二：SSE 的并发连接数限制。测试时一切正常，上线后用户一多就出现连接排队。原因是 HTTP/1.1 下同域名最多 6 个 SSE 连接。后来升级到 HTTP/2 才解决。如果你的应用需要大量并发 SSE，HTTP/2 是必须的。

坑三：MCP Server 的认证 token 过期。MCP Server 的 token 通常有有效期，过期后所有工具调用都会失败。我加了一个定时任务，每小时检查一次 token 有效性，快过期时自动刷新。这个机制上线后再也没出现过批量工具调用失败的情况。

7. 一些关于 Agent 开发的个人体会

做 Agent 开发这一年多，最大的感受是：Agent 的瓶颈不在模型能力，而在工程可靠性。模型再聪明，如果工具调用动不动超时、记忆检索经常召回无关内容、流式输出时不时断开，用户体验就是灾难。

AgentScope 给我的最大价值不是它提供了多少功能，而是它的架构设计让我能把精力放在业务逻辑上，而不是整天修框架的 bug。DDD 的分层让代码可维护，SSE 让交互流畅，MCP 让工具接入标准化，记忆系统让 Agent 真正有了“记住”的能力。

如果你也在做类似的项目，我的建议是：先把记忆系统和流式输出做扎实，再考虑多 Agent 和复杂工具编排。这两块是用户体验的基础，基础不牢，上层功能再多也是空中楼阁。

最后分享一个调试小技巧：我会在开发环境把所有 SSE 事件同时写一份到日志文件，格式是[时间戳] [事件类型] [内容摘要]。排查问题时直接看日志，比在浏览器 DevTools 里翻 Network 面板高效得多。这个习惯帮我省了大量调试时间。