在 LangChain 生态里折腾深度 Agent(我一直习惯叫它 DeepAgents)的人,迟早会撞上同一堵墙:记忆。前 10 轮对话,模型像贴身秘书;到第 30 轮,它可能连用户在第 3 轮就确认过的筛选条件都忘得一干二净。去年我们给一个企业级数据分析 Agent 做长对话压测,40 轮里模型把用户早就确认过的三项约束反反复复追问了三次,不是因为上下文窗口不够,而是因为我们把所有历史不加区分地全部塞给了模型。
这篇速通指南的第二篇,主题就一个:用 Summarization 中间件给 Agent 的记忆做加减法。所谓“减法”,是把冗长的历史对话压缩成一条结构化摘要;所谓“加法”,是确保压缩时把关键信息完整保留下来;而中间件本身,则是夹在记忆读写与模型调用之间的一道自动阀门。适合谁看?已经跑通基础 Agent,但面对长对话明显力不从心的团队;也适合那些还没做记忆策略,想在成本、延迟、效果之间一次到位的新手。
1. 为什么 Agent 的记忆需要“加减法”——从上下文窗口超限说起
1.1 一个具体的超限场景
我们当时的模拟场景是客服问答加数据分析:用户陆续提出十几个问题,过程中 Agent 要调用工具查询库存、价格、历史订单。每完成一轮,工具调用产生的 tool_input、tool_output 以及中间推理消息都会追加进历史记录,平均一轮下来新增 400 到 800 token。到了第 20 轮,历史累计超过 12k token;到第 40 轮,接近 25k token。
虽然那个模型宣称上下文窗口有 32k,理论上“塞得下”,但行为已经明显劣化。把发送给模型的原始消息逐条拉出来看就会发现,大量内容是重复的:上一轮打印过的数据表、工具返回的状态码、用户反复措辞确认同样的事项。真正有信息量的内容可能只占三成。
很多人第一反应是“窗口不够大,换更大的窗口”。但换更大的窗口只是把悬崖向后挪了挪,模型对超长上下文的注意力衰减并没有消失。尤其是工具调用密集的 Agent,每轮都要把历史工具结果重新发给模型,token 费用也在同步线性上涨。这个教训让我意识到,记忆问题不能靠“扩容”解决,必须靠结构性的压缩。
1.2 加减法的语义:不是删除,而是压缩
市面上最常见的两种补救方案,一种简单粗暴地“截断”,只保留最近 10 轮,模型因此忘记用户三天前提过的约束;另一种是“全量重发”,每轮把整个历史塞给模型,成本与延迟一起失控。
Summarization 中间件走的是第三条路:老消息不是删除,而是先由轻量模型压缩成结构化摘要,再替换进消息流。这样主体事实不丢,体积却大幅下降。这也是为什么我要用“加减法”来比喻:减法是把冗余历史压薄,加法是确保压缩的过程中该记的都留下。
更准确地说,“加法”藏在摘要的写法里。压缩不是让摘要模型自由发挥写段剧情,而是显式要求它保留用户偏好、待办事项、已确认结论、关键数字这些值得长期记忆的信息。我们后来把摘要字段化,规定必须包含“用户要求”“待办”“关键数据”三类条目,这个变化比换更强的摘要模型带来的效果提升还要明显。
1.3 中间件的位置:一条消息流水线
用一段粗略的流程来描述:用户消息进来后,先进入记忆模块,由 Summarization 中间件判断“当前历史是否需要压缩”;如果需要,它调用摘要模型、更新记忆状态,然后把 [历史摘要 + 最近消息 + 新消息] 一并交给主 Agent。整个过程对上层业务透明,Agent 只觉得每次拿到的历史都是干净、紧凑、可读的。
为什么非要用“中间件”这种形态?因为记忆策略往往要在多个不同 Agent 之间复用。我们把压缩逻辑抽成标准组件后,任何新 Agent 只要接上同一套组件,就能自动获得一致的行为。这和 HTTP 中间件很像——它不关心业务本身,只是在请求进入核心处理器之前做一道通用的预处理。
2. Summarization 中间件怎么工作:三步编排
2.1 第一步:判断是否触发摘要
中间件的入口是一段很朴素的判断逻辑:每次有新的用户消息进来,先统计当前历史消息的 token 总量,再和预设阈值比较。触发条件通常是“token 超过阈值”或“消息条数超过阈值”的组合。
token 估算有两种办法:用开源 token 估算库精确计算,或者用字符数除以 2 做粗略估算。实测下来,英文场景粗略估算勉强能用;中文场景误差很大,一个汉字往往会被切成好几个 token。所以建议至少用估算库,否则触发时机可能差出一倍——要么过早压缩浪费成本,要么过晚压缩导致请求超限。
触发策略还应该叠加轮数条件。比如设定“历史超过 50 条 message 且 token 大于 5000 时才压缩”;如果只有 20 条消息但 token 已经很高,那多半是工具输出撑大的,同样应该触发。我们内部判断逻辑设计成两个独立条件,任一个满足都会进入压缩流程,灵活度更高。
2.2 第二步:增量摘要,而非全量总结
压缩的核心是增量摘要,不是全量总结。增量摘要的公式非常简单:
新摘要 = 摘要模型(旧摘要 + 本轮新增消息)
这样做有两个好处。第一,每次摘要模型的输入不会随会话无限膨胀,成本稳定;第二,旧摘要是经过前几轮迭代后的稳定结构,新摘要只是在它基础上做增量补充,格式和风格不会跳来跳去。
我第一次实现时犯的错,是让模型“从所有原始对话里提炼一份全新摘要”。到了第 30 轮,摘要调用本身就要输入 10k token,等于一次昂贵的大模型调用。改成增量之后,单次摘要成本下降了接近两倍,质量反而更稳定,因为旧摘要已经把主线固定住了。
2.3 第三步:替换消息流
摘要生成完成后,中间件需要做两件事:把已经被摘要覆盖的原始消息从 buffer 中移除;在消息列表最前面插入一条摘要消息,内容大致是“以下是本轮之前的会话摘要:\n……”。同时保留最近 N 条原始消息,目的是让主 Agent 还能看到近期对话的措辞和上下文。
这里有一个非常容易被忽略的细节:替换不是无脑丢掉,而是要在代码里记录一个 summarized_until 指针,表示“原始消息列表里从哪里开始是尚未摘要的内容”。否则下一次触发时,同一批旧消息会被重复压缩,摘要越来越臃肿,最终变成果酱面包一样的多层嵌套摘要。
3. 参数设计:阈值、窗口、模型选择
3.1 阈值怎么定
阈值是整个中间件最核心的超参数:定高了,等于没压缩;定低了,频繁调用摘要模型,白花钱。我的做法是倒推。
假设上下文窗口为 8k token,给模型生成输出预留 20%,再给系统 prompt 和工具定义预留 25% 的余量,那么历史消息最多可以占用 4.5k 左右。于是 threshold 就设成 4500,非常简单。
| 上下文窗口 | 输入预算(按留出 25% 余量) | 推荐 threshold |
|---|---|---|
| 4k | 2.2k | 2k |
| 8k | 4.5k | 4.5k |
| 32k | 20k | 20k 附近 |
不要硬顶窗口极限。我在长上下文测试里观察到明显的“中间遗忘”:内容排在前半段和历史中段的细节,模型把握得最差。与其赌它能从 30k 里找出一句话,不如主动压缩到 5k 以内,让它听得更清楚。
3.2 保留多少条最近原始消息
keep_recent 参数同样直接影响体验。设置太小,模型丢失近期对话的语气和措辞;设置太大,压缩的意义打了折扣。我们多轮实验下来,8 到 10 条是多数场景的甜点区;对工具调用频繁的 Agent 可以放宽到 12 条,因为工具结果往往比普通对话更需要近期上下文。
更精细的做法是按消息类型差异化保留:系统消息和工具结果优先被压缩,用户最近两次的原始提问尽量保留,因为用户的最新措辞里藏着真实意图。这个细节在客服类 Agent 上尤其重要——用户上一轮的“算了不改了”,和这一轮的“还是改回去吧”,必须同时可见。
3.3 摘要模型怎么选
摘要任务属于典型的高吞吐、低推理难度任务,完全没必要让主 Agent 的大模型亲自上阵。我们用的方案是:主 Agent 用强模型,摘要模型换成轻量级模型,效果完全够用,成本只有原来的十分之一。
如果你担心摘要质量下降,最有效的优化不是换更大的模型,而是把摘要模板写好。我们在模板里明确要求:“仅输出事实清单,不超过 20 行,每行 50 字以内,不要建议、不要情感判断、不要重复已有摘要”。模板结构化之后,摘要基本不会偏离主线。
4. 代码骨架:一个可落地的 Summarization 中间件
4.1 记忆状态的数据结构
先把记忆抽象成三块:summary 字符串、buffer 消息列表、summarized_until 指针。很多人只维护一个原始 messages 列表,后面做截断、替换时就会非常痛苦。
from dataclasses import dataclass, field from typing import List @dataclass class AgentMemory: summary: str = "" # 已压缩的历史摘要 buffer: List[dict] = field(default_factory=list) # 未压缩的最近消息 summarized_until: int = 0 # 已摘要到的消息条数buffer 里的消息可以用简化的字典结构,至少包含 role 和 content 两个字段。实际接入时再转换为对应版本的消息对象,不要在中间件里写死类型,方便后续切换模型或框架版本。
4.2 中间件核心逻辑
下面是去掉项目定制逻辑后的核心骨架:
class SummarizationMiddleware: def __init__(self, summarize_model, threshold_tokens=4500, keep_recent=8): self.summarize_model = summarize_model self.threshold_tokens = threshold_tokens self.keep_recent = keep_recent def estimate_tokens(self, messages: List[dict]) -> int: total = 0 for msg in messages: total += len(msg["content"]) // 2 # 精确场景改用 token 估算库 return total def should_trigger(self, memory: AgentMemory) -> bool: return self.estimate_tokens(memory.buffer) > self.threshold_tokens def summarize(self, memory: AgentMemory) -> str: prompt = self._build_summary_prompt(memory.summary, memory.buffer) return self.summarize_model(prompt) def _build_summary_prompt(self, old_summary: str, new_messages: List[dict]) -> str: conversation = "\n".join( f"{msg['role']}: {msg['content']}" for msg in new_messages ) return f"""你是对话记忆压缩器。 请基于已有摘要和新增对话,生成更新后的摘要。 规则: - 仅保留事实:未完成任务、用户偏好、已确认结论、关键数据 - 丢弃寒暄词、工具中间输出、重复表达 - 不超过20行,每行不超过50字 - 保持中性,不评价,不建议 已有摘要: {old_summary} 新增对话: {conversation} 更新后的摘要:"""这段代码只是骨架,但“触发、压缩、更新”三个动作都在里面了。实际项目中,摘要模型接口可能带温度参数,记得把 temperature 调低,比如 0.1 到 0.2,减少摘要生成时的随机性。
4.3 在每一轮 Agent 主循环中调用
def run_agent_loop(user_input: str, memory: AgentMemory, middleware): if middleware.should_trigger(memory): memory.summary = middleware.summarize(memory) # 保留最近 keep_recent 条原始消息 new_buffer_len = len(memory.buffer) memory.buffer = memory.buffer[-middleware.keep_recent:] memory.summarized_until += new_buffer_len - len(memory.buffer) request_messages = [ {"role": "system", "content": f"历史摘要:{memory.summary}"}, ] + memory.buffer + [{"role": "user", "content": user_input}] answer = agent.invoke(request_messages) memory.buffer.append({"role": "user", "content": user_input}) memory.buffer.append({"role": "assistant", "content": answer})注意第 5 行那两行的指针计算逻辑:先记录压缩前 buffer 的长度,再计算被截掉多少条,最后累加到 summarized_until。这样能避免下一次触发时,把刚写入的摘要又当成新消息处理。
第一次接入时最常见的报错来自消息类型。不同框架版本对 system/user/assistant 消息的类型派生不一样,建议在中间件内部统一转换成当前环境的 Message 对象后再传入 agent,不要在业务代码里分别处理。
5. 实测数据与调优记录
5.1 一组对照数据
我们在一个模拟客服 Agent 上跑过一组对照实验,上下文窗口 8k,threshold 取 4500,keep_recent 取 8,摘要模型用轻量级模型。结果如下:
| 轮次 | 无中间件单次输入 token | 有中间件单次输入 token | 摘要调用次数 |
|---|---|---|---|
| 10 | 约 4200 | 约 3200 | 0 |
| 20 | 约 9800 | 约 5100 | 1 |
| 30 | 约 15200 | 约 5400 | 2 |
| 40 | 约 23000 | 约 5600 | 2 |
第 40 轮以后,无中间件一侧的请求已经逼近甚至超过窗口上限,必须截断,效果必然劣化;有中间件的一侧,输入稳定在 5 到 6k,而且因为摘要里保留了既定结论,模型中途几乎不再重复问用户已确认过的内容。
5.2 踩过的四个坑
第一个坑:摘要被当成系统指令。模型看到 role 为 system 的历史摘要,会把它当作不可质疑的规则,结果照着摘要里的旧信息回答,反而忽略了最新 buffer。解决办法是把摘要消息的 role 从 system 改成普通 assistant,或者在摘要内容前加一句“以下为历史事实摘要,如有冲突请以最新用户消息为准”。
第二个坑:重复压缩同一批消息。指针没维护好时,每次触发都会把全部消息重新打包给摘要模型,摘要越叠越长,最后变成多层嵌套。解决方式就是前面说的 summarized_until 指针,同时每次压缩后立即截断 buffer。
第三个坑:摘要丢失用户偏好。如果模板里不明确列出“必须保留用户偏好”,摘要模型会倾向于写主线剧情,把“用户要求下周前必须交付”“推荐使用某种方案”这类细节丢掉。把保留字段写进模板后,问题基本消失。
第四个坑:摘要模型和主模型风格不一致。摘要里混入感叹词、情绪化描述,主模型会被带偏。解决方法是模板限定“中性事实描述,不做评价”,并持续监控摘要消息里的情感倾向。
5.3 我们最后调出的稳定配置
经过十几轮参数扫描,最终稳定配置是:threshold 设 5000,keep_recent 设 8,摘要模板要求 10 行以内、每行不超过 40 字,且必须包含“用户要求”和“待办”两类字段。这个配置在单 Agent 和多个子 Agent 并行场景下都很稳。
需要特别说明的是,参数不能盲目照搬。如果业务里工具输出特别大,比如经常返回整段 JSON,threshold 可以适当下调到 3500,否则触发时可能已经晚了。如果业务里用户提问非常简短、每轮 token 很少,threshold 可以上调,减少不必要的摘要调用。
6. 记忆加减法的更多玩法
6.1 对工具输出单独做摘要
对话不是唯一的记忆膨胀源,工具输出往往更猛。一个查询接口返回的 JSON 动辄几千 token,下一次工具调用时又原样带着。这时候可以对工具输出单独加一层摘要中间件,只保留核心字段、出错信息、影响决策的数值。这一层压缩不碰对话,风险更低,但收益往往比对话摘要还大。
6.2 分代理记忆:全局加会话
多 Agent 场景下,每个子 Agent 都有自己的对话记忆,但部分信息应该全局共享。我们的做法是单独维护一份共享摘要,由上层 Agent 统一读写,避免每个子 Agent 都保存一份重复的用户画像。这个结构本质上还是 Summarization 中间件,只不过作用对象从单个 Agent 的消息流变成了多个 Agent 间的共享状态。
6.3 摘要与检索记忆结合
摘要解决了“近期记忆的压缩”,但解决不了“很久以前某次对话的细节召回”。源码里那几句“用户曾经提过某型号存在兼容问题”如果被摘要压掉,后续就再也找不回来。
我们正在试验的方向是:摘要里保留关键实体和结论,同时把每次会话的原始消息做向量化存储。当 Agent 需要用到某条细节时,先通过摘要中的线索判断是否存在关联,再触发表格式检索。摘要负责确保主线不丢,向量库负责按需找回细节,两者配合才是完整的记忆体系。
最后给一个最省事的上手建议。如果你现在只想快速让长对话“不变傻”,别一上来就搞分层记忆和向量检索,先做 Summarization 中间件。阈值从 4500 起步,keep_recent 设 8,摘要模板按“事实清单、不评价、限行数”来写。跑通之后,再根据你实际业务里的 token 分布去调这三个参数。
我在好几个项目里都用同一套思路落地,每次效果都立竿见影。如果你也踩到过摘要污染或者重复压缩的坑,说明你已经在这条正确的路上继续往前走了。下一篇速通指南,我会把摘要中间件和向量检索记忆的组合细节展开聊聊,那部分才是真正能把“速通”变成“长跑不喘”的关键。