第7期：多轮会话压缩

前言

第6期我们实现了用户特征记忆，AI 能跨会话记住用户偏好了。项目运行一段时间后，发现了一个新问题：长会话的 token 膨胀。

用户在一个会话里聊几十轮后，传给 AI 的历史消息越来越长。每次请求的上下文不断增大，不仅消耗更多 token，还增加了响应延迟。更严重的是，一旦上下文超过模型限制，会话就"炸"了。

这个问题业内怎么做？LangChain 提供了 ConversationSummaryBufferMemory——把旧消息压缩成摘要，保留最近的消息原文。本期我们来实现类似的能力。

方案设计

核心思路很简单：全量消息存 MySQL，只把摘要和部分消息传给 AI。

┌──────────────────────────────────────────────┐
│                 MySQL message 表              │
│  ┌──┐ ┌──┐ ┌──┐ ┌──┐ ┌──┐ ┌──┐ ┌──┐ ┌──┐  │
│  │M1│ │M2│ │M3│ │M4│ │M5│ │M6│ │M7│ │M8│  │
│  └──┘ └──┘ └──┘ └──┘ └──┘ └──┘ └──┘ └──┘  │  ← 全量保存
│                    │                         │
│    超过阈值时，最旧的被压缩                     │
│                    ▼                         │
│  ┌──────────┐ ┌──┐ ┌──┐ ┌──┐ ┌──┐ ┌──┐ ┌──┐│
│  │  摘要 A   │ │M4│ │M5│ │M6│ │M7│ │M8│ │M9││
│  └──────────┘ └──┘ └──┘ └──┘ └──┘ └──┘ └──┘│
└──────────────────────┬───────────────────────┘│▼传给 AI = [摘要(system)] + [保留的原始消息]

关键设计点：

决策	选择
摘要放哪	system prompt
压缩时机	同步（chat 开头调 compress）
触发条件	token 估算 > max_tokens × 0.5
压缩目标	压到 max_tokens × 0.25（留足余量）
弹出粒度	一条一条，从最旧开始弹
摘要模型	复用主对话模型（glm-4-flash）

压缩时机选同步而非异步，有个心路历程：起初想的是异步压缩，用户先看到回复再默默压缩。但分析后发现一旦触发第一次压缩，剩余消息紧贴阈值，后续几乎每次用户提问都会触发压缩——异步和同步差别不大了。索性 chat 开头同步调 compress，代码简单，而且读取的数据一定是压缩后的最新状态，不存在"异步还没处理完"的尴尬。

数据库设计

conversation 表加三个字段：

ALTER TABLE conversationADD COLUMN `compress_summary` TEXT NULL COMMENT '会话摘要',ADD COLUMN `compress_last_index` BIGINT NULL COMMENT '上次压缩到的消息ID',ADD COLUMN `compress_version` INT DEFAULT 0 COMMENT '乐观锁版本号';

compress_summary：存储 AI 生成的会话摘要
compress_last_index：记录上次压缩到哪条消息，后续读取时只取此 ID 之后的消息
compress_version：乐观锁，防止并发请求下两次 compress() 互相覆盖（如用户快速连发两条消息，两个请求线程同时执行压缩）

后端实现

配置类

@Component
@ConfigurationProperties(prefix = "ai.compress")
@Data
public class AiCompressProperties {private Long maxTokens = 128000L;private double ratio = 0.5;        // 硬阈值比例private double targetRatio = 0.25; // 异步压缩目标比例private double summaryRatio = 0.1; // 摘要字数相对 target 的比例
}

对应的 application.yml：

ai:compress:max-tokens: 128000ratio: 0.5target-ratio: 0.25summary-ratio: 0.1

Token 估算工具

不需要精确计算，简单的字符 × 2 即可满足需求：

public class AiTokenUtils {public static int estimateToken(String text) {if (text == null) return 0;return text.length() * 2;}public static int estimateToken(List<Message> messages) {return messages.stream().mapToInt(m -> estimateToken(m.getContent())).sum();}
}

核心压缩方法

