企业官网建设流程全解析

做过的AI应用越多，越觉得一个反直觉的现实：大多数Agent产品翻车，不是死在“模型不够聪明”，而是死在“这个Agent是失忆的”。用户周六晚上跟Agent梳理了投资组合的逻辑，周一换个设备继续问，Agent完全想不起来你是谁、上次聊了什么。这种体验放在任何产品里都是劝退级的。

所以当团队要做AI Agent落地时，我把“记忆型”列成了第一优先级需求。项目复盘下来，我们选择基于AgentScope从零搭了一套生产级记忆型AI Agent——注意这里的“从零”不是从零手写框架，而是从零把一个能扛并发、能记住用户的Agent完整落地到线上。这篇文章就是这套项目的全景拆解：为什么选AgentScope、记忆系统怎么设计、核心消息循环怎么接、并发怎么扛、以及实测中踩到的两个典型大坑。适合有Python后端基础、正在琢磨怎么把Agent从Demo变成真正可交付系统的开发者参考。

1. 为什么是 AgentScope：从自研 800 行代码到选用框架的完整取舍

1.1 自研Agent骨架的真实代价

我第一版Agent其实是自研的，当时想法很简单：一个循环、一个消息列表、一个工具分发器，总共不到800行代码就能跑通单轮问答。但需求一迭代，事情就失控了。

最典型的问题是消息流转。模型返回的内容可能触发工具调用，工具返回结果又需要重新喂给模型，这期间上下文怎么组织、谁的消息发给谁、怎么防止工具结果把上下文撑爆，这些都要自己维护。第二个问题是状态回溯，多轮对话中间一旦出错，很难定位是哪一轮的哪个环节引入了幻觉。第三个问题是并发，Python的GIL和大模型的长耗时推理叠加在一起，自研队列方案从一开始就写着“以后要重构”。

这不是说自研不行，而是说Agent的复杂度在消息传递和状态管理这两块，跟普通后端服务完全不同。框架的核心价值不是省代码量，而是帮你把架构约束在一个成熟模型里。AgentScope对我是那个合适的约束——它把Agent之间的交互抽象成消息驱动的机制，生产者只管发消息，消费者只管处理消息，这种松耦合的设计对后续做分布式部署非常友好。

1.2 同赛道框架对比：为什么不是 LangChain 或 CrewAI

选型阶段我也对比过几个主流方案，测试结论放出来供参考：

LangChain我也认真跑过，它的生态确实大，但链条式抽象在Agent这种动态循环场景下有一个尴尬问题：每一层都包装了一层Prompt和解析逻辑，线上定位一个错误要穿越四五个抽象层，排查成本极高。CrewAI的思路适合“多个角色分工”的场景，但它的核心还是编排而非运行时，对并发、重演、故障恢复这些生产级问题覆盖得不够。AgentScope和其他框架最本质的差异在于它把Msg当成一等公民，整个系统的数据流是显式的、可重放、可观察的，这对我做服务化改造非常关键。

对比之后的判断是：考虑线上稳定性和团队后续维护成本，AgentScope 2.0的消息模型和分布式运行时更契合“生产级”这三个字。

2. 记忆是 Agent 的核心资产：先想清楚怎么存、怎么取、怎么删

2.1 记忆分类：短期窗口、长期事实与语义记忆

动手写代码之前，我花了两天时间画记忆架构图。很多人的误区是把“记忆”当成一个黑盒，什么信息都往里塞，最后要么上下文爆炸，要么检索结果全是噪音。我最后的方案是把记忆拆成三个层次：

• 短期记忆：当前会话内最近N轮的上下文，作用是把对话串起来。限制非常明确——必须控制token量，超出就滚动丢掉最旧的内容。
• 长期事实记忆：用户偏好、重要事实、历史结论。这类信息结构化程度高，用数据库存最合适，需要精确读写。
• 语义记忆：模糊的、需要“联想”的信息，比如“用户上次提到过自己关注高股息策略”，靠向量检索召回。

三者不是替代关系而是分工关系。短期记忆解决“上下文连贯”，长期事实记忆解决“跨会话持久”，语义记忆解决“我不记得具体说了什么但记得大概方向”。我见过不少项目想用向量库一把梭解决所有记忆需求，结果对话窗口照样爆，精确信息又查不到，所以分类这步省不得。

