侯瑞哲的博客

解读 Headroom：一个 20k star 的 AI Agent 压缩层

Headroom 是一个在 LLM 调用前压缩上下文的工具，声称能减少 60-95% 的 token，同时保持答案质量。2026 年初发布，半年内积累了 20k+ star。

这篇文章不是介绍怎么用它，而是从源码角度分析它的架构设计——有哪些工程思路值得学习。

一、它解决的问题

AI Agent 在处理任务时，输入给 LLM 的内容可能极度冗余。一个典型的场景：Agent 搜索了 100 个代码文件找某个 bug，把全部内容送进 LLM——其中 90% 都是不相关的代码。一次搜索就消耗了 17,000 个 token，绝大部分是噪音。

传统解法是"截断"（超过限制就丢掉），这会丢失重要信息。Headroom 的做法是"压缩"——识别哪些内容是关键的，剔除噪音，保留语义，同时把原文存起来，LLM 需要时可以取回。

实测数据：代码搜索 100 个结果从 17,765 token 压缩到 1,408 token，节省 92%，准确率保持 97% 以上。

二、整体架构：从输入到 LLM

用户请求（messages / tool outputs / logs / RAG chunks）
    │
    ▼
CacheAligner         — 稳定 prompt 前缀，命中 provider KV 缓存
    │
    ▼
ContentRouter        — ML 检测内容类型，分发到对应压缩器
    ├── SmartCrusher     — JSON / 结构化数据
    ├── CodeCompressor   — 代码（AST 级别）
    └── Kompress-base    — 文本 / 日志（自研 HuggingFace 模型）
    │
    ▼
CCR                  — 存储原文，注入检索工具
    │
    ▼
Memory               — 跨 Agent 共享状态
    │
    ▼
LLM Provider

不同模块职责清晰，接下来逐个分析最有价值的设计。

三、用 ML 做内容类型检测，而不是规则

ContentRouter 需要判断一段文本是 JSON、代码、日志还是普通文本，以便选择对应的压缩算法。朴素的实现是写一堆 if：

# 朴素做法（不是 Headroom 的做法）
if text.strip().startswith('{'):
    compress_as_json(text)
elif text.startswith('def ') or text.startswith('class '):
    compress_as_code(text)
else:
    compress_as_text(text)

Headroom 用的是 Google 的 Magika——一个专门做内容类型检测的深度学习模型：

"""ML-based content type detection using Google's Magika.

Magika is a deep learning model for content type detection that:
- Runs locally (~5ms latency)
- Supports 100+ content types
- Has 99%+ accuracy on supported types
- Requires no configuration
"""

class ContentType(Enum):
    JSON = "json"
    CODE = "code"
    LOG = "log"
    DIFF = "diff"
    MARKDOWN = "markdown"
    TEXT = "text"
    UNKNOWN = "unknown"

@dataclass
class DetectionResult:
    content_type: ContentType
    confidence: float      # 0.0 到 1.0
    raw_label: str         # Magika 原始标签
    language: str | None   # 代码时：python / javascript / ...
    metadata: dict

这里有一个值得注意的注释：# This is the ONLY place where we map labels - no hardcoding elsewhere。所有从 Magika 标签到内部 ContentType 的映射，集中在检测器这一个文件里，其他地方不允许硬编码。这个"单一真相来源"原则，让维护和修改变得容易。

工程启示：对于分类问题，当规则边界模糊（JSON 里嵌着代码怎么算？日志里有 JSON 怎么算？）时，ML 模型比规则更健壮。5ms 的延迟对于在 LLM 调用前做预处理来说几乎可以忽略，而准确率从 80% 提升到 99% 是实质性的改变。

四、管道生命周期模式与扩展点

Headroom 定义了一套完整的管道生命周期，每个阶段都可以被扩展：

class PipelineStage(str, Enum):
    SETUP = "setup"
    PRE_START = "pre_start"
    POST_START = "post_start"
    INPUT_RECEIVED = "input_received"
    INPUT_CACHED = "input_cached"      # CacheAligner 在这里
    INPUT_ROUTED = "input_routed"      # ContentRouter 在这里
    INPUT_COMPRESSED = "input_compressed"
    INPUT_REMEMBERED = "input_remembered"  # Memory 在这里
    PRE_SEND = "pre_send"
    POST_SEND = "post_send"
    RESPONSE_RECEIVED = "response_received"

