压缩阈值:为什么你的 AI 在 1M 窗口下从不压缩?
专栏信息
《从零到一构建跨平台 AI 助手:WeClaw 实战指南》专栏
本文是模块八第 3 篇,深入讲解动态压缩阈值的分档策略设计。
作者与项目
作者简介:翁勇刚 WENG YONGGANG 新概念龙虾-WeClaw 开发团队负责人,一群专注于跨平台 AI 应用的实践者 理念:"再复杂的技术,也能用代码讲清楚"
- 项目地址:https://github.com/wyg5208/weclaw.git
- 官网地址:https://weclaw.link
- 作者 CSDN:https://blog.csdn.net/yweng18
摘要
本文结构概览: 本文从一个看似无害的阈值公式出发,揭示它在 1M 窗口模型下的致命缺陷,然后介绍三档自适应策略的设计与实现,最后通过日志去重优化解决了 ReAct 循环中的 WARNING 刷屏问题。
背景:压缩阈值决定了"什么时候开始压缩上下文"。一个看似简单的参数,却在不同窗口大小的模型下表现出截然不同的行为。
核心问题:threshold = max(32K, context_window * 0.6) 这个公式在 128K 窗口下运行良好,为什么在 1M 窗口下就失效了?
解决方案:引入分档自适应策略,按窗口大小划分为三档,每档使用不同的压缩比例,并设置绝对上限封顶。
关键成果:
- 32K-2M 全窗口范围内的阈值均处于合理区间
- 消除了"大窗口永不压缩"的隐患
- 日志去重后,WARNING 从每步重复变为一次性提示
适合读者:LLM Agent 开发者,尤其在使用大窗口模型(128K+)时
阅读时长:约 10 分钟
关键词:压缩阈值、分档策略、上下文窗口、自适应、日志去重
一、一个看似无害的公式
1.1 旧方案的阈值计算
在 WeClaw 的旧版本中,压缩阈值的计算非常直观:
# 旧方案:简单的线性公式
def calc_trigger_point(context_window, ratio=0.6):
"""计算触发压缩的 token 阈值"""
min_threshold = 32000 # 最低 32K tokens
return max(min_threshold, int(context_window * ratio))
逻辑很简单:取上下文窗口的 60% 作为阈值,但不低于 32K。
1.2 在 128K 窗口下工作良好
模型:GPT-4o (128K 窗口)
阈值:max(32000, 128000 * 0.6) = 76800 tokens
预留:128000 - 76800 = 51200 tokens(用于输出和 SP)
状态:合理,压缩在对话较长时正常触发
1.3 在 1M 窗口下失效
模型:Gemini 1.5 Pro (1M 窗口)
阈值:max(32000, 1000000 * 0.6) = 600000 tokens
预留:1000000 - 600000 = 400000 tokens(远超实际需求)
状态:阈值过高,压缩几乎永远不触发!
问题:600K tokens 是什么概念?相当于 45 万字中文、或 300 页 PDF 的文本量。一个普通用户的对话很难达到这个量级,压缩机制形同虚设。
1.4 更严重的 2M 窗口
模型:Gemini 2.0 (2M 窗口)
阈值:max(32000, 2000000 * 0.6) = 1200000 tokens
实际:超过绝对安全上限,系统报错
状态:完全失效!
[图片: 阈值曲线对比图 | 生成方式: Python matplotlib 脚本,X 轴 context_window(32K-2M), Y 轴 threshold, 两条线(旧方案线性/新方案分档), 标注旧方案在 1M+ 窗口下的"失效区间"]
# Python 绘图脚本
import matplotlib.pyplot as plt
import numpy as np
windows = np.linspace(32000, 2000000, 100)
# 旧方案
old_thresholds = np.maximum(32000, windows * 0.6)
# 新方案(分档)
def new_threshold(w):
if w <= 128000:
return min(int(w * 0.6), 76800)
elif w <= 500000:
return min(int(w * 0.5), 200000)
else:
return min(int(w * 0.3), 300000)
new_thresholds = np.array([new_threshold(w) for w in windows])
fig, ax = plt.subplots(figsize=(12, 6))
ax.plot(windows / 1000, old_thresholds / 1000, 'r--', label='Old (linear)')
ax.plot(windows / 1000, new_thresholds / 1000, 'g-', label='New (tiered)')
ax.axhline(y=300, color='gray', linestyle=':', label='Absolute Cap (300K)')
ax.fill_between(windows/1000, old_thresholds/1000, alpha=0.1, color='red',
where=old_thresholds > 300000, label='Failure Zone')
ax.set_xlabel('Context Window (K tokens)')
ax.set_ylabel('Threshold (K tokens)')
ax.set_title('Compression Threshold: Old vs New Strategy')
ax.legend()
plt.tight_layout()
plt.savefig('threshold_comparison.png', dpi=150)
二、根因分析:阈值基于"原始窗口"而非"有效窗口"
2.1 窗口的三层结构
┌──────────────────────────────────────┐
│ 模型原始窗口 (1M tokens) │
│ ┌────────────────────────────────┐ │
│ │ 有效窗口 (~800K tokens) │ │
│ │ ┌─────────────────────────┐ │ │
│ │ │ 安全可用 (~600K tokens) │ │ │
│ │ │ (扣除 SP + 输出预留) │ │ │
│ │ └─────────────────────────┘ │ │
│ │ SP: ~20K 输出预留: ~200K │ │
│ └────────────────────────────────┘ │
│ 安全余量: ~200K │
└──────────────────────────────────────┘
旧方案的错误在于:用原始窗口计算阈值,但实际需要的是安全可用窗口。
2.2 真实的有效空间
| 组件 | 典型占用 | 说明 |
|---|---|---|
| System Prompt | 10K-25K | 身份+工具+指引 |
| 输出预留 | 窗口 * 15-25% | 模型回复空间 |
| 安全余量 | 窗口 * 10-15% | 防止 token 估算误差 |
| 可用于历史消息 | 窗口 * 50-65% | 这才是有效空间 |
一个 1M 窗口的模型,实际可用于历史消息的空间约 500-650K tokens。旧方案的 600K 阈值恰好踩在这个边界上,压缩自然很难触发。
三、分档策略设计:三档 + 封顶
3.1 设计原则
- 小窗口高比例:32K-128K 窗口空间紧张,60% 阈值确保充分利用
- 中窗口中比例:128K-500K 窗口适中,50% 平衡压缩时机和上下文保留
- 大窗口低比例:500K+ 窗口充裕,30% 确保压缩及时触发
- 绝对上限封顶:无论窗口多大,阈值不超过 300K tokens
3.2 核心实现
# 分档自适应阈值计算
COMPRESSION_CAP = 300_000 # 绝对上限
def calc_dynamic_trigger(effective_window, is_casual=False):
"""计算动态压缩阈值
Args:
effective_window: 有效窗口大小(扣除 SP 和输出预留后)
is_casual: 是否为闲聊模式(闲聊可提高阈值)
Returns:
int: 压缩触发的 token 阈值
"""
if effective_window <= 128_000:
# 小窗口:高比例,充分利用空间
ratio = 0.60
tier_cap = 76_800 # 128K * 0.6
elif effective_window <= 500_000:
# 中窗口:中等比例
ratio = 0.50
tier_cap = 200_000
else:
# 大窗口:低比例,确保及时触发
ratio = 0.30
tier_cap = COMPRESSION_CAP
threshold = min(int(effective_window * ratio), tier_cap)
# 闲聊模式放宽 20%(允许更长的对话历史)
if is_casual:
threshold = min(int(threshold * 1.2), COMPRESSION_CAP)
# 绝对下限保护
threshold = max(threshold, 16_000)
return threshold
3.3 各窗口下的阈值表现
| 模型窗口 | 旧方案阈值 | 新方案阈值 | 改善效果 |
|---|---|---|---|
| 32K | 32K | 19.2K | 更及时触发 |
| 64K | 38.4K | 38.4K | 基本一致 |
| 128K | 76.8K | 76.8K | 基本一致 |
| 256K | 153.6K | 128K | 提前触发 |
| 512K | 307.2K | 200K | 显著提前 |
| 1M | 600K | 300K | 从"不触发"到"正常触发" |
| 2M | 1.2M | 300K | 从"失效"到"正常工作" |
[图片: "阈值失效" vs "分档生效"对比日志截图 | 生成方式: 文生图 PROMPT: "Split comparison of two server log screens, left side showing 'WARNING: threshold too high 600000 tokens, compression may never trigger' repeated many times in red text, right side showing clean single 'INFO: compression triggered at 300000 tokens' in green text, dark terminal theme"]
四、验证:四种窗口下的实际表现
4.1 单元测试设计
import pytest
@pytest.mark.parametrize("window,expected_range", [
(32_000, (16_000, 20_000)), # 小窗口
(128_000, (70_000, 80_000)), # 中窗口
(1_000_000, (280_000, 310_000)), # 大窗口
(2_000_000, (280_000, 310_000)), # 超大窗口
])
def test_threshold_ranges(window, expected_range):
"""验证各窗口下阈值处于合理范围"""
threshold = calc_dynamic_trigger(window)
assert expected_range[0] <= threshold <= expected_range[1], \
f"Window {window}: threshold {threshold} not in {expected_range}"
def test_absolute_cap():
"""验证绝对上限封顶"""
# 即使窗口 10M,阈值也不超过 300K
assert calc_dynamic_trigger(10_000_000) <= COMPRESSION_CAP
def test_casual_mode_boost():
"""验证闲聊模式的阈值放宽"""
normal = calc_dynamic_trigger(128_000, is_casual=False)
casual = calc_dynamic_trigger(128_000, is_casual=True)
assert casual > normal # 闲聊模式阈值更高
assert casual <= COMPRESSION_CAP # 但不超过上限
4.2 运行结果
$ pytest tests/test_threshold.py -v
test_threshold_ranges[32000] PASSED (19200 in [16000, 20000])
test_threshold_ranges[128000] PASSED (76800 in [70000, 80000])
test_threshold_ranges[1000000] PASSED (300000 in [280000, 310000])
test_threshold_ranges[2000000] PASSED (300000 in [280000, 310000])
test_absolute_cap PASSED
test_casual_mode_boost PASSED
五、踩坑:ReAct 循环中的 WARNING 刷屏
5.1 问题现象
在一次正常的 ReAct 推理循环中,日志输出是这样的:
[Agent] Step 1: calling tool search_web...
[Context] WARNING: threshold is high (300000 tokens), compression may trigger late
[Agent] Step 2: calling tool read_file...
[Context] WARNING: threshold is high (300000 tokens), compression may trigger late
[Agent] Step 3: calling tool analyze_data...
[Context] WARNING: threshold is high (300000 tokens), compression may trigger late
[Agent] Step 4: calling tool write_report...
[Context] WARNING: threshold is high (300000 tokens), compression may trigger late
... (重复 N 次)
同一个 WARNING 在每个 ReAct 步骤都重复打印!
5.2 根因
ReAct 循环的每一步都会调用 get_messages() 获取当前上下文,而每次调用都会重新计算阈值并打印 WARNING:
# 问题代码
def get_messages(self):
threshold = calc_dynamic_trigger(self.window_size)
if threshold > 150_000:
logger.warning(f"threshold is high ({threshold} tokens)...")
# ... 后续逻辑
5.3 修复:集合去重
# 修复方案:模块级集合记录已警告的 (threshold, window) 组合
_warned_combos: set[tuple[int, int]] = set()
def get_messages(self):
threshold = calc_dynamic_trigger(self.window_size)
combo = (threshold, self.window_size)
if threshold > 150_000 and combo not in _warned_combos:
_warned_combos.add(combo)
logger.warning(f"threshold is high ({threshold} tokens)...")
5.4 另一个重复:阈值更新日志
DialogManager 的 update_threshold() 方法也存在类似问题:
# 问题:每次更新都打印,即使阈值没变
def update_threshold(self, new_value):
self._threshold = new_value
logger.info(f"Threshold updated to {new_value}")
# 修复:仅在阈值变化时打印
def update_threshold(self, new_value):
if new_value != self._threshold:
old = self._threshold
self._threshold = new_value
logger.info(f"Threshold updated: {old} -> {new_value}")
self._clear_cache() # 仅变化时清缓存
六、总结与展望
6.1 核心要点回顾
- 旧公式在大窗口下失效:线性比例的阈值在高窗口下过于保守
- 分档策略解决问题:按窗口大小选择不同比例,加上绝对上限
- 日志去重同样重要:ReAct 循环中的重复 WARNING 影响日志可读性
6.2 一个实用建议
如果你的 Agent 使用 128K+ 窗口模型,务必检查压缩阈值的计算方式。 很多框架默认使用简单的线性公式,在大窗口下可能导致压缩机制完全不触发。
下期预告:《三级容灾:当辅助 LLM 挂了,你的摘要怎么办》
- 压缩失败的后果:从"全丢"到"零丢失"
- 三级容灾的完整设计:辅助模型 -> 主模型重试 -> 静态回退
- 静态回退的巧妙设计:零 LLM 成本也能保留关键信息
敬请期待!
附录:参考资料
- 上一篇:《架构选型:三个开源项目的上下文管理哲学》(本系列第 47 篇)
- 下一篇:《三级容灾:当辅助 LLM 挂了,你的摘要怎么办》(本系列第 49 篇)
版权声明:本文为 CSDN 博主「翁勇刚」的原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接及本声明。