@Override
public void compress(String conversationId, Long userId) {Long maxTokens = aiCompressProperties.getMaxTokens();double ratio = aiCompressProperties.getRatio();double targetRatio = aiCompressProperties.getTargetRatio();double summaryRatio = aiCompressProperties.getSummaryRatio();int ratioToken = (int) (ratio * maxTokens);int targetRatioToken = (int) (targetRatio * maxTokens);List<Message> messageList = getMessageList(conversationId, userId);int totalToken = AiTokenUtils.estimateToken(messageList);// 1. 总 token 不超过阈值，不压缩if (totalToken <= ratioToken) {return;}// 2. 从最旧开始一条一条 pop，直到 token ≤ targetList<Message> remaining = new ArrayList<>(messageList);List<Message> pruned = new ArrayList<>();int currentToken = totalToken;while (currentToken > targetRatioToken && !remaining.isEmpty()) {Message pop = remaining.remove(0);pruned.add(pop);currentToken -= AiTokenUtils.estimateToken(pop.getContent());}Conversation conversation = getOne(new LambdaQueryWrapper<Conversation>().eq(Conversation::getUserId, userId).eq(Conversation::getConversationId, conversationId));// 3. 被 pop 的消息 + 旧摘要 → 新摘要String newSummary = generateSummary(conversation.getCompressSummary(), pruned,(int)(targetRatioToken * summaryRatio));Long compressLastIndex = pruned.get(pruned.size() - 1).getId();// 4. 乐观锁更新boolean updated = update(new LambdaUpdateWrapper<Conversation>().eq(Conversation::getId, conversation.getId()).eq(Conversation::getCompressVersion, conversation.getCompressVersion()).set(Conversation::getCompressSummary, newSummary).set(Conversation::getCompressLastIndex, compressLastIndex).setSql("compress_version = compress_version + 1"));if (!updated) {log.warn("压缩乐观锁冲突，conversationId={}，version={}",conversationId, conversation.getCompressVersion());}
}

几个细节：

为什么压到 target（0.25）而不是 ratio（0.5）：留足余量。如果只压到 0.5，用户再问一两句就超阈值了，又会触发压缩，陷入频繁压缩的循环。压到 0.25 后，有足够空间让用户自由对话
乐观锁干什么用：虽然 compress() 只在一处（chat 开头）同步调用，但极端情况下用户快速连发两条消息，两个请求同时处理同一个会话，两次 compress() 可能同时执行。乐观锁确保只有一个生效，后写的因 version 不匹配自动丢弃

AI 摘要 Prompt

private String generateSummary(String existingSummary,List<Message> prunedMessages, int maxSummaryText) {StringBuilder prompt = new StringBuilder();prompt.append("请对以下对话内容进行摘要，保留关键信息"+ "（用户意图、重要问答、已确认的事实）。");if (existingSummary != null && !existingSummary.isEmpty()) {prompt.append("\n\n已有的历史摘要：\n").append(existingSummary);}prompt.append("\n\n新增的对话内容：\n");for (Message msg : prunedMessages) {String role = "USER".equals(msg.getRole()) ? "用户" : "助理";prompt.append(role).append("：").append(msg.getContent()).append("\n");}prompt.append("\n请输出更新后的摘要（" + maxSummaryText + "字以内）：");return chatClient.prompt().user(prompt.toString()).call().content();
}

摘要字数的设计：targetRatioToken × summaryRatio。假设 target 是 32000 token（128000 × 0.25），32000 × 0.1 = 3200 token ≈ 1600 字。这一般够用，实际测试下来 AI 输出的摘要通常在 200-500 字之间。

Graph 改造：注入摘要

ChatNode 接收摘要

在 ChatNode.apply() 中增加一段，拼入 system prompt：

String conversationSummary = state.value("conversationSummary", "");
if (!conversationSummary.isEmpty()) {systemPrompt += "\n\n以下是之前的对话摘要"+ "（其中包含了历史对话的关键信息）：\n"+ conversationSummary;
}

完整的 system prompt 结构变为：

你是一个公司内部智能助手...
回答规则：...以下是之前的对话摘要：              ← 本次新增
（历史关键信息的压缩摘要）以下是知识库内容...                  ← ragContext
以下是你对用户的了解...               ← userMemorySummary

GraphConfig 注册 state key

keyStrategyFactory = () -> Map.of("message", new ReplaceStrategy(),"messages", new ReplaceStrategy(),"conversationSummary", new ReplaceStrategy(),  // 新增...
);

Chat 流程整合

chat() 方法的完整流程：