扩展接口是一个 Python Protocol：

class PipelineExtension(Protocol):
    """Request lifecycle extension contract for the canonical pipeline."""

    def on_pipeline_event(self, event: PipelineEvent) -> PipelineEvent | None:
        """Handle a canonical pipeline event."""

扩展通过 Python entry points 动态发现：

ENTRY_POINT_GROUP = "headroom.pipeline_extension"

def discover_pipeline_extensions() -> list[PipelineExtension]:
    """Load registered pipeline extensions from Python entry points."""
    entries = importlib.metadata.entry_points(group=ENTRY_POINT_GROUP)
    ...

这意味着：任何人可以发布一个 Python 包，在 pyproject.toml 里注册 entry point，Headroom 就会自动发现并加载它，不需要修改 Headroom 本身的代码。

工程启示：这是一个非常干净的插件系统设计：

  • 用 Protocol 而不是抽象基类，更符合 Python 的鸭子类型精神，也不需要继承
  • 通过 entry points 实现"零侵入"插件发现，不需要主程序知道插件的存在
  • 管道阶段用 Enum 明确定义，防止拼写错误，也让 IDE 能给出补全
  • PipelineEvent 把所有可变状态封装在一个对象里，扩展可以修改或替换它

五、CacheAligner：让 Provider KV 缓存真正命中

LLM Provider（Anthropic、OpenAI）有 KV Cache 机制：如果连续两次请求的 prompt 前缀相同，第二次会命中缓存，延迟和费用大幅降低。

问题是：即使"概念上"前缀相同，实际发出的 prompt 可能因为以下原因不同：

  • 时间戳或请求 ID 注入在前缀里
  • 每次请求时动态生成的内容放在了系统提示的前半段
  • 工具列表的顺序不稳定

CacheAligner 的工作是"稳定前缀"——把可变内容后移，确保静态内容（系统提示、工具定义）始终出现在 prompt 的相同位置，最大化 KV 缓存命中率。

工程启示：KV Cache 对 LLM 调用成本的影响非常大（Anthropic 的 cache hit 有 90% 的折扣）。这个优化点很容易被忽视——开发者通常关注"发送了什么内容"，但很少关注"内容的顺序和位置"。Headroom 把这个优化做成了透明的基础设施，用户不需要主动思考它。

六、CCR：把可逆性设计进压缩

大多数压缩方案是单向的——压缩完原文就没了。Headroom 的 CCR（Content-Compressed Retrieval）不是这样设计的：

  1. 压缩时，原文存储在本地（本地优先，数据不离开你的机器）
  2. 压缩后的内容里，注入一个 headroom_retrieve 工具调用
  3. LLM 如果觉得需要原文（精细分析、验证细节），可以调用这个工具获取
  4. LLM 不需要时，就用压缩后的摘要，节省 token
压缩前：10,144 tokens（完整日志文件）
压缩后：1,260 tokens（摘要 + headroom_retrieve 工具声明）
LLM 找到 FATAL 错误后：调用 headroom_retrieve 获取那几行原始日志

这个设计体现了一个重要原则：不要让压缩丢失信息，而是把信息从"总是可见"变成"按需可见"。这和分页加载（懒加载）的思路本质相同——数据还在，只是延迟读取。

工程启示：CCR 模块的文件结构很有教学价值：

  • batch_processor.py：批量处理多条消息
  • batch_store.py：存储原文的本地数据库
  • context_tracker.py：追踪哪些内容被压缩了
  • tool_injection.py：向 LLM 注入 headroom_retrieve 工具
  • response_handler.py：处理 LLM 的检索请求，返回原文

这五个职责的划分非常清晰，每个文件只做一件事。

七、零代码改动的代理模式

Headroom 提供三种接入方式，设计得很有层次：

# 方式一：Library — 最灵活，改代码
from headroom import compress
messages = compress(messages, model="claude-opus-4-7")

# 方式二：Proxy — 零代码改动，拦截 HTTP 请求
headroom proxy --port 8787
# 然后改一行环境变量：
# ANTHROPIC_BASE_URL=http://localhost:8787

# 方式三：Agent wrap — 包装整个 agent 进程
headroom wrap claude

代理模式的工程价值在于：它把压缩逻辑从应用代码里解耦了出来。你的应用只是在和一个"假装是 Anthropic API"的本地服务通信，压缩是透明发生的。这和 Service Mesh 里的 sidecar 模式是同一个思路——把横切关注点（cross-cutting concerns）抽离到独立的基础设施层。

