1. 为什么在生产环境里我选了 AgentScope1.1 记忆型 Agent 的真正难点在哪做 AI Agent 这几年我最大的感受是跑通一个 demo 非常简单把它推到生产环境还能稳定输出完全是另一门手艺。尤其是带记忆的 Agent——它可不是在 prompt 里加一句“请记住我们之前聊过什么”就完事了。大模型本身是天然无状态的一次请求结束这个“对话的大脑”就把刚才的内容全忘了。看起来最简单的解法是把历史对话一起传给模型但一进入真实场景就会撞上几堵墙第一上下文窗口是有限的。现在很多模型号称能支持 128K、256K 甚至 1M token但窗口大不等于可以塞满。塞太多历史模型对近期指令的关注力会被稀释响应延迟也在上涨成本更是一路飙升。第二用户的记忆不是线性的。他周一问过“我们这个产品的目标客户画像”周五可能又问“把那个画像里 25 到 35 岁的人群专门拉出来分析一下”。这种跨会话、跨主题的引用靠简单的“把聊天记录滚一遍”是根本取不出来的。第三真正难的是遗忘。哪些信息必须长期保存哪些只是一次性闲聊用户一个月前说“我比较看重性价比”今天说“预算已经批下来了选最贵的方案也行”你要把旧记忆覆盖掉吗这个“什么时候该更新、什么时候该忘”的决策才是记忆型 Agent 的灵魂。所以我始终认为记忆系统不是简单的数据库它是一个完整的读写链路想清楚写什么、存哪里、怎么取、何时丢才算把“记忆”这个词落地了。1.2 AgentScope 的核心抽象消息循环与 ReAct选型的时候我对比了 LangChain、AutoGen还有一些别的智能体框架最后定在 AgentScope 上。它不搞各种花哨的抽象而是把 Agent 之间的交互收敛成一套非常扎实的消息机制每个 agent 本质上是一个接收消息、处理消息、产出消息的节点消息本身可以串联成完整的调用链。这对我们做记忆型 Agent 特别重要。因为一个带记忆的助理业务不可能是单个模型在处理一切它背后至少有三个角色在协作一个负责意图判断和任务规划的主控 Agent、一个负责工具调用的执行模块、还有一个负责读写长期记忆的检索服务。这些角色之间的通信如果过于零散调试起来就是噩梦。AgentScope 用统一的消息对象把所有中间过程都串联起来Comp 类型和reply()的循环调用也很直观。另一个让我下决心的是它的 ReActAgent。ReAct 范式Reasoning Acting本质上是让模型在“思考-行动-观察”之间循环模型先分析当前情况决定要调用哪个工具框架执行工具并把结果喂回去模型再基于观察继续推理直到产出最终答案。AgentScope 已经把这条循环封装得非常干净我要做的就是注册工具、配好模型、设计好提示剩下的循环控制交给框架。坦白说框架的封装能力只是及格线真正让我踏实的是它对生产环境的理解。AgentScope 提供了消息流可视化、分布式部署支持、内存管理相关的构件这些东西短期内看不出价值等业务量上来以后会省掉大量折腾。1.3 2.0 的 RAG as Service 让我少写了几千行检索代码选择 AgentScope 还有一个更现实的原因2.0 版本提出的 RAG as Service 思路几乎正好打在我们记忆型 Agent 的痛点。传统的 RAG 链路每个人都很熟加载文档 - 清洗 - 分块 - 向量化 - 建索引 - 检索 - 重排序 - 把结果拼进 prompt。这一条链路听起来成熟但每一步都有大量细节分块粒度怎么定文档里有多级标题表格怎么办同一个知识库里既有用户私人记忆又有通用知识检索时怎么区分RAG as Service 的核心理念是把“文档解析、切分、嵌入、检索、重排”这一整套链路服务化对上层应用暴露统一接口。我用它来处理用户历史记录和私有知识库业务侧不需要关心底层是向量库还是全文检索引擎也不需要在代码里维护分块策略。而且它给的不只是检索能力还有检索结果的注入策略设计。这一点很重要因为 Agent 在不同阶段需要的记忆类型不一样。比如用户刚问完“帮我整理一下上月的数据”接下来追问“那个数据里华东区增长最快的是哪个产品”——这里第二句不需要重新检索所有资料只需要继承上一轮的中间结果就能回答。这种“何时检索、检索后如何注入、哪些记忆不该参与当前推理”的控制力比提供一个检索 API 要难得多也值钱得多。当然RAG as Service 不是银弹。它把一些固定链路固化下来换来的代价是灵活性下降。比如我们自研的某些特殊的元数据过滤逻辑最终还是要在服务之上再包一层。但整体来看前期省下的代码量非常可观。2. 记忆型 Agent 的整体架构拆解2.1 三层记忆模型工作记忆、长期记忆、策略记忆在动手写代码之前我花了整整两天设计记忆架构。参考了认知科学里工作记忆与长期记忆的划分我把系统的记忆分成了三层各自的存储介质和处理方式完全不同。工作记忆对应的是当前会话内、正在处理的任务状态。它存放在 ReActAgent 内部的消息列表里有明确的窗口上限。这一层只关心“这个任务正在进行的中间过程”用户这句话说完了、这个任务结束了里面的内容大部分就应该被清空。长期记忆才是真正意义上的“记住”。它存储在外部向量数据库和关系型数据库里承接三类内容用户明确陈述过的偏好“我比较看重响应速度”、历史发生过的事实“上周我们确认过预算上限是 5 万”、跨会话的知识片段“上一版方案里我们采用了 A 供应商”。这层记忆会持续积累由检索机制在恰当的时候取出来。策略记忆是我自己加的一层也是很多团队容易忽略的把过去成功的决策路径存下来作为新任务时的参考示例。比如用户的公司是做企业服务的他问“帮我写一份新产品的宣传文案”Agent 如果记得上次给这个用户写文案时用户明确说过“喜欢直白、数据驱动、拒绝浮夸修辞”它就能少走弯路。这层本质上是把经验变成 few-shot 示例。之所以要分三层而不是把所有内容都丢进同一个向量库核心原因是成本和正确性。工作记忆如果是用向量库存每一次取回都有几十毫秒的检索延迟没必要长期记忆如果没有结构化的时间、来源、用户 ID 等元数据就根本无法支撑精确筛选策略记忆如果混在普通历史里会被高频的日常对话淹没。2.2 一条用户消息在记忆系统里走了七步架构设计完之后我习惯用一条具体的请求把整条链路走一遍。就拿一个很典型的场景来说用户问“我上次让你整理的那份报表数据源到底在哪里”这条消息进来之后会经历七个环节。第一步判断是否需要检索长期记忆。系统先做一个初步判断如果这条消息只是闲聊或者完全新的话题就不需要触发检索。我用的是一个轻量规则加模型判断的组合如果消息里带有“上次、之前、我记得、那个”这类时间关联词或者涉及具体业务实体就强制触发检索。第二步查询改写。用户口语里的“那份报表”在向量空间里很难直接匹配到有价值的信息。需要基于当前上下文把问题改写成适合作检索引擎的完整查询比如“用户上周让我整理的关于华东区销售数据的 Excel 报表数据源来自哪”。第三步混合检索。单靠向量检索不够尤其是涉及表格标题、英文缩写、产品型号这些实体时关键词精确匹配往往更有效。我的做法是向量检索和 BM25 关键词检索并行跑再用 RRFReciprocal Rank Fusion把两份结果合并。第四步重排序与过滤。合并的结果根据三个维度重排相关性得分、时间衰减系数、来源可信度。用户一个月前提到“那份报表”和五分钟前提到“那份报表”时间权重完全不同。第五步注入上下文。经过筛选后的记忆片段会拼接到当前对话上下文中同时带上时间戳和来源标记方便模型理解这些记忆的时效性。第六步模型推理与行动。Agent 基于合成后的上下文做 ReAct 循环如果发现还需要更多信息可以再去检索或调用工具。第七步写回。在这次对话中如果出现了新的关键事实比如用户补充了“报表的数据源主要是 CRM 系统的导出接口”系统会提炼成结构化记忆写入长期记忆库中。这七步里前四步的质量直接决定了 Agent 的智慧程度。很多团队的检索召回率低问题往往出在第二步和第四步——没有做查询改写也没有引入时间衰减。2.3 最常见的错误把整段对话历史全塞给模型我现在可以很坦率地说第一版实现里我自己就踩过这个坑为了图省事直接把所有对话历史拼进 system prompt。后果是上线第一天就发现响应慢得离谱token 消耗是预算的三倍而且模型经常被一堆无关的历史信息带偏。举个印象很深的例子。用户问“那个项目什么时候能交付”系统里其实有两年间的各种项目记录结果模型把一年前另一个项目的交付时间当成当前问题的答案给出了。原因很简单所有历史都堆在那里模型没有能力做精确的权重筛选。后来我把方案改成了滑动窗口 摘要记忆 按需展开的组合。工作的消息列表只保留最近几轮更早的内容在每次会话结束时让模型生成一个 200 到 300 字的摘要存起来。如果用户问的是很细节的事摘要里没有系统才会去向量库里按语义检索原始片段。这种策略的收益非常明显日常推理时上下文里的噪声大幅降低token 消耗下降了 60% 以上而回答的准确率反而提升了。有时候少给信息、给精准信息大模型的表现反而更好这是我在实践中非常深刻的体会。3. 用 AgentScope 从零实现一个带记忆的 Agent3.1 初始化模型环境先搞定配置AgentScope 的接入门槛不高Python 3.9 以上就可以。我做的时候习惯先起一个干净的虚拟环境避免跟其他项目互相污染依赖。示例代码如下python -m venv .venv source .venv/bin/activate pip install -U agentscope # 选用你实际的模型服务客户端这里以 OpenAI 兼容接口为例 pip install openai # 如果使用 Milvus 做长期记忆存储 pip install pymilvus装完依赖之后最关键的步骤是配置模型。AgentScope 支持通过模型配置同时接入多个大模型这对我很有用复杂推理我用强模型记忆摘要用轻量模型两者在同一个 Agent 流程里各司其职。配置的方式是在初始化阶段传入 model_configs一个 OpenAI 兼容服务的配置大概长这样import agentscope agentscope.init( model_configs[ { model_name: main_reasoner, model_type: openai_chat, api_key: 你的密钥, base_url: 你的模型服务地址, }, { model_name: fast_summarizer, model_type: openai_chat, api_key: 你的密钥, base_url: 你的模型服务地址, }, ] )这里提示一下不同版本的 AgentScope 对 model_type 的支持略有差异接入前最好以官方的中文文档为准。我当时就碰到过版本不一致导致模型调用报错的问题排查半天才发现是升级后配置字段变了。3.2 用 ReActAgent 搭出工具型 Agent 骨架模型配置好之后就可以创建 Agent 了。AgentScope 的 ReActAgent 真正做到了“专注业务”而不是“天天搞框架”。我只需要把工具注册进去把短期记忆的载体指定好一个能自动思考、调用工具、观察结果、再思考的 Agent 循环就跑起来了。工具注册用 ServiceToolkit 非常顺手。以长期记忆检索为例from agentscope.agent import ReActAgent from agentscope.service import ServiceToolkit memory_store build_memory_store() toolkit ServiceToolkit() toolkit.register def recall_memory(user_id: str, query: str, top_k: int 5) - str: 根据用户 ID 检索长期记忆返回格式化为文本的记忆片段列表。 hits memory_store.search(user_id, query, top_k) return format_memory_hits(hits) def build_agent(): agent ReActAgent( nameassistant, modelmain_reasoner, toolstoolkit, memoryNone, # 短期记忆会由外部会话层管理后面会说明 ) return agentrecall_memory这个工具名很直白模型在 ReAct 循环中会自动意识到“如果遇到需要回忆历史信息的问题应该调用recall_memory工具来获取长期记忆。”这就是 Agent 的魔力所在工具列表本身就是模型决策空间的一部分。我还会注册若干个辅助工具比如用于获取实时数据的 HTTP 查询、用于记录新记忆的save_memory工具。注意如果一次性给 Agent 注册二三十个工具模型的选择会变得混乱还会增加不必要的 token 消耗。我的经验是核心工具控制在六到八个以内够用就行。3.3 把长期记忆接到向量检索服务长期记忆的存储我选的是 Milvus主要看中它支持标量过滤和向量检索的组合能力。在实际的存储设计中每条记忆不仅仅是文本还必须带上 user_id、timestamp、memory_type、source_message_id 这些元数据。检索的时候先按 user_id 过滤再做向量相似度搜索这样就避开了记忆串号的问题。Milvus 写入和检索的逻辑非常直接下面是一个简化示意from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType def build_memory_store(embed_fn): connections.connect(hostlocalhost, port19530) fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(nameuser_id, dtypeDataType.VARCHAR, max_length64), FieldSchema(namecontent, dtypeDataType.VARCHAR, max_length8192), FieldSchema(nametimestamp, dtypeDataType.INT64), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dim1024), ] schema CollectionSchema(fields) collection Collection(namelong_term_memory, schemaschema) def add_memory(user_id, content, timestamp): embedding embed_fn(content) collection.insert([[user_id], [content], [timestamp], [embedding]]) def search(user_id, query, top_k5): embedding embed_fn(query) res collection.search( data[embedding], anns_fieldembedding, param{metric_type: IP}, limittop_k, exprfuser_id {user_id}, ) return [hit.entity.get(content) for hit in res[0]] return add_memory, search这里最核心的一个要点是写入和检索必须使用同一个 embedding 模型。很多人换模型版本或者用不同的服务来处理写入和查询导致向量分布完全不在一个空间里召回结果自然惨不忍睹。所以我会把 embedding 服务单独打包全项目复用一个模型实例版本更新要做全量重嵌入。还有一点经验是向量维度的选择要与内容的粒度匹配。如果记忆大多是短句比如“用户偏好简洁的回答”嵌入模型选 1024 维的通用模型即可。如果记忆是长文档片段则要考虑分块策略而不是一鼓作气把整段塞进去。3.4 记忆摘要与裁剪策略落地记忆系统的另一根支柱是摘要策略。我在上一节提到滑动窗口只保留最近几轮对话那么更早的内容去哪了答案是被“摘要化”了。我的做法是每当当前会话的消息数超过 12 条系统就会启动一次摘要流程。用轻量模型把最开始的几轮对话压缩成一段 200 字左右的摘要随后在上下文中用摘要替换原始消息。这一步对保持推理质量至关重要。摘要有两个层次。第一个层次是对话轮次摘要记录“用户在什么时候、聊了什么话题、得到了什么结论”。第二个层次是原子事实抽取从对话中提取出可以长期复用的结构化信息比如用户的偏好、确定的决策、明确的时间节点。这类信息直接写入长期记忆库而不是留在对话上下文里。我在实际操作中把两个层次分得很开。比如用户说“以后都给我用中文回复”这属于原子事实必须马上单独写入长期记忆。用户说“今天我们先讨论一下方案框架吧”这属于对话进展最多放到轮次摘要里。如果混在一起检索的时候就会经常拿到“今天是讨论框架”这种过程性信息而不是“偏好中文回复”这种稳定事实。关于裁剪策略还有一个容易忽略的细节工具返回的冗长结果。很多 Agent 框架会把工具返回的原文直接拼进 ReAct 循环的观察消息比如搜索接口每次返回 20 条记录。我建议在工具函数内部就做压缩只保留模型实际可能用到的字段。让工具的返回值本身简洁比事后裁剪要省事得多。4. 生产级部署从 demo 到服务的最后一公里4.1 会话隔离绝不能让记忆串号如果你只在自己电脑上跑 demo用户 ID 可能想都不想就随手传个 test。但一旦上线这就是事故隐患。我在生产环境里遇到的第一个线上故障就是因为多个用户的请求复用了同一个 Agent 实例和同一个向量库集合结果用户 A 的消息被检索到用户 B 的对话上下文里用户 B 直接在群里问“为什么助手知道我上周问过什么”。这种事故对信任伤害极大。解决它的方案在技术上并不复杂但必须从第一天就刻进代码的基因里。我用的是三层隔离第一层会话层的短期隔离。每次用户请求都会创建独立的会话上下文对象session_id 由网关生成结束之后释放。Agent 实例本身保持无状态。第二层存储层的长期隔离。向量库的每次检索都强制加上user_id过滤这条规则在写入端和读取端同时生效。为了保险我在写入前会校验当前调用链里的 user_id 与要写入的记忆的 owner_id 一致。第三层接入层的路由隔离。在 Redis 里维护会话状态映射同一个用户的多次请求稳定路由到同一个后端 worker这样短期上下文不需要跨进程传递。实际上我最终的架构走向了完全无状态Worker 本身不保存任何东西短期上下文也存到 Redis 里请求进来时先加载上下文处理完再写回。这个过程会有额外的序列化和网络开销但换来的完整隔离性非常值得。4.2 可观测性确保每次记忆调用都能复盘记忆型 Agent 有一个风格跟普通 API 服务非常不同的地方它里面的内部状态特别多模型每一步在想什么、检索到了哪些内容、最后用了哪部分信息全都是黑盒。如果观测能力跟不上出了问题就只能靠猜。我做的第一个关键设施是全链路 trace_id。每一个用户请求从进入网关开始分配一个 ID贯穿整个 Agent 循环、每一次工具调用、每一次向量检索。日志里必须包含的信息有检索的原始 query、改写后的 query、召回的记录 ID 列表、每一条的相关性分数、注入 prompt 后各部分的 token 数、模型响应的耗时、工具执行结果摘要。这套日志体系对排查那种“模型回答看起来不对劲”的问题帮助极大。比如用户问“我上次那个方案通过了没”模型答非所问用 trace 一看就能定位到检索召回的前三条内容里根本没有跟“方案审批”相关的记录问题出在检索质量而不是模型推理。AgentScope 本身提供的 Studio 可视化工具在开发和压测阶段也很有用能直观看到消息是如何在不同节点间流转的。但在生产环境我们仍然以结构化日志和指标监控为主把每次 Agent 运行的决策轨迹固定下来这样复盘时才有据可查。4.3 成本与性能省 token 的几种实操生产级项目绕不开成本问题而我发现记忆型 Agent 的成本大头往往不是模型推理而是大量的历史信息堆积导致 token 超支。有几个操作层面的省法很有效第一是缓存聚合查询。用户经常会在同一天里多次问类似的问题比如“看一下我们昨天的转化数据”第一次做完整检索后结果缓存在 Redis 里过期的 key 设计在分钟级别同一用户的重复查询就可以直接命中缓存省掉一次模型推理和多次检索。第二是模型分级。不是所有的对话都需要最强模型来处理。寒暄、确认信息、简单的状态查询交给轻量模型就能完成只有复杂的推理、多步骤规划才让重量模型接手。在 AgentScope 里可以同时配置多个模型在代码里按消息类型动态选择使用不同模型这一点我是非常推荐的。第三是设置工具调用上限。ReAct 循环如果出现了异常模型有可能陷入“调用工具-结果不对-再调用工具”的死循环。务必给 agent 设置最大循环次数比如默认 8 次超过之后必须强制收敛输出当前能得到的最优结论。这样既保障服务稳定性也避免了模型意外乱试工具导致的高额费用。另外还有一个容易被忽视的成本点embedding 调用。检索链路里的查询改写和重排序如果每一步都做 embedding消耗积少成多。建议在服务内做一个简单的 embedding 结果缓存按文本的哈希值做 key哪怕只有几十的命中率长期跑下来省下的也相当可观。5. 常见问题排查与避坑实录5.1 检索老是召回不相关的内容这大概是记忆型 Agent 上线后第一个一定会被暴露的问题。用户感觉“系统好像有记忆但老是记错”多数时候不是模型的问题是检索召回的内容不对。我遇到过一个很典型的案例用户说“帮我查一下上次那个客户投诉的处理结果”系统召回了几条之前关于客户投诉模板编写的对话却没有召回真正的投诉事件记录。排查后发现两个原因第一原始事件记录里用户根本没有提“投诉”这个关键词而是用“那个难缠的客户”指代而向量检索时这条文本的语义权重偏向于“客户”没在乎“投诉”第二当时的记录把整个对话全文作为一条记忆写入太长被切分后丢失了关键信息。我的解决方法有两步。第一写入时做原子事实拆分每条记忆尽量只包含一个主谓宾结构控制在 100 字以内。第二检索时增加查询改写步骤用模型把当前问题转成几种不同的检索表达式再做结果合并。这个改动把召回准确率提升了很多。5.2 上下文越来越长token 爆了如果你只做了一层简单的“把最近的 N 轮对话存进上下文”业务一旦跑几周还是会遇到 token 失控。原因很简单有的工具返回结果特别大高频对话一轮就可能塞进几千 token。我采用的兜底机制叫动态重要性裁剪。当上下文总 token 数超过阈值时系统会按优先级丢弃内容。优先级从低到高依次是工具返回的原始数据、历史对话轮次、模型自己的中间推理、当前任务的指令。前两类可以放心裁剪因为工具数据可以重跑或重新检索历史对话已经摘要化中间推理如果太长会被替换为结构化结论。还有一个小技巧如果模型在一次 ReAct 循环中已经输出了最终答案那中间的所有推理环节在下一轮请求时都没有保留价值。我会在上下文组装前把这些中间步骤清理掉而不是等 token 快满了才处理。这相当于给 Agent 做“呼吸”让每一次推理都轻装上阵。5.3 并发一上来用户的记忆互相串这是我在没有做无状态化之前遇到的严重教训。早期我们把agent对象放在内存中多个用户请求共用一个实例。起初并发量低问题不显现峰值压力测试一上去用户的消息和记忆就开始互串。用户问 A 的事情Agent 却带着 B 的历史一起回答。后来我把短期状态全部外置到 Redisagent 对象每次处理前重建或清空内部的消息列表从外部加载当前用户的上下文。同时长期记忆库的检索永远带着 user_id 过滤。双重保障之下这类问题再也没有出现过。这里要特别提醒的一点是在代码里禁止把任何跨请求共享的全局变量当作消息传进 agent。哪怕只是临时代码也会变成生产隐患。所有状态要么写在用户的会话 key 下要么写在显式的函数参数里。5.4 如果重新做一遍我会从第一天就盯住三件事复盘整个项目如果要重来一遍我会在启动初期就把三件事做进地基里而不是后来补课。第一记忆的元数据规范。每条记忆从出生那一刻就必须带齐 user_id、时间戳、来源会话 ID、内容类型。一开始我觉得这些字段多写几行代码麻烦后来全部排障都依赖它们去过滤和回溯。没有这三个字段记忆系统就是一座无法检索的废墟。第二检索链路的指标体系。衡量检索质量不能只靠“用户是否满意”这种模糊反馈而是定义明确的指标召回率、前三位命中率、注入后模型是否实际使用了召回内容、上下文里的噪声比。每天看这些指标的变化能让你在用户投诉之前就发现质量下滑。第三记忆的死亡的策略。除了写入和读取系统还应该有定期的清理和合并机制。比如用户过去一年都没有再提某条旧知识它该被降低权重或归档比如两条记忆内容高度重叠应该合并。这个机制一开始不建数据积累到一定规模后检索效率会一路恶化到不可收拾。我个人的体会是做记忆型 Agent真正难的从来不是把记忆存进去而是知道在什么时机、用什么记忆、以及勇敢地删除那些过期的噪音。AgentScope 提供了一个消息循环清晰、工具编排顺手、检索链路可以高度定制的底座但记忆的取舍策略、线上的观测体系和数据质量的治理依然需要自己一点一点磨。如果你也在做类似的项目希望这篇内容能帮你少走几段弯路。