摘要

大语言模型的上下文窗口是有限资源。在长对话场景中,Token 数量不可避免地逼近模型的上下文长度上限,此时系统面临两难选择:截断历史导致信息丢失,或超出限制导致 API 报错。Hermes Agent 的上下文压缩引擎(ContextCompressor)实现了一套三阶段有损压缩算法,在保持对话连续性的同时将 Token 消耗控制在安全阈值内。本文从源码层面详细分析该机制的算法设计、边界处理、工具调用配对修复策略以及容错方案,为 AI Agent 开发者应对长对话上下文管理提供技术参考。

1. 问题背景

1.1 上下文窗口的工程约束

以 128K Token 上下文的模型为例,一次典型的编程辅助对话可能包含:

  • System prompt:2K-5K tokens
  • 文件读取结果:每次 3K-10K tokens
  • 终端命令输出:每次 1K-5K tokens
  • 代码补丁操作:每次 2K-8K tokens

经过 20-30 轮工具调用后,对话的累计 Token 数可轻松超过 80K,距离上下文上限已无剩余空间供模型生成回复。

1.2 常见方案的不足

方案 不足
硬截断(丢弃最早 N 条消息) 丢失任务目标和关键决策信息
固定窗口滑动 无法区分重要和不重要的历史消息
全量摘要 每次压缩都是全量重建,计算成本高
简单丢弃 tool result 可能破坏 tool_call/tool_result 配对,导致 API 报错

Hermes 的 ContextCompressor 针对上述问题,设计了一套兼顾信息保留率和结构正确性的分层压缩策略。

2. 整体架构

ContextCompressor 位于 agent/context_compressor.py,继承自 ContextEngine 基类,是 Hermes Agent 的默认上下文引擎。其核心算法由四个阶段组成:

消息列表(N 条,T tokens)
        │
        │  触发条件:T >= context_length × threshold_percent
        ▼
Phase 1:廉价预剪枝(无 LLM 调用)
        │  - Tool result 去重
        │  - Tool result 内容摘要化
        │  - 历史图片 base64 清除
        ▼
Phase 2:边界确定
        │  - HEAD:system prompt + 首轮交互(不压缩)
        │  - TAIL:最近 ~20K tokens 的消息(不压缩)
        │  - MIDDLE:介于两者之间的消息(待压缩)
        ▼
Phase 3:LLM 结构化总结
        │  - 将 MIDDLE 序列化为文本
        │  - 用结构化 prompt 生成 checkpoint 摘要
        │  - 支持迭代更新(增量合并历史摘要)
        ▼
Phase 4:消息重组 + 配对修复
        │  - HEAD + SUMMARY + TAIL
        │  - _sanitize_tool_pairs 修复孤儿配对
        │  - 角色交替验证
        ▼
压缩后消息列表(M 条,T' tokens,T' << T)

3. 触发条件

def should_compress(self, prompt_tokens: int = None) -> bool:
    return prompt_tokens >= self.threshold_tokens

其中:

threshold_tokens = max(
    int(context_length * threshold_percent),  # 默认 50%
    MINIMUM_CONTEXT_LENGTH,                   # 下限保护
)

以 128K 模型为例,阈值约为 64K tokens。在 50% 处触发而非等到 100%,确保压缩完成后仍有充裕空间供模型生成回复和执行后续工具调用。

4. Phase 1:廉价预剪枝

此阶段不调用 LLM,仅通过规则化操作降低 Token 消耗。

4.1 Tool Result 去重

当同一文件被多次读取时,仅保留最新一次的完整内容:

# 基于 content MD5 前 12 位去重
h = hashlib.md5(content.encode()).hexdigest()[:12]
if h in content_hashes:
    msg["content"] = "[Duplicate tool output — same content as a more recent call]"
else:
    content_hashes[h] = (index, tool_call_id)

4.2 Tool Result 摘要化

超出保护区的旧 tool result 被替换为信息密度更高的单行摘要:

# 原始(5000 字符):
{"role": "tool", "content": "module.exports = {\n  entry: './src/index.js'...(全文)"}

# 替换后(约 60 字符):
{"role": "tool", "content": "[read_file] read webpack.config.js from line 1 (5,000 chars)"}

关键设计:替换仅修改 content 字段,保留 tool_call_id,确保配对关系不受影响。

4.3 历史图片清除