Headroom 的 Rust 侧有一个 headroom-proxy crate，代理的高性能部分用 Rust 实现。这也说明了 Python+Rust 混合架构的合理性：Python 用于业务逻辑和胶水代码，Rust 用于需要高吞吐的网络代理部分。

工程启示：设计对外接口时，考虑"用户能不改代码就用起来吗"。代理模式、ASGI 中间件、SDK 包装器——提供多个接入层次，让不同场景的用户都能以最小成本接入。

八、headroom learn：从失败中学习

这是整个项目里最有意思的功能之一。

AI Agent 在执行任务时会失败，失败的原因往往是重复的：每次都去读一个不存在的文件路径，每次都用错一个 API，每次都忘记某个项目的约定……

headroom learn 的工作：

  1. 分析历史 Agent 会话记录（Claude Code 的 session log、Codex 的历史等）
  2. 识别重复出现的失败模式
  3. 把这些模式写入 CLAUDE.md / AGENTS.md——这是 Claude Code 等工具会在每次启动时读取的"工程规范文件"

效果是：Agent 把自己的失败经验沉淀成了项目的"集体记忆"，下次不再犯同样的错。

工程启示：这个功能的本质是一个反馈循环——系统从自己的运行历史里提取改进点，自动写入规范文档。人工系统也应该有类似的机制：不只是记录错误，还要把错误的规律总结成可执行的规则，写进代码规范或自动化检查里。

九、Python + Rust 混合架构

仓库里有四个 Rust crate：

  • headroom-core：核心数据结构和算法
  • headroom-parity：Python/Rust 行为一致性测试
  • headroom-proxy：HTTP 代理服务器
  • headroom-py：Rust 代码的 Python 绑定（用 PyO3）

有一个 headroom-parity crate 专门用于验证 Python 和 Rust 实现的行为一致性——这说明有些算法同时在 Python 和 Rust 里实现了，Python 版本用于灵活性和可读性，Rust 版本用于性能。

工程启示：混合架构的代价是维护两套实现，但 parity 测试是一个聪明的工程解法——用自动化测试保证两套实现的等价性，而不是靠人工保证。这个模式在需要高性能但又需要灵活性的系统里很常见（SQLite 的 WASM 端口、NumPy 的 BLAS 后端都有类似设计）。

十、工程师能从中学什么

把这个项目的设计思路归纳成几条可以直接带走的原则：

1. 用 ML 替代规则的边界在哪里
规则简单直观，但边界模糊的分类问题（内容类型检测）规则很难写好。当 ML 模型足够轻量（5ms）且准确率有实质性提升时，ML 优于规则。

2. 单一真相来源（SSOT）防止分散的硬编码
所有从 Magika 标签到 ContentType 的映射，集中在一个文件的一个地方。这个约束用注释明确声明，防止开发者在其他地方重复映射逻辑。

3. Protocol 比抽象基类更轻
Python Protocol 实现了结构型子类型（structural subtyping）——只要有 on_pipeline_event 方法，就符合 PipelineExtension 接口，不需要继承。这让第三方实现更自然。

4. Entry points 实现零侵入插件系统
通过 Python entry points 发现插件，主程序不需要知道插件存在，插件也不需要修改主程序。这是 Python 生态里实现插件系统的最干净方式。

5. 可逆性设计：信息从"总是可见"变成"按需可见"
CCR 不丢弃原文，而是把"原文放在哪里取"的索引注入 prompt。这个思路可以推广：任何时候你需要压缩或摘要，考虑是否能保留原文并提供按需检索的接口，而不是直接截断。

6. KV 缓存优化是隐形的成本节省
Provider 的 KV 缓存 hit 可以节省 90% 费用，但很少有开发者主动优化 prompt 前缀稳定性。把这个优化做进基础设施层，让上层应用无感知地受益。

7. 多层接入方式满足不同场景
Library（最灵活）、Proxy（零代码）、Wrap（完整包装）——同一个功能，三种接入成本，覆盖了从"我要深度集成"到"我一行代码都不想改"的全部用户。

8. 从运行历史中提取改进点
headroom learn 的思路：系统能自动分析自己的失败记录，把规律性的失败转化成可执行的规则，写入规范文档。这个反馈循环可以推广到任何有历史日志的系统。