public Flux<String> chat(ConversationReq req) {// 1. 存用户消息Message userMsg = saveUserMessage(req);// 2. 先压缩（方法内部判断，不需要则秒退）this.compress(req.getConversationId(), UserContext.getUserId());// 3. 获取压缩后的会话信息Conversation conversation = getConversation(req.getConversationId());// 4. 判断是否有摘要，决定传哪些消息String summaryToSend = conversation.getCompressSummary();List<Message> messagesToSend;if (summaryToSend != null) {// 有摘要：只取 compressLastIndex 之后的消息messagesToSend = filterAfterIndex(messageList,conversation.getCompressLastIndex());} else {messagesToSend = messageList;  // 全部}// 5. 构建 historyMessages 传给 graph state// 6. state 增加 conversationSummary// 7. 流式返回// 8. doFinally：存 AI 回复 + 提取用户特征
}

核心设计思路是先存消息，再压缩，再读取。这样 compress() 拿到的数据是最完整最新的，过滤后的消息也一定是正确的。

效果图

开始聊天：
7-开始聊天

多聊一些内容：
7-聊天多内容

触发压缩后的数据库记录，记住这个消息id 119：

7-数据库压缩结果

注入摘要后的实际效果，可以看到 system prompt 新增了之前对话的摘要内容：

7-压缩摘要注入效果

压缩后实际传给 AI 的消息数量明显减少，不再是全量历史：

7-消息截断效果1

对应数据库消息记录（按时间倒序）可以看到确实是消息id 119之后的
7-消息截断效果2

AI正常回复：
7-AI正常回复

踩坑记录

1. 乐观锁的必要性

压缩从异步改为同步后，表面上只有一处调用 compress()，看似不需要锁了。但实际测试时发现，快速连发两条消息会出现两个请求同时处理同一个会话的情况。compress() 不是原子操作——先查消息、再调 AI 摘要、最后写库——中间可能被另一个请求打断。

compress_version 乐观锁解决了这个问题：写库时带上当前版本号，如果期间被别的请求改过了，版本号不匹配，更新失败，结果丢弃。虽然概率低，但作为 defensive coding 保留下来。

2. 压缩后频繁触发

第一次实现时压缩目标是 ratio（0.5），每次只压到阈值就停。结果用户说两句话 token 又超了，又触发压缩，陷入频繁压缩的循环。

解决方案：压到比阈值更低的位置（targetRatio = 0.25）。好比水位线，涨到 80% 才放水，但一次放到 40%，而不是放到 79%。这样下次涨到 80% 还需要一段时间，大幅降低压缩频率。

3. 弹出粒度：单条 vs 一对

一开始按「一对」（user + assistant）弹出，但用户消息可能只有几个字，助理消息却很长。按对弹出容易弹过头。改为按单条弹出后，粒度更细，token 控制更精确。

架构总览

压缩模块在整体架构中的位置：

┌─────────────────────────────────────────────────┐
│            ConversationServiceImpl                │
│                                                   │
│  chat()                                           │
│   ├─ 存用户消息                                    │
│   ├─ compress()  ←─ 先压缩，确保数据最新             │
│   ├─ 加载 conversation（取摘要 + 最后压缩索引）      │
│   ├─ 过滤消息（有摘要则只取索引之后的消息）            │
│   ├─ 构建 historyMessages                          │
│   ├─ 调 graph（state 含 conversationSummary）       │
│   └─ doFinally                                     │
│        ├─ 存 AI 回复                                │
│        └─ 提取用户特征                               │
│                                                   │
│  compress()                                        │
│   ├─ estimateToken（字符×2）                       │
│   ├─ 判断是否 > ratio × maxTokens                   │
│   ├─ 从最旧 pop 到 ≤ targetRatio × maxTokens        │
│   ├─ pruned + existingSummary → AI → newSummary     │
│   └─ 乐观锁写入 DB                                  │
└──────────────────────┬──────────────────────────────┘│▼
┌─────────────────────────────────────────────────┐
│                message 表（全量保存）               │
│  所有原始消息一条不少，压缩只影响"传给 AI 什么"      │
└─────────────────────────────────────────────────┘│▼
┌─────────────────────────────────────────────────┐
│               传给 AI 的上下文                      │
│  [摘要(system)] + [compressLastIndex 后的消息]    │
└─────────────────────────────────────────────────┘