数据截至 2026-08-18(上游 commit b77d61291399)

压缩主干 — 一段内容怎么被路由到对的压缩器

30 秒导读: Headroom 的主干代码本身一个字节都不压缩。它做的是分诊:判断这段内容是 JSON、代码、grep 输出还是日志,交给对应的压缩器,并且在任何一步出问题时把原始消息原样还回去。本章只讲这条主干骨架,不讲各压缩器内部算法。

1. 这一章讲什么(零基础也能懂)

一句话定义: 主干 = 从 compress(messages) 到拿回压缩结果之间的那条调度链。

它解决的问题: 一次 agent 请求里塞的东西五花八门 —— 一条 rg 的搜索输出、一坨 pytest 日志、一个 4 万 token 的 JSON 接口返回、一段 Python 源码。用同一个压缩器对付它们,结果一定是灾难:把源码交给语义压缩器,标识符会被删掉;把日志交给 JSON 压缩器,它根本解析不了。

所以要先分诊,再动手。

用起来什么样: 调用方视角只有一个函数,分诊完全在里面发生。

from headroom import compress

result = compress(messages, model="gpt-4o")
result.messages            # 压缩后的消息,格式不变
result.tokens_saved        # 省了多少 token
result.transforms_applied  # ['router:smart_crusher:0.31', 'router:protected:user_message', ...]

一句话直觉: 把它当医院分诊台。分诊台不治病,它只做三件事 —— 看你哪不舒服(内容检测)、挂哪个科(策略选择)、要是科室今天关门就转普通门诊(回退)。真正治病的是各科医生(压缩器,见第 2 章)。

本章明确不讲:各压缩器内部算法(02)、CCR 原文存取(03)、净成本闸门与缓存经济学(04)。

2. 顶层全景:三层入口

从外到里有三层,每层问的问题不一样。先看清这三层的分工,后面的七步才不会串味。

                调用方
                  │  compress(messages, model=...)
                  ▼
 ┌────────────────────────────────────────────────────┐
 │ ① 库入口 compress.py                                │
 │   建配置 · 取单例 pipeline · 膨胀回退 · 异常兜底      │
 └───────────────────────┬────────────────────────────┘
                         ▼
 ┌────────────────────────────────────────────────────┐
 │ ② 变换流水线 TransformPipeline.apply                │
 │   数字前后各数一次 token · 熔断 · 只挂两级变换        │
 └───────────┬──────────────────────────┬─────────────┘
             ▼                          ▼
   ┌──────────────────┐      ┌────────────────────────┐
   │ CacheAligner     │  →   │ ③ ContentRouter         │
   │ 稳定前缀保缓存    │      │   分诊台(本章主角)      │
   └──────────────────┘      └────────────────────────┘

部件职责表:

部件	干什么	在哪个文件
`compress`	面向用户的一函数入口;配置归一、异常与膨胀兜底	`headroom/compress.py:171`
`CompressConfig`	用户能拧的开关(压不压 user 消息、保护最近几条…)	`headroom/compress.py:78`
`TransformPipeline`	顺序跑变换、数 token、熔断、埋点	`headroom/transforms/pipeline.py:86`
`CacheAligner`	稳定前缀,保住 provider 的 KV 缓存(见 04)	`headroom/transforms/cache_aligner.py:243`
`ContentRouter`	分诊:消息级筛选 + 内容级路由	`headroom/transforms/content_router.py:1694`
`PipelineExtensionManager`	十一个生命周期阶段的第三方扩展点	`headroom/pipeline.py:109`

2.1 一个容易看错的地方:ContentRouter 有两个 `compress`

ContentRouter 里有两个层级完全不同的方法,名字都跟压缩有关,读源码时极易混淆:

方法	层级	回答的问题	位置
`ContentRouter.apply`	消息级	这条消息该不该动?	`content_router.py:4681`
`ContentRouter.compress`	内容级	这一块文本该怎么压?	`content_router.py:2145`

apply 先把整个消息列表过一遍筛子(冻结前缀、user 消息、太小的、最近的代码、被保护的工具输出……全部原样放行),只把活下来的内容块交给 compress。本章第 3-4 节讲的"七步",讲的是 compress 这一层;消息级筛子在 §4.0 单独说。

ContentRouter.should_apply 永远返回 True(content_router.py:6574)—— 路由器自己不做"要不要跑"的判断,那是每条消息的事。

3. 七步流水线:一块内容的完整旅程

怎么读这张图: 从上往下是执行顺序,左边一列的箭头是"提前返回" —— 命中即停,不再往下走。

        一块文本 content
             │
   ① 内容检测 ──────────► 检测挂了 ─► 退回纯 Python 正则检测器
             │
   ② 策略选择
             │
       ┌─────┴─────┐
   混合│           │纯净
       ▼           ▼
   ③ 切段并逐段路由   │
       └─────┬─────┘
             │
   ④ 无损优先 ──────► 折叠成功 ─► 直接返回(不进有损)★
             │
   ⑤ 有损压缩 ──────► 压缩器不可用/抛异常 ─► 原文透传
             │
   ⑥ token 校验 ────► 没变小 ─► 原文 + 记进 skip 集
             │
             ▼
        压缩结果

★ 第 ④ 步的"命中即停"是整条链的性格:能无损省下的,绝不冒有损的险。

七步速查表:

步	一句话	失败/未命中时怎么办	主要符号
① 内容检测	这是 JSON、代码、grep 还是日志?	降级到纯 Python 正则检测器	`_detect_content`
② 策略选择	该类型对应哪个压缩器?	落到 `fallback_strategy`(默认 KOMPRESS)	`_determine_strategy`
③ 混合切分	一段里既有散文又有 JSON,拆开各压各的	切不出段就整块透传	`_compress_mixed`
④ 无损优先	先做零精度损失的字节折叠	折不动就返回原文,继续往下	`_lossless_first`
⑤ 有损压缩	交给对应压缩器真压	逐级回退:专用器 → Kompress → Log → 透传	`_apply_strategy_to_content`
⑥ token 校验	真的变小了吗?	没变小 → 用原文,并标记"这块别再试了"	`apply` 的 `accept_ratio` 闸门
⑦ 失败回退	整条链炸了怎么办?	返回原始 messages,一个字节不改	`compress` 的 `except`

4. 逐步拆解

4.0 前置:消息级筛子(谁根本不进这条链)

在七步开始前,ContentRouter.apply 先按顺序问一串"要不要跳过"。顺序本身有意义 —— 越前面的越是硬性保护。

顺位	跳过条件	为什么	位置
1	`i < frozen_message_count`	已被 provider 前缀缓存钉住,改一个字节就是全价重算	`content_router.py:5064`
2	内容是 list(Anthropic content blocks)	走单独的 block 路径	`content_router.py:5082`
3	`headroom_retrieve` 的结果	那是刚取回的原文,再压会写出赎不回的新标记	`content_router.py:5142`
4	被排除工具(Read/Glob…)的近期输出	仍可做无损折叠,但不许有损	`content_router.py:5142`
5	文件读取类命令的输出	agent 要拿原始字节去打补丁	`content_router.py:5195`
6	`role == "user"`(默认)	用户说的话是待分析的主体	`content_router.py:5211`
7	`role in {system, developer}`(默认)	缓存最热的指令字节	`content_router.py:5219`
8	token 数 `< min_tokens`	太小,压了也不省	`content_router.py:5226`
9	最近 N 条里的代码	正在编辑的代码要原样	`content_router.py:5262`
10	已带 CCR 标记的内容	二次压缩会把标记本身当原文吞掉	`content_router.py:5279`

第 9 条有个易误读点:protect_recent 只保护最近的代码,不保护最近的一切。判断里带着 and is_code(content_router.py:5262),所以最近几条里的日志、JSON 照压不误。

第 10 条靠一个小函数守着:

# content_router.py:133  _is_already_compressed
return any(marker in text for marker in _ALREADY_COMPRESSED_MARKERS)