计算机视觉工具(如 browser_snapshotvision_analyze)产生的 base64 图片数据可达 1MB+,对后续对话无持续价值。系统将非最新的图片 part 替换为文本占位符:

compressed = _strip_historical_media(compressed)
# [image: screenshot 1280x720, 847KB] → 约 30 tokens 的文本描述

5. Phase 2:边界确定

5.1 三区划分

# HEAD:system prompt + protect_first_n 条消息
compress_start = self._protect_head_size(messages)

# TAIL:从末尾反向累计至 tail_token_budget(约 20K tokens)
compress_end = self._find_tail_cut_by_tokens(messages, compress_start)

# MIDDLE = messages[compress_start : compress_end]

5.2 边界对齐

压缩边界不允许落在 tool_call/tool_result 组的中间位置:

def _align_boundary_forward(self, messages, idx):
    """起点对齐:跳过边界处的孤立 tool result"""
    while idx < len(messages) and messages[idx].get("role") == "tool":
        idx += 1
    return idx

def _align_boundary_backward(self, messages, idx):
    """终点对齐:不切断 assistant(tool_calls) + tool results 组"""
    # 如果 idx-1 是 tool,回溯到父 assistant 消息之前
    # 让整组要么完整进入 MIDDLE 被总结,要么完整留在 TAIL

这是防止 tool_call/tool_result 配对被破坏的第一道防线。

6. Phase 3:LLM 结构化总结

6.1 序列化

MIDDLE 区域的消息被序列化为标注角色的文本格式:

def _serialize_for_summary(self, turns):
    # 所有内容在序列化前经过敏感信息脱敏
    content = redact_sensitive_text(msg.get("content"))
    
    # 保留工具调用的名称和参数(截断后)
    # [ASSISTANT]: 分析代码结构
    # [Tool calls:
    #   terminal({"command": "find src -name '*.py' | head -20"})
    # ]
    # [TOOL RESULT call_abc]: src/main.py\nsrc/utils.py\n...

6.2 结构化总结 Prompt

总结模型收到的 prompt 包含一个固定的模板结构:

## Active Task       — 当前未完成的用户请求(逐字保留)
## Goal              — 用户的整体目标
## Completed Actions — 已完成操作的编号列表(含工具名、目标、结果)
## Active State      — 当前工作状态(目录、分支、文件、进程)
## In Progress       — 压缩发生时正在进行的工作
## Key Decisions     — 关键技术决策及其原因
## Relevant Files    — 涉及的文件列表
## Remaining Work    — 剩余待完成的工作
## Critical Context  — 不可丢失的具体数值、错误信息等

该模板的设计确保总结不是自由文本,而是结构化的 checkpoint 记录,模型在后续对话中可快速定位所需信息。

6.3 迭代更新

当对话再次需要压缩时(已经经历过一次压缩),系统不从头总结,而是在前一次摘要基础上增量更新:

if self._previous_summary:
    prompt = f"""
    PREVIOUS SUMMARY:
    {self._previous_summary}

    NEW TURNS TO INCORPORATE:
    {content_to_summarize}

    Update the summary: PRESERVE existing info, ADD new progress...
    """

这避免了每次压缩的信息损失叠加效应。

6.4 总结模型独立配置

# config.yaml
auxiliary:
  compression:
    model: "gpt-4o-mini"      # 用低成本模型做总结
    provider: "openrouter"
    timeout: 30

总结任务对模型能力要求低于主对话任务,使用更快更便宜的模型可显著降低压缩延迟和成本。当配置的总结模型不可用时,系统自动回退到主模型重试。

7. Phase 4:消息重组与配对修复

7.1 消息重组

compressed = HEAD + [summary_message] + TAIL

摘要消息的 role 需要避免与相邻消息产生同角色冲突:

# 优先选择不与 HEAD 末尾冲突的角色
if last_head_role in {"assistant", "tool"}:
    summary_role = "user"
else:
    summary_role = "assistant"

# 如果两个角色都会冲突 → 合并到 TAIL 第一条消息中
if summary_role == first_tail_role:
    _merge_summary_into_tail = True

7.2 _sanitize_tool_pairs:配对修复

即使边界对齐做了预防,极端情况下仍可能出现孤儿配对。系统通过事后修复兜底:

def _sanitize_tool_pairs(self, messages):
    # 收集所有存活的 tool_call_id
    surviving_call_ids = {id for msg in messages for tc in msg.get("tool_calls", [])}
    # 收集所有存活的 tool_result 的 call_id
    result_call_ids = {msg["tool_call_id"] for msg in messages if msg["role"] == "tool"}

    # 情况 1:tool_result 的 call_id 无对应 tool_call → 删除
    orphaned_results = result_call_ids - surviving_call_ids
    messages = [m for m in messages if m.get("tool_call_id") not in orphaned_results]

    # 情况 2:tool_call 的 id 无对应 tool_result → 插入 stub
    missing_results = surviving_call_ids - result_call_ids
    for tc in msg.get("tool_calls", []):
        if tc.id in missing_results:
            insert_after(msg, {
                "role": "tool",
                "tool_call_id": tc.id,
                "content": "[Result from earlier conversation — see context summary above]"
            })

这保证发送给 LLM API 的消息列表始终满足配对约束,不会因压缩操作导致请求失败。

8. 安全与防护

8.1 敏感信息脱敏

序列化和总结输出均经过 redact_sensitive_text 处理:

# 发给总结模型之前
content = redact_sensitive_text(msg.get("content"))
# 总结模型输出之后
summary = redact_sensitive_text(content.strip())

API key、token、密码等敏感值被替换为 [REDACTED],防止通过摘要泄露到辅助模型或持久化存储中。

8.2 反抖动机制

当压缩效果不佳时(节省率低于 10%),系统记录无效压缩次数,避免在每次 API 调用后反复触发低效压缩循环:

if savings_pct < 10:
    self._ineffective_compression_count += 1

8.3 失败处理策略

总结模型调用可能因网络、配额或模型可用性问题失败。系统提供两种可配置的降级方案:

配置 行为
abort_on_summary_failure=True 放弃压缩,保留原始消息,冻结对话
abort_on_summary_failure=False(默认) 插入确定性兜底摘要,从工具调用中提取关键信息

兜底摘要不依赖 LLM,通过程序化分析 tool_call 名称、参数中的文件路径、终端命令以及错误输出,构建最低限度的连续性锚点。

8.4 失败冷却期

总结失败后进入 30-60 秒冷却期,避免对不可用的总结模型发起密集重试:

self._summary_failure_cooldown_until = time.monotonic() + cooldown_seconds

用户可通过 /compress 命令手动触发绕过冷却期。

9. 配置参数

参数 默认值 说明
threshold_percent 0.50 触发阈值(占 context_length 的比例)
protect_first_n 3 HEAD 保护消息数(不含 system prompt)
summary_target_ratio 0.20 TAIL token 预算(占阈值的比例)
abort_on_summary_failure False 总结失败时是否中止压缩
auxiliary.compression.model 无(用主模型) 总结任务使用的模型

10. 工程启示

基于对 Hermes ContextCompressor 的分析,可提炼以下长对话上下文管理的设计原则:

  1. 分层压缩优于单一策略:廉价的规则化预剪枝(去重、摘要化)能在不调用 LLM 的情况下显著降低 Token 消耗,应优先执行
  2. 保护区设计是关键:HEAD(任务目标)和 TAIL(最近工作状态)不可压缩,中间区域的信息密度相对最低,是最佳压缩目标
  3. 结构化总结优于自由文本:固定模板确保关键信息(Active Task、Completed Actions、Relevant Files)不会被遗漏
  4. 迭代更新优于全量重建:利用前一次摘要作为种子进行增量更新,避免信息损失累积
  5. Tool call/result 配对是硬约束:边界对齐(预防)+ 配对修复(事后兜底)双重保障结构正确性
  6. 失败不能静默:压缩失败时必须向上层明确报告状态,由调用方决定是冻结对话还是降级处理

11. 结论

Hermes Agent 的上下文压缩机制展示了一种在工程实践中平衡信息保留率、结构正确性和运行成本的成熟方案。其三阶段分层设计、工具调用配对修复、迭代摘要更新等策略,为 AI Agent 开发者应对长对话上下文管理问题提供了可直接借鉴的技术路径。随着模型上下文窗口的持续扩大,该机制的触发阈值和保护策略可相应调整,但其核心设计思路——在有限资源下最大化关键信息保留——仍将长期适用。


本文基于 hermes-agent v0.15.1 源码分析,项目地址:https://github.com/NousResearch/hermes-agent

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