2.2 记忆存储选型与表结构设计

存储方案上我选择了Redis加PostgreSQL的组合。短期记忆放Redis，因为读写快、天然支持TTL过期；长期事实和语义记忆落PostgreSQL，先跑通业务，后续量级上来了再平滑迁移到pgvector。

Redis这边结构很简单，每个会话一个hash：

session:{session_id} -> hash user_id: 用户ID history: 最近20轮消息的JSON数组 updated_at: 最后活跃时间 TTL: 24小时

PostgreSQL那边我建了一张记忆表，核心字段如下：

CREATE TABLE memory_store ( id BIGSERIAL PRIMARY KEY, user_id VARCHAR(64) NOT NULL, agent_id VARCHAR(64) NOT NULL, memory_type VARCHAR(16) NOT NULL DEFAULT 'fact', content TEXT NOT NULL, source_session_id VARCHAR(64), embedding JSONB, hit_count INTEGER NOT NULL DEFAULT 0, status SMALLINT NOT NULL DEFAULT 1, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE INDEX idx_memory_user ON memory_store (user_id, agent_id, status);

两个容易被忽略的字段我想多说一句。一个是hit_count，每次检索命中的记忆加一，后面做记忆淘汰和权重排序都靠它。另一个是status，我故意不做物理删除，用户要求“忘掉某条记忆”时把状态置为0，既保留了审计能力，也避免了误删后无法恢复的尴尬。

2.3 写入与检索的触发时机：别让记忆库变成垃圾场

记忆系统的成败不在存储容量的上限，而在写入检索的节奏。无脑写入的结果是记忆库里一半是废话，检索出来的全是噪音。

我的策略是：模型返回结果之后，由Agent先对当轮对话做一次摘要提炼，把“值得长期记住的信息”提炼出来，然后异步写入记忆库。注意一定走异步，模型返回的链路里不能卡数据库写入，否则用户端延迟直接翻倍。

检索侧更讲究。不是每轮用户消息都要去查记忆库——如果用户只是在闲聊或者问菜谱，查历史反而会引入不相关内容干扰模型。我在Agent的意图判断阶段加了一个pre-retrieval判断，只有判断当前问题可能涉及用户历史信息或个性化内容时才触发语义检索。检索命中之后，把结果拼进system prompt的“已知记忆”区域，而不是直接混在对话上下文里，这样模型更容易区分哪些是事实来自记忆、哪些是用户本轮新说的。

3. 搭建核心流程：AgentScope 的消息循环与记忆插件接入

3.1 一个可运行的Agent骨架

AgentScope 2.0里一切围绕Msg流转，Agent之间通过消息通信。我基于当前版本的接口形态整理了一个简化版骨架，实际API请以安装的版本为准，但核心思路是稳定的：

from agentscope.agent import AgentBase from agentscope.message import Msg class MemoryRetrievalTool: """记忆检索工具：负责从长期记忆库中召回相关内容""" def __init__(self, store): self.store = store def lookup(self, user_id: str, query: str, top_k: int = 5): return self.store.search(user_id=user_id, query=query, top_k=top_k) class CustomerServiceAgent(AgentBase): def __init__(self, name, model, memory_tool, **kwargs): super().__init__(name, model_config=kwargs) self.memory_tool = memory_tool self.short_term = {} # 简化演示：本地短期窗口，生产环境应替换为Redis def reply(self, msg: Msg) -> Msg: user_id = msg.metadata.get("user_id") session_id = msg.metadata.get("session_id") # 1. 取短期记忆（当前会话的上下文） history = self._get_history(session_id) # 2. 按需检索长期记忆 memories = self.memory_tool.lookup(user_id, msg.content) # 3. 组装带记忆的Prompt prompt = self._build_prompt(msg.content, history, memories) # 4. 调用模型并处理工具调用 model_resp = self.model(prompt) # 5. 更新短期窗口，异步写长期记忆（略） self._update_history(session_id, msg.content, model_resp) return Msg(name=self.name, content=model_resp, role="assistant")

这段代码最重要的不是某一行的写法，而是顺序。记忆检索发生在Prompt组装之前，模型调用发生在记忆检索之后，这个顺序保证了模型看到的永远是“带着记忆的Prompt”。

3.2 记忆在消息循环中的执行顺序拆解

我把一次完整请求的执行顺序列出来，这个序列是整个项目最值得细嚼的部分：

1. 收到用户Msg，从metadata里取user_id和session_id。
2. 用session_id从Redis取最近20轮短期上下文。
3. 意图判断，决定是否需要触发长期记忆检索。
4. 如果触发，把检索到的记忆写入system prompt的“已知记忆”段落。
5. 组装完整Prompt，调用模型。
6. 如果模型返回的是工具调用请求，执行工具并把结果作为新的消息继续循环。
7. 最终返回用户可见的文本。
8. 异步执行记忆提炼，把值得长期保存的信息写入数据库。

这里第4步和第6步是两个最容易出问题的点。第4步如果记忆内容太长，反而会稀释模型对用户当前意图的注意力，所以要对检索结果做截断——每条记忆不超过50个token，最多取5条。第6步则是工具返回体污染上下文的重灾区，工具可能返回一张巨型表格，全塞进上下文既贵又乱，我的做法是先做摘要再回填。

3.3 贯穿全链路的会话标识：session_id是记忆的地基

再好的记忆系统，如果找不到“这段对话属于谁”，一切归零。我把session_id作为全链路的地基来设计：前端请求必须携带业务生成的session_id，后端把它塞进Msg的metadata，日志打点里也带它，数据库每条记忆记录也反查source_session_id。

这个设计在单机部署时看不出价值，一旦上了多实例、消息走队列，session_id就是串联所有环节的唯一主键。团队里有个同事一开始用时间戳拼用户ID生成session_id，结果用户从网页端打开一个、从API端又打开一个，两边各聊各的，长期记忆互相覆盖。最后统一改成：用户在某Agent下的会话ID，由Agent侧生成并返回给前端，后续所有轮次都必须回传这个ID。这个小改动，直接根治了会话分裂问题。

4. 生产级改造：并发上量时的服务化、限流与可观测性

4.1 从单体脚本到服务化：不要在请求线程里同步跑Agent

本地脚本跑得再顺，上了生产也只有被并发打爆这一条路。这里最容易踩的坑是用FastAPI起一个接口，然后在请求处理函数里直接调Agent。看似合理，实际上模型推理是秒级甚至几十秒级的阻塞操作，FastAPI的异步事件循环会被完全占死，新的请求全被堵在门口。

我的做法是引入worker隔离：请求进入FastAPI后，把任务投递到线程池或消息队列，由独立worker完成Agent推理。生产环境更严谨的方案是把Agent作为一个“服务方”独立部署，通过消息代理与上游通信，这也正是AgentScope分布式运行时的设计方向。先用线程池把服务化跑通，后续替换成真正的分布式消息通道，改动面非常小。

from fastapi import FastAPI from concurrent.futures import ThreadPoolExecutor import asyncio app = FastAPI() executor = ThreadPoolExecutor(max_workers=8) agent = CustomerServiceAgent(name="agent", model=...) @app.post("/agent/chat") async def chat(payload: dict): # 关键：把阻塞的Agent调用丢到线程池，不占事件循环 loop = asyncio.get_running_loop() result = await loop.run_in_executor(executor, agent.run_once, payload) return result

线程池大小不是拍脑袋定的。我根据单次请求平均耗时8秒、单机目标并发16路来估算，线程池至少需要16个worker才能扛住，并且要配合后续的限流策略才能保证高峰期不雪崩。上线前我用Locust跑了一轮，发现8个worker时P95延迟飙升到27秒，调到16个之后稳定在11秒左右，这个调参过程一定要自己做压测，别照搬别人的数字。

4.2 稳定性的三道闸门：限流、超时、重试

大模型服务的稳定性，本质上是跟“不确定性”博弈。模型API可能慢、可能超时、可能返回空、可能直接报错，所以我在Agent外层和工具调用层分别上了三道闸门。

限流这边，我用Redis做了双层令牌桶：单用户维度每分钟最多10次请求，全局限每分钟200次。用户多了可以把全局限流改成按Agent实例维度分别限制，避免一个客户的大量请求饿死其他用户。

超时设置我建议分成两层而非一个总超时。模型调用首token等待不超过30秒，单次Agent完整流程不超过120秒。连接超时、读取超时分别设置，不然一个连接卡住能拖死两个worker。

重试只覆盖“值得重试”的场景。模型API返回5xx或网络抖动导致超时，可以指数退避重试，最多3次；业务逻辑报错、参数不合法这类错误，重试多少次都会重复失败，直接抛出并记日志。一个容易忽略的点是：重试要保证幂等性。如果上一次调用模型已经生成了内容只是响应超时，重试会不会导致用户收到重复的答案？我通过在数据库中记录每轮请求的状态来避免：重试前检查该request_id对应的执行状态，避免重复写入最终结果。

4.3 可观测性：没有日志和Trace的Agent迟早崩在线上

Agent的运行链路比普通API长得多，模型调用、记忆检索、工具执行、Prompt组装，每一环都可能出问题。没有可观测性的Agent，出了事故只能靠猜，这是我这次项目中最深刻的教训之一。

我给每个关键环节加了结构化日志，核心字段固定下来：

引入“记忆命中率”这个指标是个转折点。上线第一周我发现检索触发率有70%，但命中率只有22%，也就是说大部分检索都是在瞎查。顺着日志一查，发现是embedding模型和写入内容不匹配，一些摘要写得太泛，检索时根本匹配不上。这个指标后来成了记忆系统最重要的健康度指标，比什么“回答准确率”都先看它。

5. 实测中的意外情况：并发冲高与记忆串台的完整排查链路

5.1 现象一：压测到50 QPS时大量504，问题出在记忆检索阻塞事件循环

这个坑我们是在上线前压测时暴露的。当时目标并发是50 QPS，Locust一拉起来，服务端先是少量请求超5秒，然后越来越多请求直接504，看面板上服务进程的CPU并不高，这就很反常——CPU不高、请求却在堆积，基本可以断定是某个操作在阻塞等待。

我的排查链路是这样的。第一步看接入层，Nginx日志显示部分请求的耗时超过30秒，网关先顶不住直接断开。第二步看服务层，Gunicorn的worker数和连接数都正常，但一次请求的内部耗时占比显示，Agent实际执行推理只占30%，剩下70%的时间全卡在memory.lookup这一步。第三步看记忆检索，发现我们用的向量检索客户端是同步阻塞实现，而它正好跑在FastAPI的异步处理函数里。

根因到此清楚：FastAPI的事件循环被同步检索卡死，任凭CPU空转也处理不了新请求。修复方案是把记忆检索改成异步调用，通过asyncio.to_thread或是专门的线程池承载，同时给检索操作单独设超时，避免底层检索服务抖动拖垮整条链路。修完后重新压测，P95从27秒降到了12秒，504基本消失。

5.2 现象二：部署两个实例后用户记忆“串台”与丢失

上线后我们横向扩到了两个实例，紧接着就收到了用户投诉：“我前面的对话内容怎么突然就没了，而且有时候回答像另一个人在说话。”这个问题比504更隐蔽，因为它不是崩溃，而是逻辑错误。

排查的第一步是复现。用户A连续发了三条消息，服务端日志显示这三条请求分别落在了不同的实例上。第二步查短期记忆的存储位置，发现我为了演示方便把短期记忆存在了Agent实例的进程字典里，第一轮落在实例1，第二轮落在实例2，两边各记各的，用户视角就是“对话断了”。第三步查长期记忆，发现部分事实记忆写入正常，但读取侧读到的上下文却是错乱的同一个session_id在两个实例上被并发读写——不串台才怪。

根因是会话状态没有外置。修复的核心思路其实只有一句话：短期记忆和对话进度这类状态，永远不要放在进程内。我把短期窗口整体搬到Redis，所有实例共用一份会话数据，并且用Redis的原子操作保证读写不互相覆盖。长期记忆库因为本来就在PostgreSQL里，问题不大，但读取时也要注意加一层缓存一致性控制。

验证方式很直接：杀掉其中一个实例，用户继续原来的session_id发起请求，对话能完整接上；再杀掉另一个，过几分钟后重建，短期记忆从Redis恢复，长期记忆从数据库恢复，用户全程无感知。做到这一步，才敢说这套系统有点“生产级”的底气了。

5.3 从事故中提炼的三条配置原则

踩完这两个坑，我把可复用的经验收敛成三条原则，写进了团队的配置基线：

• 同步阻塞型操作，一律不允许出现在事件循环的IO路径上。凡涉及数据库、向量检索、外部API，要么用异步客户端，要么丢进线程池。
• 会话态和短时状态，统一放外部存储。进程内状态只允许存放只读配置和临时计算结果，任何跨请求共享的可变状态都要外置。
• 每一层独立设超时，且必须有日志。模型层、检索层、工具层各自有自己的超时阈值，任何一个环节慢了，日志都告诉我“是谁慢了50毫秒、300毫秒、还是10秒”，这样排障才能从“盲猜”变成“按图索骥”。

6. 学习路径与后续扩展：从跑通 Demo 到吃透 AgentScope

6.1 官方资料的正确打开方式：别从头到尾读文档

AgentScope的文档其实写得不错，但我不建议零基础的同学从头到尾按目录顺序读，那样第三天就在“分布式运行时”章节里睡着了。我自己的攻略是：先照着官方教程里最早的两个example跑通一个简单的对话Agent，不过是改成中文场景，然后用一天时间专门搞懂消息机制。消息机制是AgentScope的灵魂，理解了Msg在Agent间怎么流动，后面什么Pipeline、多Agent协作、分布式部署都不会再有障碍。

读完消息机制再去看Agent机制，重点理解一个Agent内部一次推理的完整生命周期。最后才是分布式运行时，那时候带着问题去看效率更高——比如“我的Agent怎么拆成多个服务部署”“消息重放怎么做故障恢复”。

6.2 三个梯度练手项目：每档都有明确的验收标准

我在带新人时通常安排三挡练手任务，每一档的产出和验收标准都很清晰：

第一档：做一个带长期记忆的TODO管理Agent。模型读用户指令，把待办事项写进记忆库，下次打开还能列出。验收标准是关闭进程后重启，用户的历史待办仍然可查。

第二档：给Agent接入外部检索工具，做一个RAG式客服。让它根据文档库回答用户问题，并把回答依据写入长期记忆。验收标准是同一问题第二遍询问时，回答速度和准确率都有明显提升，日志里memory_hit字段为true。

第三档：把Agent服务化并用Docker部署两个实例，短期记忆上Redis，配合Locust压测。验收标准是50 QPS持续压测10分钟不出现504，杀掉一个实例会话不断。

三档做完基本就掌握了记忆型Agent的完整链路，后续再加什么工具、换什么模型，都是在这个骨架上长肉，不会伤筋动骨。

6.3 个人体会：别急着上向量库，先把会话记忆跑通

最后说点真正的体会。我看到很多开发者做记忆型Agent，一上来就上向量数据库、上RAG、上embedding，搞得很复杂。但我的建议是第一版老老实实做会话记忆加事实记忆，也就是Redis短期窗口加数据库结构化存储，先跑起来。我们第一版上线就是靠这两层撑起业务，跑了两周积累了真实的记忆命中率数据，才决定引入语义记忆。

因为语义记忆引入后要面对数据质量问题、embedding模型选择问题、检索相关性调优问题，这些在没有真实流量之前全是空转。先让记忆系统“有记忆可用”，再让它“记得更聪明”，这个节奏最稳。生产环境里一个记忆Agent最贵的成本不是数据库也不是推理，而是无效检索消耗掉的延迟和token费用，控制住这一项，整个系统的生产成本才真正可控。

回头看这个项目，AgentScope帮我把消息循环和状态管理这两个最容易被写乱的环节约束住了，剩下的记忆设计、并发改造和故障排查，本质上还是后端工程的功夫。想做好生产级Agent，先把记忆系统的存取逻辑和并发边界想清楚，工具和框架都是放大器，真正决定上限的还是你对细节的把控。