Hour 4: 压缩机制
一、概述
当对话历史太长时，Hermes 会自动压缩历史消息，用 LLM 生成的摘要替代原始消息。
目的： 避免超出模型的上下文窗口限制。
二、压缩触发条件
判断逻辑
python
# conversation_loop.py:4126
if agent.compression_enabled and _compressor.should_compress(_real_tokens):
messages, active_system_prompt = agent._compress_context(...)should_compress() 逻辑
python
# context_compressor.py:823
def should_compress(self, prompt_tokens: int = None) -> bool:
tokens = prompt_tokens if prompt_tokens is not None else self.last_prompt_tokens
# 条件 1: 超过阈值
if tokens < self.threshold_tokens:
return False
# 条件 2: 防抖 — 最近压缩效果差（<10% 节省）则跳过
if self._ineffective_compression_count >= 2:
return False
return True阈值计算：
python
threshold_tokens = context_length * threshold_percent # 默认 80%
threshold_tokens = max(threshold_tokens, MINIMUM_CONTEXT_LENGTH)三、压缩流程
整体流程图
上下文太长 (should_compress = True)
│
▼
compress_context() ← conversation_compression.py:281
│
├── 1. 检查压缩可行性
│ │
│ └── aux model 可用?
│
├── 2. 获取压缩锁
│ │
│ └── 防止并发压缩（多个 agent 同时压缩同一 session）
│
├── 3. 通知 memory manager
│ │
│ └── agent._memory_manager.on_pre_compress(messages)
│
├── 4. 调用 context_compressor.compress()
│ │
│ └── 使用 LLM 生成摘要
│
├── 5. 压缩成功?
│ ├── 是 → 用摘要替换历史，保存到新 session
│ └── 否 → 返回原始消息，不压缩
│
└── 6. 重建 system prompt详细步骤
1. 压缩可行性检查（懒加载）
python
# conversation_compression.py:321
if not getattr(agent, "_compression_feasibility_checked", False):
check_compression_model_feasibility(agent)
agent._compression_feasibility_checked = True首次压缩时才检查，节省 ~400ms 冷启动时间。
2. 压缩锁
防止多个 agent 同时压缩同一 session：
python
_lock_db = agent._session_db
_lock_acquired = _lock_db.try_acquire_compression_lock(_lock_sid, _lock_holder)
if not _lock_acquired:
return messages, _existing_sp # 放弃本次压缩3. 调用压缩器
python
compressed = agent.context_compressor.compress(
messages,
current_tokens=approx_tokens,
focus_topic=focus_topic,
force=force
)4. 处理结果
- 压缩成功 → 用摘要替换历史
- 压缩失败 → 返回原始消息，不丢失数据
四、ContextCompressor 算法
压缩算法详解
python
class ContextCompressor(ContextEngine):
"""
算法:
1. 修剪旧工具结果（无 LLM 调用）
2. 保护头部消息（system prompt + 首次对话）
3. 保护尾部消息（最近 ~20K tokens）
4. 用 LLM 总结中间的消息
5. 后续压缩时迭代更新摘要
"""步骤 1: 工具输出修剪
python
# context_compressor.py:849
def _prune_old_tool_results(messages, protect_tail_count, protect_tail_tokens):
"""
用摘要替换旧的工具结果内容
例如:
[terminal] ran `npm test` -> exit 0, 47 lines output
[read_file] read config.py from line 1 (3,400 chars)
"""步骤 2-3: 保护头尾
- 头部保护： system prompt + 第一轮对话
- 尾部保护： 最近 ~20K tokens（或最后 N 条消息）
步骤 4: LLM 摘要
用辅助模型（cheap/fast）生成结构化摘要：
python
SUMMARY_PREFIX = (
"[CONTEXT COMPACTION — REFERENCE ONLY] Earlier turns were compacted "
"into the summary below. This is a handoff from a previous context "
"window — treat it as background reference, NOT as active instructions. "
...
")步骤 5: 迭代更新
后续压缩时，LLM 会更新之前的摘要而非重新生成。
五、原地压缩 vs Session 旋转
两种模式
| 模式 | 配置 | 行为 |
|---|---|---|
| 原地压缩 | compression.in_place = True | 保持 session_id 不变 |
| Session 旋转 | compression.in_place = False | 创建新 session_id |
原地压缩（默认关闭）
python
in_place = bool(getattr(agent, "compression_in_place", False))
if in_place:
# 软存档旧消息，插入压缩后的消息
# session_id 保持不变
session.archive_and_compact(compressed, ...)
else:
# 结束旧 session，创建新 session
session.end_session(...)
new_session = session.create_session(...)六、压缩失败处理
失败场景
- Aux model 不可用 → 返回原始消息
- LLM 生成摘要失败 → 返回原始消息，显示警告
- 并发锁冲突 → 放弃本次压缩，下次重试
失败不丢失数据
python
# conversation_compression.py:469
if getattr(agent.context_compressor, "_last_compress_aborted", False):
return messages, _existing_sp # 返回原始消息七、手动触发压缩
用户可以手动触发压缩：
bash
/compress # 普通压缩
/compress <topic> # 聚焦压缩（保留与 topic 相关的内容）force=True 绕过摘要失败冷却时间。
八、关键文件位置
| 功能 | 文件:行号 |
|---|---|
| 压缩入口 | conversation_compression.py:281 |
| ContextCompressor | context_compressor.py:593 |
| 触发判断 | context_compressor.py:823 |
| 工具修剪 | context_compressor.py:849 |
| 压缩调用 | conversation_loop.py:4126 |
九、配置项
yaml
# config.yaml
auxiliary:
compression:
model: gpt-4o-mini # 用于摘要的辅助模型
compression:
enabled: true
threshold_percent: 0.8 # 80% 上下文时触发
in_place: false # 是否原地压缩总结
Hour 4 核心理解：
- 压缩在上下文超过阈值时自动触发（默认 80%）
- 使用独立的辅助模型生成摘要
- 保护头部和尾部消息，只压缩中间部分
- 压缩失败不丢失数据，返回原始消息
- 支持原地压缩（不创建新 session）