_ALREADY_COMPRESSED_MARKERS(content_router.py:126)列了三种形状:Retrieve more: hash=、Retrieve original: hash=、<<ccr:。注释里记着一个真实事故(#2694):<<ccr: 一度不在这张表里,于是二次压缩把标记本身当原文哈希入库,原始字节的唯一把手就此消失。

幸存下来的内容块,才进七步。

4.1 第一步:内容检测 —— 双实现 + 看门狗

要解决的小问题: 手上只有一坨字符串,得知道它是什么。

类型清单(content_detector.py:25 ContentType):

枚举值	是什么
`JSON_ARRAY`	结构化 JSON
`SOURCE_CODE`	Python/JS/TS/Go/Rust/Java/C#/PHP 源码
`SEARCH_RESULTS`	grep/ripgrep 的 `file:line:content`
`BUILD_OUTPUT`	编译/测试/lint 日志
`GIT_DIFF`	统一 diff
`HTML`	网页
`TABULAR`	CSV/TSV/markdown 表
`STRUCTURED_CONFIG`	YAML/TOML/INI
`PLAIN_TEXT`	兜底

检测结果是 DetectionResult(content_type, confidence, metadata)(content_detector.py:39)。

两套实现,一个入口。 _detect_content(content_router.py:904)是唯一入口,底下有两个后端:

后端	实现	何时用
`rust`	原生 `headroom._core.detect_content_type`,magika 模型链	非 Windows 默认
`python`	纯正则检测器 `_regex_detect_content_type`	Windows 默认;原生出任何问题时的降级目标

选择逻辑在 _resolve_detect_backend(content_router.py:808):环境变量 HEADROOM_DETECT_BACKEND 强制,否则 "python" if sys.platform == "win32" else "rust"。

看门狗为什么必须存在。 原生检测器在 Windows 上第一次调用可能卡在 ONNX Runtime 的 Once 初始化里,0% CPU 永久等待,而已经释放 GIL 的原生调用没法从 Python 侧取消。所以 _rust_detect_watchdogged(content_router.py:836)把原生调用丢进 daemon 线程,主线程 worker.join(timeout):

# content_router.py:862-864
worker.join(timeout)
if worker.is_alive():
    raise TimeoutError(f"native detect_content_type exceeded {timeout:.1f}s watchdog")

超时预算 5 秒,可用 HEADROOM_DETECT_TIMEOUT_SECS 调(content_router.py:820)。卡住的那个线程救不回来,但调用方被释放了 —— 在代理场景里这等于救回一个压缩线程池的 worker。

超时之后还有一道进程级熔断:_detect_native_unhealthy = True(content_router.py:986),此后所有调用直奔纯 Python 检测器,不再每次都白等 5 秒、每次都再漏一个僵死线程(content_router.py:936)。

三个检测后的纠偏。 原生检测器有几种已知误判,主干在结果上打补丁:

误判	纠偏动作	位置
工具输出的 `<output>…</output>` 外壳让整块被读成 XML/HTML	检测前先剥壳(只在整串就是一个信封时)	`_strip_detection_envelope`(`:887`)
grep 输出、构建日志里密集的标点被读成 HTML	结构化 log/search 检测器positively认领时,推翻 HTML 判决	`content_router.py:1007`
magika 把 YAML/TOML/INI 判成 SourceCode	结构化 config 检测器置信度 ≥0.7 时改判	`content_router.py:1016`
原生判 PLAIN_TEXT	再跑一次正则检测器,若它有更具体的答案就用它	`content_router.py:1021`

4.2 第二步:策略选择 —— 从类型到压缩器

要解决的小问题: 类型知道了,交给谁?

_strategy_from_detection(content_router.py:2364)就是一张查表:

ContentType	CompressionStrategy	谁来干活
`JSON_ARRAY`	`SMART_CRUSHER`	SmartCrusher
`SOURCE_CODE`	`CODE_AWARE`	CodeCompressor(AST 感知)
`SEARCH_RESULTS`	`SEARCH`	SearchCompressor
`BUILD_OUTPUT`	`LOG`	LogCompressor
`GIT_DIFF`	`DIFF`	DiffCompressor
`HTML`	`HTML`	HTMLExtractor
`TABULAR`	`TABULAR`	TabularCompressor
`STRUCTURED_CONFIG`	`CONFIG`	ConfigCompressor
`PLAIN_TEXT`	`TEXT`	Kompress(ML)
未命中	`config.fallback_strategy`	默认 `KOMPRESS`(`:1543`)

策略枚举本身在 content_router.py:1354,共 12 个值(上表 9 个 + KOMPRESS、MIXED、PASSTHROUGH)。

一个反直觉的开关。 prefer_code_aware_for_code 关掉时,代码不会退回 Kompress,而是退到 PASSTHROUGH(content_router.py:2404)。源码注释解释得很直白:这个开关的本意是"让代码原样通过别被搞坏",而 Kompress 在 4.5 万 token 的 Python 上会压到 912 token、事实召回率 11%,那种结果对 agent 毫无用处。关掉专用压缩器 ≠ 换个压缩器,而是不压。

混合内容的判定与推翻。 _determine_strategy(content_router.py:2316)先看是不是混合:

# content_router.py:2348  —— 正则说混合,但原生检测器有把握说是源码时,信原生的
if detection.content_type == ContentType.SOURCE_CODE and detection.confidence >= 0.8:
    return self._strategy_from_detection(detection)
return CompressionStrategy.MIXED

原因写在上面的注释里:is_mixed_content 用的是廉价正则,Python 文件里行首的 {/[ 触发 has_json_blocks,docstring 触发 has_prose —— 一坨纯 Python 就被判成 MIXED,然后白白付一次切段的延迟,却什么也压不掉。

一处性能细节: compress 已经在上一行算过 is_mixed_content 和 _detect_content,所以把结果作为参数传进来复用(content_router.py:2206-2208 → :2331-2334)。原生检测是路由器每条消息最贵的开销,重算一次就是纯浪费。

4.3 第三步:混合内容切分

要解决的小问题: 一条 assistant 消息里,散文、代码块、JSON 常常混在一起。整体判个类型没意义。

切段规则(mixed_content.py:85 split_into_sections),按行扫描,四类段落:

触发	切出的段	备注
行首是代码围栏标记	`SOURCE_CODE`,记 `language`、`is_code_fence=True`	扫到下一个围栏为止
行首是 `[` 或 `{` 且能括号配平	`JSON_ARRAY`	配平扫描见下
行匹配 `^\S+:\d+:`	`SEARCH_RESULTS`	连续吃到不匹配为止
其余	`PLAIN_TEXT`	吃到下一个触发点

每段是一个 ContentSection(mixed_content.py:12),带 content / content_type / language / 行号 / is_code_fence。

是不是混合,靠五个信号投票(mixed_content.py:35 mixed_content_indicators):有代码围栏、有 JSON 块、有被文字包着的合法 JSON、散文特征超过 5 处、有搜索结果行。命中 ≥2 个就算混合(mixed_content.py:30)。

切完之后逐段路由(content_router.py:2470 起的循环):每段独立走 §4.2 的查表 + §4.4-4.5 的压缩,最后用 "\n\n" 拼回去(content_router.py:2514)。代码段会把围栏标记补回去(content_router.py:2500)。

切段前先把受保护的标签块摘出来。 _compress_mixed 开头调 protect_tags(content_router.py:2437),把 <system-reminder>…</system-reminder> 这类块换成占位符再切。原因很实在:切段边界会把这种成对标签劈到两个 section 里,后面每段单独保护时看到的是没配对的半个标签,保护等于没做 —— 而 Claude Code 正是用 <system-reminder> 装 CLAUDE.md 的。

更狠的一条:任何含占位符的段落一律原样透传,连压缩器的门都不让进(content_router.py:2471)。因为 restore_tags 遇到被改写过的占位符会直接丢弃整个被保护块 —— 压缩器啃掉一个占位符,后果比它想修的乱码还严重。

一个藏在切段里的性能坑。 _extract_json_block(mixed_content.py:214)对每个 { 开头的候选行做括号配平扫描。一旦某次扫到结尾都没配平,后面每个候选都会重走同一条尾巴 —— 复杂度是平方级。注释里留了实测数字:1200 行 JS 风格对象日志耗时 4643ms,每翻倍涨 4 倍。

修法很克制:只有在"确实有一次扫到结尾没配平"之后才建 memo(mixed_content.py:120、:89)。因为对最常见的漂亮 JSON,第一次就配平,memo 无处可用,反而因为字典读写让这条路慢了约 2 倍。

4.4 第四步:无损优先 —— 全流程的地板

要解决的小问题: 有损压缩有精度代价,能白拿的省法应该先拿。

思路: 每种格式都有"删了不丢信息"的部分 —— 日志里的 ANSI 颜色码、重复行、diff 里的 index <hex>..<hex> 记账行、grep 里重复的文件路径前缀。这些先折掉,零代价。

_lossless_first(content_router.py:2567)干三件事:

# 示意,非源码 —— 演示 best-of 的选法
primary = {SEARCH: "search", LOG: "log", DIFF: "diff", CONFIG: "config"}.get(strategy)
order = ([primary] if primary else []) + ["search", "paths", "log", "diff", "text"]

best, label = content, None
for kind in order:
    cand = compact_lossless(content, kind)   # 自验证:折不动就原样返回
    if len(cand) < len(best):                # 谁折得最小要谁
        best, label = cand, f"lossless_{kind}"

重点看两处:先试策略暗示的那种折叠,再试其它所有种。为什么要试其它种?因为检测器会misroute —— 对 .py 文件跑 grep -n,结果常被判成 SOURCE_CODE,但它实实在在是 grep 输出,试一遍 search 折叠就能捡回这份收益(content_router.py:2604-2610)。

这么乱试不会出事,靠的是可逆契约。 compact_lossless(lossless_compaction.py:438)每种折叠都当场自验证:

kind	折什么	自验证方式	位置
`log`	剥 ANSI + 合并重复行	`expand_runs(cand) == 剥色后的原文`	`:450`
`search`	提取重复的文件名/目录名做标题	两种折法各自 `inverse(cand) == content`,取更小的	`:459`
`paths`	纯路径列表折父目录	`path_unheading(cand) == content`	`:473`
`diff`	删 `index a..b` 记账行	纯减法,无逆函数检查	`:480`
`text`	合并空行	`expand_runs(cand) == content`	`:486`
`config`	合并重复行 + 重复多行段落	逆序两次逆变换后比对	`:493`

验证不过就返回原文,而且它永不抛异常(整个函数体裹在 try/except: return content 里,lossless_compaction.py:500)。这就是主干敢乱试所有 kind 的底气。

唯一的例外要单独防。 diff 是六种里唯一没有逆函数检查的:它无脑删掉任何长得像 index <hex>..<hex> 的行。在恰好含这种行的非 diff 内容上,那行就被静默且不可恢复地删掉了。所以主干专门加了一道:非 diff 策略且内容不像 diff 时,把 diff 从候选里剔除(content_router.py:2626-2631)。

"优先"是字面意义上的提前返回。 在 _apply_strategy_to_content 里,只要无损折叠真的产生了收益,函数就地返回,压根不进有损分支(content_router.py:3189、:3205)。只有在 HEADROOM_LOSSLESS_THEN_LOSSY 打开时,才会在折叠结果上再叠一层 Kompress,而且要求它再多省 5% 以上才收(content_router.py:3203-3213,阈值 _DEFAULT_LOSSY_MIN_EXTRA_SAVINGS = 0.05,见 :1740)。

还有一个更硬的模式。 config.lossless 为真时(HEADROOM_LOSSLESS=1),这是一份"绝不产生不可恢复损失"的契约:折得动就返回折叠结果,折不动就原文透传,绝不退而求其次上一个无标记的有损结果(content_router.py:3154-3157)。

这一步的 memo。 _lossless_first 是纯函数,但一次请求里同一块内容会被调两次(准入探测一次、真跑一次),所以缓存了 (hash(content), len(content), strategy, provider 代次) → 结果(content_router.py:2599)。缓存故意不加锁:并发撞车最多多算一次折叠,而在每个块的热路径上加锁没有任何正确性收益(content_router.py:2657-2663)。上限 256 条,满了整个字典清空,不做 LRU(content_router.py:1756、:2655)。

4.5 第五步:有损压缩 —— 只讲调度,不讲算法

要解决的小问题: 无损折不动的内容(代码、JSON、散文),得真压。

_apply_strategy_to_content(content_router.py:3066)是主干最长的一个函数,但骨架就是一串带回退的尝试。按执行顺序:

① 嵌套 JSON 重路由    整块不是 JSON,但内部含配平 JSON 片段 → 每段递归走本函数
② STAGE 0 无损折叠     §4.4;lossless-only 模式在这里就结束
③ 相关性切分          LOG/SEARCH:高相关记录留原文,低价值尾巴给 Kompress
④ 折叠命中即返回       ★ 有损分支从这里之后才开始
⑤ 外部压缩器          仅当运营方显式选了第三方压缩器
⑥ 内建 if/elif 分发    按策略调对应压缩器
⑦ 零收益回退          专用器没省下东西 → Kompress → (仅 JSON) Log
⑧ 记 TOIN、返回

嵌套 JSON 重路由(content_router.py:3107)值得单说:线性切段器看不见嵌在文本中间的 JSON,所以这里用 route_embedded_json 把每个配平片段单独送回本函数,再把结果拼回原位置、周围字节保持精确。递归调用带 _allow_embedded=False,那是一次性防重入闸,不是深度上限。

内建分发(content_router.py:3233 起的 try)每个分支形状一样:检查 config.enable_<x> → 拿懒加载的压缩器 → 通过 registry 调用 → 记 token 数与决策原因。整个 try 块外面套着一个 except Exception(content_router.py:3480),任何压缩器抛异常都只是让 compressed 保持 None,最后落到函数底部的原文透传(content_router.py:3637)。

零收益回退链(content_router.py:3487 起)是主干里最有代表性的一段兜底:

SMART_CRUSHER / CODE_AWARE / TABULAR / CONFIG 没省下东西
        │
        ├─► 试 Kompress ────► 更小?收下
        │                     └─► 不更小
        │                          │
        └──────────────────────────┴─► 仅当策略是 SMART_CRUSHER
                                       且内容是合法 JSON ─► 试 LogCompressor

那个 "且内容是合法 JSON" 的条件(content_router.py:3534-3538)修的是一个真实事故(#1306):原生检测器按形状而非可解析性分类,一段被截断的 JSON 工具输出被判成 json_array → SmartCrusher 解析不了、原样返回 → Kompress 也透传 → LogCompressor 把这几千行当"日志"折成一个 CCR 标记。在没配 CCR 取回的部署里,那就是 99.9% 的数据消失。

哪些压缩器可以挂进这条链、原文怎么被存起来,是第 2 章和第 3 章的事。

4.6 第六步:token 校验 —— 三道闸门

要解决的小问题: 压缩器说它压完了,但"压完"不等于"变小了",更不等于"该收"。

主干在三个不同高度各设了一道闸:

闸门	在哪	判据	不过关怎么办
块级接受闸	`content_router.py:5482`	`accept_ratio < min_ratio`	用原文 + `mark_skip`
可逆性闸	`content_router.py:5487`	工具输出被无标记有损压缩器改写	用原文 + `mark_skip`
请求级膨胀闸	`compress.py:278`	`tokens_after > tokens_before`	整批退回原始 messages

第一道:比例。 min_ratio 默认 1.0(content_router.py:1601-1602),也就是只要真的变小就收,不设省量下限。注释解释了为什么没有下限:缓存代价该由净成本闸门专门判(见 04),精度该由可逆性闸判,这条比例线不该越俎代庖。

无损结果在这里要换一把尺子量。折叠靠合并重复前缀省字节,但按词数算比例几乎不动(标题行甚至会把比例推过 1.0),那样会白白丢掉一个免费且可恢复的收益。所以命中无损的结果改用真实 token 数衡量(content_router.py:5471-5479)。

第二道:可逆性。 工具输出是 ground truth。如果一个有损总结器(KOMPRESS / TEXT / CODE_AWARE,见 content_router.py:1734 的 LOSSY_UNMARKED_STRATEGIES)交出的结果里找不到 CCR 取回标记,那这段就是不可恢复的 —— agent 会拿着一段无法核对的摘要去行动。这种结果宁可不要(content_router.py:5487-5498)。

第三道:膨胀。 库路径以前没有这道闸,是后来补齐代理侧已有的守卫(compress.py:275-291):

if tokens_after > tokens_before:
    logger.warning("Optimization inflated tokens (%d -> %d); reverting to original messages", ...)
    return CompressResult(messages=messages, ..., transforms_applied=["inflation_guard:reverted"])

注意返回值里那个 transforms_applied=["inflation_guard:reverted"] —— 回退这件事本身被记进了结果,调用方能看见发生了什么,而不是收到一个悄悄没压的结果。

还有一道不在闸门表里但同样重要的守卫:空输出。 压缩绝不允许把非空输入变成空(content_router.py:2235-2246)。原因很具体:Anthropic 会因为空的 user 消息直接 400 掉整个请求。任何变换产出空白,就地回退成原文。

记住"这块别再试了"。 没过闸的内容会进 CompressionCache 的 skip 集(content_router.py:1213)。这是个两层缓存:

层	存什么	命中效果	方法
Tier 1 skip	压不动的内容哈希	瞬间跳过,内存只是一堆 int	`is_skipped`(`:1281`) / `mark_skip`(`:1299`)
Tier 2 result	压得动的内容 → 压缩结果	直接复用压缩文本	`get`(`:1257`) / `put`(`:1294`)

两层都是 30 分钟 TTL,没有条数上限 —— 内存增长被会话时长天然框住。缓存 key 里带了运行时的 target_ratio(content_router.py:5296),因为同一段内容在不同比例下是不同结果,不能互相顶用;工具消息还会再叠一层命名空间隔离(content_router.py:5300-5302),免得被门禁拦下的工具条目污染普通条目。

4.7 第七步:失败回退 —— 主干的性格

整条链的姿态可以用一句话概括:任何一层出事,都退回上一层能给出的最完整结果;最外层的最完整结果就是"原始 messages"。

        异常发生在哪                     退到哪
 ───────────────────────────────────────────────────────
 单个压缩器抛异常              ─► 该块原文透传
 无损折叠验证不过              ─► 返回该块原文
 原生检测器 panic / 超时        ─► 纯 Python 正则检测器
 单块压缩超过 20s 期限          ─► PASSTHROUGH,记 warning
 某个变换抛异常                ─► 记一次熔断计数,异常继续上抛
 连续 3 次流水线失败            ─► 熔断打开,60s 内全部透传
 compress() 里任何未捕获异常    ─► 返回原始 messages + 记一次失败指标
 压完反而更大                  ─► 返回原始 messages

最外层长这样(compress.py:349-362):

except Exception as e:
    get_otel_metrics().record_compression_failure(model=model, operation="compress", error_type=type(e).__name__)
    logger.warning("Compression failed, returning original messages: %s", e)
    return CompressResult(messages=messages, tokens_before=0, tokens_after=0, ...)

指标先记、日志是 warning 不是 error、消息原样还回去 —— 压缩失败是可观测事件,不是可用性事故。

5. 主干外面的三层壳

5.1 用户开关:CompressConfig

CompressConfig(compress.py:78)是唯一的用户旋钮面板。每个开关的真实语义:

字段	默认	真实含义
`compress_user_messages`	`False`	压不压 user 消息。文档压缩 / RAG 场景要开
`compress_system_messages`	`True`	关掉可原样保留 system prompt(语音 agent 的工具定义不能动)
`protect_recent`	`4`	最近 N 条里的代码不压;`0` = 全压
`protect_analysis_context`	`True`	检测到 "analyze"/"review" 意图时保护代码
`frozen_message_count`	`0`	前 N 条已被 provider 前缀缓存钉住,变换一律不改写
`target_ratio`	`None`	Kompress 的保留比例;`None` = 模型自己定(约保留 15%)
`min_tokens_to_compress`	`250`	低于此 token 数的消息不动
`kompress_model`	`None`	Kompress 模型 ID;设成 `'disabled'` 可完全跳过 ML 压缩
`savings_profile`	`None`	具名高省量档位,如 `'agent-90'`

两个易错点:

min_tokens_to_compress 的默认值有两个。 CompressConfig 里是 250,但 ContentRouter.apply 在拿不到这个 kwarg 时用的是 50(content_router.py:4738)。走 compress() 永远是 250;直接构造 pipeline 的调用方拿到的是 50。
frozen_message_count 谁来填。 代理侧的 handler 自动算好传进来;库模式下自己管对话循环的调用方,应当传上一次请求的消息条数(compress.py:118-125)。

配置对象在入口处先 replace(config) 复制一份(compress.py:219),注释点明了防的是什么场景:一个长期共享的 per-agent 配置,被每个临时覆盖单个选项的请求悄悄改写。

5.2 单例与熔断

单例。 _get_pipeline(compress.py:398)是标准的双检锁:先无锁读,再加锁复检,最后构造。流水线本身持有懒加载的压缩器和缓存,重建一次代价不小。

只剩两级。 _build_default_transforms(transforms/pipeline.py:133)按顺序挂:

顺位	变换	条件
0	`ToolResultInterceptorTransform`	rollout 开关 `tool_result_interceptors` 打开时(`:144`)
1	`CacheAligner`	`config.cache_aligner.enabled`(`:151`)
2	`ContentRouter`	无条件(`:167`)

注意曾经有第三级、现在没了。 类文档字符串写着:Phase B PR-B1 退役了 IntelligentContextManager / RollingWindow 那个"从历史里丢消息"的阶段,只留活区压缩(transforms/pipeline.py:95-98)。为什么"删历史消息"是错的,是第 4 章的主题。

熔断。 连续失败 N 次后,直接透传一段冷却时间,而不是每个请求都重跑一遍必然失败的变换(transforms/pipeline.py:123-131):

参数	环境变量	默认
阈值	`HEADROOM_PIPELINE_BREAKER_THRESHOLD`	3(`≤0` 关闭)
冷却	`HEADROOM_PIPELINE_BREAKER_COOLDOWN_S`	60 秒

熔断打开时返回一个特殊标记(transforms/pipeline.py:274-281):

return TransformResult(messages=messages, tokens_before=passthrough_tokens,
                       tokens_after=passthrough_tokens, transforms_applied=["pipeline:circuit_open"])

计数只在变换真的抛异常时加一(transforms/pipeline.py:369-371),全部跑完就清零(:440)。连环境变量解析都是 fail-open 的:值写错了记一条 warning、用默认值,绝不让代理起不来(transforms/pipeline.py:70-83)。

一个诊断开销的取舍。 waste-signal 检测会把原始消息再解析一遍,纯为遥测,不影响压缩结果。但在超大对话上这一遍要几十秒,足以撑爆 Anthropic 的压缩超时,让代理 fail-open 丢掉一个已经算好的压缩结果(#296)。所以定了个天花板:

# transforms/pipeline.py:39
MAX_WASTE_SIGNAL_DETECTION_TOKENS = 100_000

超过就跳过诊断、保住结果(transforms/pipeline.py:488-497)。另外省得少于 100 token 也不跑,那是噪声(:43)。

token 只数两次。 全流水线只在开头和结尾各做一次完整 token 统计(transforms/pipeline.py:284、:445),中间每步的数字直接用变换自己报的,避免每步 O(N) 重数(transforms/pipeline.py:377-380)。

5.3 生命周期与扩展点

headroom/pipeline.py 定义的是一套跨实现的规范骨架:十一个阶段(pipeline.py:16 PipelineStage),按规范顺序排在 CANONICAL_PIPELINE_STAGES(:32)。

setup → pre_start → post_start
        ↓
   input_received → input_cached → input_routed → input_compressed → input_remembered
        ↓
   pre_send → post_send → response_received

compress() 实际只发三个:INPUT_RECEIVED(compress.py:240)、INPUT_ROUTED(compress.py:295,且仅当有 router: 开头的标记时)、INPUT_COMPRESSED(compress.py:308)。判断有没有路由标记的是 summarize_routing_markers(pipeline.py:103),一行的事:挑出所有 router: 前缀的条目。

扩展契约:实现一个 on_pipeline_event(event) -> PipelineEvent | None 就行(pipeline.py:67)。扩展可以就地改 messages / tools / headers / metadata,也可以返回一个新的 event 顶替。

发现方式:Python entry point,组名 headroom.pipeline_extension(pipeline.py:13)。discover_pipeline_extensions(pipeline.py:74)全程 fail-open —— 枚举失败、加载失败、实例化失败,各自记日志跳过,坏掉的第三方包不会拖垮发现流程。

分发也是 fail-open 的:某个扩展在某阶段抛异常,记 warning 后继续下一个扩展(pipeline.py:167-174)。

一个细节:compress() 构造 manager 时传的是 discover=False(compress.py:228)—— 库路径不自动发现 entry point,只认显式传进来的 hooks。

5.4 外部压缩器怎么挂进来

compressor_registry.py 是给第三方压缩器留的接缝,三个关键设计:

一、边界上只走纯数据。 契约里每个字段都是 str / int / bool / list / dict,不让任何 Python 专属对象(tokenizer 实例、store 句柄、富配置类)跨过去 —— 这样同一份契约可以用别的语言(比如 Rust)实现(compressor_registry.py:12-25)。

类型	装什么	位置
`CompressorDescriptor`	静态能力元数据:`name` / `content_types` / `lossless` / `cost_tier` / `recoverable`	`:62`
`CompressInput`	`content` / `content_type` / `query` / `config` / `budget`	`:83`
`CompressOutput`	压缩结果 + 前后 token 数 + 标记 + `hash → 原文` 恢复表 + `compressed` 标志	`:103`

CompressOutput.compressed 这个布尔位专门区分"真压了但没变小"和"根本没压(透传)",默认 True,好让不设它的老压缩器行为不变(compressor_registry.py:117-124)。

二、发现与选中是两件事。 discover()(:202)只加载并注册,绝不调用 compress;select()(:255)才决定谁真的上场:

传入	结果
`None` / 空集合	谁都不启用(opt-in 默认)
`"*"`	全部启用
具体名字	只启用"既被请求又已注册"的;请求了但没注册的记 warning 后跳过

设计意图写得很明白:光装一个第三方包,不该悄悄改变行为(compressor_registry.py:29-36)。

三、内建压缩器也被登记进来,但不走这条路分发。 _build_compressor_registry(content_router.py:457)给九个内建压缩器各注册一个委托适配器,再跑一次外部发现。适配器最终转调的还是路由器自己的 _get_* getter,所以真压时输出与历史直调逐字节一致。外部注册名撞车内建名时按 replace=False 跳过,第三方永远盖不掉内建的清单条目。

外部压缩器真正上场的位置只有一个:内建 if/elif 之前的那道 _try_external_compressor(content_router.py:2817)。它的 fail-open 条件列得很干净 —— 没选中、类型不匹配、抛异常、输出畸形或为空、压完反而更大,一律返回 None,内建分发原样接手。

类型匹配靠一张 ContentType → MIME 的桥表(content_router.py:489 _CONTENT_TYPE_TO_MIME),匹配规则接受精确匹配、* / */* 全通配、text/* 类型通配(content_router.py:502)。

还有一句边界值得记住:外部压缩器只在有损/CCR 模式下才可能被调到 —— lossless-only 模式在 STAGE 0 就返回了,所以第三方压缩器永远无法往一个"无损契约"的会话里注入不可恢复的损失(content_router.py:2842-2845)。

6. 兜底姿态总表

把散落各处的 fail-open 汇总成一张表,这是本章最该带走的东西:

出事的地方	姿态	代价	依据
原生检测器超时	换纯 Python 检测器 + 进程级熔断	一个僵死线程	`content_router.py:982-991`
原生检测器 panic	换纯 Python 检测器	一次 warning	`content_router.py:972-998`
无损折叠验证失败	返回原文	无	`lossless_compaction.py:455` 等
压缩器抛异常	该块原文透传	一次 warning	`content_router.py:3480-3483`
单块压缩超 20s	PASSTHROUGH	一个僵死线程	`content_router.py:5420-5430`
压缩产出空白	回退原文	无	`content_router.py:2235-2246`
压完没变小	原文 + skip 集	无	`content_router.py:5535-5539`
有损结果无 CCR 标记(工具输出)	原文 + skip 集	放弃这份收益	`content_router.py:5487-5498`
连续 3 次流水线失败	熔断 60s 全透传	这段时间不省 token	`transforms/pipeline.py:274-281`
请求整体 token 变大	退回原始 messages	无	`compress.py:278-291`
顶层任何未捕获异常	返回原始 messages	一次失败指标	`compress.py:349-362`
第三方扩展抛异常	跳过它,继续下一个	一次 warning	`pipeline.py:167-174`
第三方压缩器出任何问题	落回内建分发	无	`content_router.py:2817-2865`
熔断环境变量写错	用默认值	一次 warning	`transforms/pipeline.py:70-83`

一句话:整条主干没有任何一处会因为"压缩这件事失败了"而让请求失败。

7. 巧妙之处

其一:把"能不能安全乱试"变成压缩器自己的责任。 compact_lossless 每种折叠都自验证,折不动就原样返回、永不抛异常。有了这条契约,主干才敢对同一块内容把六种折叠全试一遍取最小,顺手捡回检测器误判丢掉的收益(content_router.py:2604-2610)。

其二:唯一没有逆函数的那一种,被单独关起来。 diff 折叠是纯减法,于是主干显式把它从非 diff 内容的候选里剔除,并且连第三方无损 provider 都不给看 diff 内容 —— "内建的折叠我至少能推理,第三方的推理不了,那就干脆不给"(content_router.py:2649-2655)。

其三:占位符段落一律不进压缩器。 不是"保护得更小心",而是根本不让它进门。因为下游的 restore_tags 遇到被改写的占位符会丢弃整个块,压缩器啃掉一个字符的后果比原问题更严重(content_router.py:2459-2465)。

其四:memo 只在被证明需要时才建。 _extract_json_block 的缓存要等到"确实有一次扫到底没配平"才开始建 —— 因为对最常见的输入,memo 净拖慢约 2 倍(mixed_content.py:120)。

其五:遥测让路给结果。 waste-signal 检测是纯诊断,但在大请求上它能把已经算好的压缩结果拖到超时作废。于是给它设了 10 万 token 的天花板,超了就不诊断(transforms/pipeline.py:39、:488)。

其六:回退这件事本身是可观测的。 膨胀回退返回 transforms_applied=["inflation_guard:reverted"],熔断透传返回 ["pipeline:circuit_open"],零收益回退把整条尝试链记进 strategy_chain(如 [smart_crusher, kompress, log],content_router.py:3088-3093)。日志读者不用去解析决策原因字符串,就能看出是怎么走到最终那个压缩器的。

8. 边界与局限

主干不保证省下 token,只保证不变大。 min_ratio 默认 1.0 意味着"只要变小一点就收",真正的省量取决于各压缩器,以及内容本身可压不可压。
两个看门狗都杀不掉卡住的线程。 检测看门狗和单块压缩看门狗都只能释放调用方,卡住的 daemon 线程留着跟进程一起死。源码注释直接标了 # ponytail,把真正的修法记在原生层(content_router.py:848-850、:5391)。
检测按形状,不按可解析性。 截断的 JSON 会被判成 json_array,这是 #1306 那条 log 回退加 JSON 合法性守卫的根因。
prefer_code_aware_for_code=False 等于不压代码,不是换个压缩器,容易被误读成性能开关。
min_tokens_to_compress 有两个默认值(250 vs 50),取决于走不走 compress()。
压缩器注册表目前只是清单。 内建压缩器仍由路由器自己的 if/elif 分发;registry 只在外部压缩器被显式选中时才影响真实请求路径(content_router.py:1785-1794)。

9. 代码地图

主题	文件路径	符号名
库入口 / 配置 / 膨胀回退	`headroom/compress.py`	`compress`、`CompressConfig`、`CompressResult`、`_get_pipeline`
变换编排 / 熔断 / 诊断天花板	`headroom/transforms/pipeline.py`	`TransformPipeline.apply`、`_build_default_transforms`、`_breaker_is_open`、`MAX_WASTE_SIGNAL_DETECTION_TOKENS`
生命周期规范 / 扩展点	`headroom/pipeline.py`	`PipelineStage`、`CANONICAL_PIPELINE_STAGES`、`PipelineExtensionManager`、`discover_pipeline_extensions`、`summarize_routing_markers`
消息级筛选(谁不进压缩)	`headroom/transforms/content_router.py`	`ContentRouter.apply`、`_is_already_compressed`、`CompressionCache`
内容级路由骨架	`headroom/transforms/content_router.py`	`ContentRouter.compress`、`_determine_strategy`、`_strategy_from_detection`、`_compress_mixed`、`_compress_pure`
策略分发与回退链	`headroom/transforms/content_router.py`	`_apply_strategy_to_content`、`_try_external_compressor`、`_registry_compress`
无损优先	`headroom/transforms/content_router.py`	`_lossless_first`、`_looks_like_diff`
无损折叠与可逆契约	`headroom/transforms/lossless_compaction.py`	`compact_lossless`、`collapse_runs`、`expand_runs`、`search_heading`、`diff_strip_index`
内容类型检测	`headroom/transforms/content_detector.py`	`ContentType`、`DetectionResult`、`detect_content_type`
检测后端 / 看门狗 / 纠偏	`headroom/transforms/content_router.py`	`_detect_content`、`_resolve_detect_backend`、`_rust_detect_watchdogged`、`_strip_detection_envelope`
混合内容切分	`headroom/transforms/mixed_content.py`	`ContentSection`、`is_mixed_content`、`split_into_sections`、`_extract_json_block`
外部压缩器接缝	`headroom/transforms/compressor_registry.py`	`CompressorDescriptor`、`CompressInput`、`CompressOutput`、`CompressorRegistry`

下一步读哪章:

想知道每个压缩器内部怎么保证"不破坏什么" → 02-compressors.md
想知道被删掉的原文去哪了、怎么要回来 → 03-ccr.md
想知道为什么"删历史消息"是错的、净成本闸门在算什么 → 04-cache-safety.md
想零改代码接进现成 agent → 05-proxy-and-wrap.md
想省输入之外的 token → 06-output-memory-learn.md

1. 这一章讲什么(零基础也能懂)​

2. 顶层全景:三层入口​

2.1 一个容易看错的地方:ContentRouter 有两个 compress​

3. 七步流水线:一块内容的完整旅程​

4. 逐步拆解​

4.0 前置:消息级筛子(谁根本不进这条链)​

4.1 第一步:内容检测 —— 双实现 + 看门狗​

4.2 第二步:策略选择 —— 从类型到压缩器​

4.3 第三步:混合内容切分​

4.4 第四步:无损优先 —— 全流程的地板​

4.5 第五步:有损压缩 —— 只讲调度,不讲算法​

4.6 第六步:token 校验 —— 三道闸门​

4.7 第七步:失败回退 —— 主干的性格​

5. 主干外面的三层壳​

5.1 用户开关:CompressConfig​

5.2 单例与熔断​

5.3 生命周期与扩展点​

5.4 外部压缩器怎么挂进来​

6. 兜底姿态总表​

7. 巧妙之处​

8. 边界与局限​

9. 代码地图​