在前面几篇里，我们把 Headroom 的输入侧基本都聊过了：compress() 入口和管线生命周期、ContentRouter 怎么分流内容、SmartCrusher / CodeAwareCompressor / Kompress 三个压缩器各自压什么、CCR 可逆压缩怎么把原文缓存在本地、跨 agent 记忆和 headroom learn 又怎么把经验沉淀成长期知识。它们有一个共同点：围着的都是发给模型这一路。工具输出、日志、RAG 片段、文件、对话历史，在进入大模型之前先被压缩。今天这一篇换个方向，看 Headroom 怎么压模型写回来的东西。这部分能力叫 Output token reduction（输出 token 削减），对应的模块是 Output Shaper（输出整形器）。什么是输出 token 优化众所周知，输出 token 比输入贵，比如 Opus 级别的模型，输出 token 的单价大约是输入的 5 倍。同样是一万 token，写回来的那一万比发出去的那一万贵得多。所以输入侧压得再狠，如果模型每一轮都长篇大论地写回来，账单还是下不去。那模型的输出里，哪些部分是可以省的？官方 README 把浪费点归纳成三类：寒暄与收尾语（preamble / postamble）：回答正文前的一句 Great, let me...（好的，我来...），以及正文后的一句总结陈词。这些话对结果没有信息量。复述已有上下文：模型把你刚发给它的代码、文件内容、diff、工具输出，原样再抄一遍到回复里。你已经有这些内容了，它抄一遍纯属多花钱。对机械步骤过度思考（thinking）：thinking 指模型在正式作答前的一段推理草稿，它同样按输出 token 计费。当这一轮只是读了个文件、跑了个通过的测试，续写本来是很机械的动作，却还调动高档的深度思考，这笔思考 token 就浪费了。Headroom 代理本身不生成任何输出 token，它只是个透明反向代理。所以它能动的只有请求：通过改写发出去的请求，去影响模型愿意写回来多少。Output Shaper 的两根杠杆都是这个思路。打开输出整形器Output Shaper 默认是关闭的，通过环境变量开启：export HEADROOM_OUTPUT_SHAPER=1它的配置全部走环境变量，OutputShaperSettings 这个数据类负责从环境里把设置读出来，逻辑在 output_shaper.py：@dataclass(frozen=True) class OutputShaperSettings: enabled: bool = False # HEADROOM_OUTPUT_SHAPER verbosity_level: int = 2 # HEADROOM_VERBOSITY_LEVEL，0~4 effort_router_enabled: bool = True # HEADROOM_EFFORT_ROUTER mechanical_effort: str = "low" # 机械续写时降到哪一档 @classmethod def from_env(cls) -> OutputShaperSettings: enabled = runtime_env.getenv("HEADROOM_OUTPUT_SHAPER", "").lower() in ( "1", "true", "yes", ) # ... 读取 level、router、mech，并把 level 夹在 0~4 之间 return cls(enabled=enabled, ...)四个字段各对应一个环境变量：HEADROOM_OUTPUT_SHAPER：总开关，设成 1 / true / yes 才算开，其它值都是关。整个 Output Shaper 就靠它启用。HEADROOM_VERBOSITY_LEVEL：详略级别，0 到 4 的整数，默认 2。它控制第一根杠杆「详略引导」的力度，往系统提示词里追加多强的简洁指令，0 是不干预、4 是电报体，各级指令的原文下一节会看到。HEADROOM_EFFORT_ROUTER：第二根杠杆「努力档位路由」的开关，默认是开的，只有显式设成 0 / false / no 才关。它管的是另一件事：给机械续写的轮次降低思考档位。HEADROOM_MECHANICAL_EFFORT：机械续写时把思考档位降到哪一档，默认 low，填了不认识的档位名也会回退到 low。关于后三个环境变量的含义和用法，后面详讲。这个类只负责把开关读出来，真正的整形入口是 shape_request，它的主干很直白：def shape_request(body, settings=None, level_override=None) -> ShapeResult: if settings is None: settings = OutputShaperSettings.from_env() # 读环境变量里的配置 result = ShapeResult() if not settings.enabled: return result # 开关关着：原样返回，什么都不做 # 杠杆 1：详略引导 if level > 0 and apply_verbosity_steering(body, level): result.changed = True # 杠杆 2：努力档位路由 if settings.effort_router_enabled: kind = classify_turn(body.get("messages", [])) labels = route_effort(body, kind, settings) return result开关关着时它直接返回一个空结果，请求原样放行，所以默认行为和不装 Output Shaper 完全一样。打开之后，它的核心是两根杠杆：详略引导（verbosity steering）和 努力档位路由（effort routing）。下面分别看。第一根杠杆：详略引导详略引导的做法很朴素：在系统提示词的末尾追加一段指令，告诉模型简洁一点、别复述。指令文本按强度分成 5 个等级，0 级是不干预，1 到 4 级逐级变狠，都是写死在代码里的固定字符串。四个级别的原文分别是：第 1 级，只管寒暄：Skip preamble and postamble. Do not announce what you are about to do or recap what you just did; start with the substance.跳过开场白和收尾：不要预告你打算做什么，也不要复述你刚做了什么，直接进正题。第 2 级（默认级别），开始管复述：Skip preamble and postamble; start with the substance. Never restate code, file contents, diffs, or tool output that already appear in this conversation — reference them by path and line instead. After a tool call succeeds, continue without narrating the result.在第 1 级基础上加两条：对话里已经出现过的代码、文件内容、diff、工具输出，一律不许复述，改用路径和行号来引用；工具调用成功后直接往下写，不复述结果。第 3 级，连理由和改写幅度都管：Skip preamble and postamble. Never restate code, file contents, diffs, or tool output already in this conversation — reference by path and line. Give conclusions only; omit rationale unless the user asks why. Prefer the smallest edit over rewriting whole files. Keep prose to the minimum needed to be unambiguous.在第 2 级基础上再加码：只给结论，用户不问就不讲理由；改代码优先最小编辑，不要动不动重写整个文件；文字压缩到「不产生歧义」的下限。第 4 级，进入电报体：Minimum tokens. Fragments fine. No preamble, no postamble, no restating context, no rationale. Answer, smallest-possible edits, nothing else.token 能省则省：允许不完整的句子，开场、收尾、复述、理由一概不要，只要答案和最小改动，别的什么都别写。四个级别对比着看，约束是层层加码的：1 级管寒暄，2 级管复述，3 级管理由和改写幅度，4 级进入电报体。默认用第 2 级，是一个大多数人都能接受的力度。顺带一提，这段指令是追加在系统提示词最末尾（所有 cache_control 断点之后）的，这样不会弄坏缓存前缀，原理前面几篇讲过，这里就不展开了。第二根杠杆：努力档位路由第二根杠杆针对的是前面说的第三类浪费：对机械步骤过度思考。先解释一下背景。像 Claude Code 这样的编程 agent，在一个任务里会反复循环：调用工具、拿到结果、继续、再调用工具。这些循环里的绝大多数轮次，其实只是机械的续写，比如刚读完一个文件、刚跑过一个通过的测试，模型接下来做的事情是可预期的。但 Claude Code 这类客户端往往把每一轮的努力档位（output_config.effort）都钉在 xhigh 这样的高档上，于是模型对这种机械续写也调动高强度的思考，而思考是按输出 token 计费的。努力档位路由要做的，就是识别出这种机械续写的轮次，把它的思考档位调低；而对新问题、对报错，保持全力。判断哪种轮次靠的是 classify_turn，它的分类完全基于消息的结构，不看任何关键词、不用正则：def classify_turn(messages: list[dict]) -> TurnKind: last = messages[-1] if last.get("role") != "user": return TurnKind.UNKNOWN content = last.get("content") # ... saw_tool_result = False saw_error = False for block in content: btype = block.get("type") if btype == "tool_result": saw_tool_result = True if block.get("is_error") is True: saw_error = True elif btype == "text": return TurnKind.NEW_USER_ASK # 用户又插了一句新的话 elif btype in ("image", "document"): return TurnKind.NEW_USER_ASK if saw_error: return TurnKind.ERROR_CONTINUATION # 工具报错，要认真推理 if saw_tool_result: return TurnKind.MECHANICAL_CONTINUATION # 干净的工具结果，机械续写 return TurnKind.UNKNOWN逻辑很直白：看最后一条用户消息里都是些什么块。如果里面夹了用户新打的文字、图片或文档，说明用户又提了新要求，判为 NEW_USER_ASK；如果全是工具结果、且没有一个带 is_error 标记，判为机械续写 MECHANICAL_CONTINUATION；只要有一个工具结果带了错误标记，就判为报错续写 ERROR_CONTINUATION，因为模型这时候得认真琢磨这个失败。只有被判为机械续写的轮次，route_effort 才会动手降档：def route_effort(body, kind, settings) -> list[str]: if kind is not TurnKind.MECHANICAL_CONTINUATION: return [] # 其它情况一律不碰 labels = [] output_config = body.get("output_config") if isinstance(output_config, dict): effort = output_config.get("effort") if isinstance(effort, str) and effort in _EFFORT_RANK \ and _EFFORT_RANK[effort] > _EFFORT_RANK[settings.mechanical_effort]: output_config["effort"] = settings.mechanical_effort # 只往低调 labels.append(f"output_shaper:effort:{effort}->{settings.mechanical_effort}") # ... 老模型走 thinking.budget_tokens，夹到 API 下限 return labels这段代码里有两条安全规则：只降不加：只有当客户端本来就发了 output_config.effort 这个字段，整形器才会把它往低调。绝不主动注入这个字段，因为不支持努力档位的模型碰到这个参数会直接返回 400 报错。绝不动 thinking.type：对还在用老式 thinking.budget_tokens（思考预算 token 数）的模型，整形器只把预算夹到 API 允许的下限 1024，而从不去关掉思考开关。因为历史消息里如果带着思考块，中途关掉思考也会让某些模型报 400，而且这个开关一动就会破坏消息层的缓存。学习你偏好的啰嗦度上面的详略级别可以手动用 HEADROOM_VERBOSITY_LEVEL 指定，但更省心的方式是让 Headroom 自己学。命令是：headroom learn --verbosity # 只分析、给建议 headroom learn --verbosity --apply # 分析并写入配置这背后的逻辑在 headroom/learn/verbosity.py 文件中。它的出发点是一个观察：用户几乎从不明说自己想要多简洁的回答，但用户会用行为表现出来。这些行为信号能从 Claude Code 的会话记录里提取出来，Headroom 一共提取了四个信号：打断率、快速跳过率、长输出频率和复读率，其中决定详略级别的主要看前两个：打断率（interrupt rate）：模型话说到一半，用户按下打断的比例。快速跳过率（fast-skip rate）：模型给了个长回答，用户回得飞快，快到根本不可能读完。后两个不参与定级，但也各有用途：长输出频率（long-output rate）：长回答占全部回答的比例，「长」是相对的，以你自己的回答长度中位数为参照，且至少 200 词才参评。复读率（echo ratio）：回答和它拿到的上下文之间 n-gram 重叠的比例，看回答有多少是复述。它不定级，但它正是后文「度量输出节省」那节里直接浪费那一层的指标。快速跳过率的判定不是拿一个固定秒数当界限，而是先按平均阅读速度（每分钟 250 词）算出这段回答读完需要多久，如果用户在读完时间的一半都不到就回复了，就算一次快速跳过：_READING_WPM = 250.0 # 技术类文字的平均阅读速度（词/分钟） _SKIP_READ_FRACTION = 0.5 # 回复快于读完时间的一半，判为没读 _MIN_WORDS_FOR_SKIP = 150 # 太短的回答不参与判断，没什么可跳的然后把打断率和快速跳过率加起来当作「输出太多」的压力值，压力越大、建议的级别越高。它还设了个上限：即便压力非常大，也只封顶到第 3 级，而不会自动应用最激进的第 4 级电报体。数据不足（人类消息加上打断次数合计少于 10）时回退到默认的第 2 级。上面整个推荐是一个启发式先验，还可以加一个 --llm-judge 参数，配个大模型做裁判，它把提取好的四个信号（不是原始会话）发给 LLM，让它按四级标准给个级别。除此之外，还有一个自动调节控制器做运行时的实时微调（由 HEADROOM_VERBOSITY_AUTOTUNE 开启），学到的级别只是起点，它在会话进行中继续盯实时信号，动态调整级别。思路借自拥塞控制的 AIMD（加法增、乘法减）：连续多次「用户没在读」（打断、快速跳过）才把级别上调一级，往上探要慢；一旦用户嫌回答太少，立刻回退一级并冷却一段时间，压住不再轻易上调，因为惹恼用户是代价大的事件。这一步学到的级别会写进工作区的 verbosity.json 文件。运行时 resolve_verbosity_level 会按优先级取值：环境变量显式指定的手动值最高，其次是自动调节控制器，再次是学到的 verbosity.json，最后才是默认值。怎么度量输出省了多少到这里有一个绕不开的问题：输出侧到底省了多少 token，怎么算？输入侧好办。压缩是一个纯函数，压之前多少 token、压之后多少 token 都摆在那里，两个数一减就是省下的。但输出侧不一样。当整形器让请求变得更简洁，模型吐出了 N 个输出 token，可我们永远看不到它在没被整形的情况下本来会吐多少。这是一个反事实（counterfactual）问题：每个请求只会发生一种情况，另一种平行世界里的结果观察不到。反事实是因果推断里的概念，指的是「如果当初没这么做，结果会怎样」的那个没有真实发生的情形。它天然不可直接观测。所以，既然那个平行世界观测不到，节省就只能靠估计。headroom/proxy/output_savings.py 这个模块的工作，就是把估计做得诚实。它把结果分成三个层次：estimated（合成对照估计）：对照数据其实是由上节的 headroom learn --verbosity 命令输出，这条命令对历史会话时进行扫描，学习详略级别，另外，它还会顺带按请求特征把未整形时的输出 token 数累进一份逐层基线（baseline），--apply 时写进 output_savings.json。这份基线扫的是整形器上线之前的会话，相当于用历史数据拼出一个「假如没做整形会怎样」的假想对照组，这叫合成对照（synthetic control）。拿整形后实际观测到的输出，去和同类请求的基线均值相减，累加起来就是估计的节省。这个结果会带上置信区间，并且始终标注为「估计」，而不说是「测量」。measured（A/B 留出测量）：故意扣下一小撮对话不做整形，这叫留出集（holdout）；这些对话进对照臂（control arm，也叫对照组），其余进处理臂（treatment arm，也叫试验组）正常整形。两边同类请求的均值之差，是一个无偏的因果估计。这是唯一能被称为「测量」的数字。direct waste（直接浪费，无反事实）：复读率（echo ratio），就是上一节提取的四个行为信号之一，表示响应和上下文之间的 n-gram（连续 n 个词的片段）重叠比例。它是单个响应自身的属性，不需要反事实就能测。这里最关键、也最能体现设计诚实度的，是 measured 这一层。它是怎么留出对照组的呢？答案在 assign_arm 函数中，按对话做确定性分组：def assign_arm(conversation_key: str, holdout_fraction: float) -> str: if holdout_fraction <= 0.0: return "treatment" if holdout_fraction >= 1.0: return "control" digest = hashlib.sha256(("arm:" + conversation_key).encode()).hexdigest() frac = int(digest[:8], 16) / 0xFFFFFFFF # 映射到 [0, 1) return "control" if frac < holdout_fraction else "treatment"上面的 holdout_fraction 对应环境变量 HEADROOM_OUTPUT_HOLDOUT，默认取值为 0.1。就是把大约 10% 的对话留作对照组、不做整形；设成 0 或干脆不设，全部进处理组，那就没有 measured 数字可报。函数本身的逻辑很短。先处理两个边界：比例 ≤ 0 全部当处理组，≥ 1 全部当对照组。正常落在中间时，把对话 key 前面拼上固定前缀 "arm:"，做一次 SHA-256，再取哈希的前 8 个十六进制字符，除以 0xFFFFFFFF，映射成 [0, 1) 上的一个伪随机小数 frac。frac 落在 holdout_fraction 左边就进对照组，否则进处理组。因为哈希对同一个 key 永远出同一个数，所以分组是确定性的：同一条对话每次进来都会落进同一组，不会在中途变化。分组解决的是「做不做整形」，但光有组别还不够。不同请求天然就该吐出不同长度的输出：Opus 比 Haiku 啰嗦、带工具的机械续写比纯聊天短、输入 10 万 token 的上下文和输入 1 千 token 的也不一样。如果把所有请求的均值直接相减，混在一起的异质性会把因果效应淹没。所以还要分层（stratum）：把「同类」请求放进同一格，只在同一格里比处理组和对照组。分层用的特征必须在请求发出时就能观测到，绝不能看响应本身，否则就是用结果去分桶，因果估计就偏了。stratum_key 拼的是四样东西：def stratum_key(*, turn_kind, input_tokens, model, has_tools) -> str: return "|".join(( model_family(model), # opus / sonnet / haiku ... turn_kind, # 轮次类型 input_bucket(input_tokens), # xs / s / m / l / xl "tools" if has_tools else "notools", ))输入 token 数被故意划成很粗的几档（2k / 8k / 32k / 128k 为界），模型 id 也收成家族名。层划得太细，每一格样本会稀到基线噪声很大；粗一点，格子里才有足够的数可平均。合成对照的基线、A/B 的均值差，都是按这个 key 逐层算的：estimate = Σ (该层基线均值 − 该层观测输出)，或 A/B 里 Σ (该层对照组均值 − 该层处理组均值)。只有两边都有数据的层才参与 measured 的汇总。分组（进哪只臂）和分层（落进哪一格）的信息，全都通过已有的 transforms_applied 标签通道往下传，不用改动响应处理的任何路径。响应回来时，SavingsRecorder.record_from_labels 从标签里解出（组别，分层），把这一次的输出 token 记进对应格子的账本。估计的输出是带 95% 置信区间的，_finalize 用正态近似算出上下界：@staticmethod def _finalize(total_saved, total_baseline, var, n_requests, kind): pct = (total_saved / total_baseline * 100.0) if total_baseline > 0 else 0.0 se = math.sqrt(var) lo = total_saved - 1.96 * se # 95% 区间下界 hi = total_saved + 1.96 * se # ... 换算成百分比返回可以通过 headroom output-savings 命令查看详细的结果，它优先展示 measured 的数字，没有对照组数据时退回 estimated 的数字。小结这一篇我们学习了 Headroom 的输出 token 优化：为什么输出也值得压：输出 token 的单价约为输入的 5 倍，浪费集中在三处：寒暄与收尾语、复述已有上下文、对机械步骤过度思考。代理本身不生成输出 token，只能通过改写请求去引导模型少写。两根杠杆：详略引导往系统提示词末尾追加简洁指令，五个等级层层加码（1 级管寒暄、2 级管复述、3 级管理由和改写幅度、4 级电报体），默认 2 级；努力档位路由按消息结构识别机械续写、把它的思考档位调低，同时守住两条安全规则：只降不加、绝不动思考开关。学习偏好：headroom learn --verbosity 从打断率、快速跳过率等行为信号反推你想要的级别，启发式之外还能加 LLM 裁判，运行时还有 AIMD 控制器根据实时信号继续微调。诚实的度量：输出节省是反事实问题，Headroom 把它分成三层：合成对照给出的估计（带置信区间、只标注为估计）、留出对照组的 A/B 测量（HEADROOM_OUTPUT_HOLDOUT），以及不需要反事实的直接浪费（复读率）。至此，Headroom 的两个方向就凑齐了。把输入侧压缩和输出侧削减放在一起对比，正好作为整个系列的总结：维度输入侧压缩输出侧削减作用对象发给模型的内容模型写回来的内容手法直接缩小文本体积改写请求去影响模型行为节省是否可直接观测是，压前压后两个数一减否，是反事实，需估计或 A/B 测量单价权重输入 token，单价较低输出 token，Opus 级约为输入的 5 倍相关模块transforms/proxy/output_shaper.py、output_savings.py输入侧是把已经确定的文本压小，是个纯函数，干净利落。输出侧动不了模型本身，只能通过改写请求去引导它少写，效果天生带不确定性，所以配了一整套诚实的度量。写到这里，这个系列也差不多要告一段落了。回头看这条路线：第一篇介绍 Headroom 是什么、它要解决什么问题；第二篇把 headroom wrap 和 headroom proxy 跑起来；第三篇俯瞰架构，把 ContentRouter、三大压缩器、CacheAligner、CCR 和管线生命周期串成一张图；第四篇钻进 compress() 入口和管线编排的源码；第五篇细看 SmartCrusher、CodeAwareCompressor 和在本地推理的 Kompress；第六篇看 CCR 可逆压缩；第七篇看跨 agent 记忆和 headroom learn；到这最后一篇，把方向从输入侧翻到了输出侧。从压缩内容到压缩生成，从纯函数式的确定节省到反事实的诚实估计，Headroom 这套东西的价值不只在省钱的数字，更在它对待「省了多少」这件事的态度。希望这个系列能帮你把 Headroom 从一个命令行工具，理解成一套可以借鉴的工程思路。参考Headroom GitHub 仓库官方文档：Output token reduction官方文档：Failure learning官方文档：Proxy官方文档：Benchmarks
在上一篇里，我们从架构的角度把 Headroom 的压缩层俯瞰了一遍：内容进来之后先经过 ContentRouter 分流，再交给三大压缩器处理，中间还夹着 CacheAligner 和一整套贯穿全流程的生命周期事件。那一篇讲的是各个模块的分工和位置，属于总览。今天我们换个视角，钻进源码里，顺着一次 compress() 调用往下追，目标是把「入口函数收到一批消息之后，到底发生了什么」这条主线看清楚。我们从最外层的入口函数看起。compress() 入口Headroom 对外暴露的最简单用法就是一个函数：compress()，它定义在 headroom/compress.py 里。不需要起代理、不需要写配置，把消息传进去，拿回压缩后的消息就行。函数签名如下：def compress( messages: list[dict[str, Any]], model: str = "claude-sonnet-4-5-20250929", model_limit: int = 200000, optimize: bool = True, hooks: Any = None, config: CompressConfig | None = None, **kwargs: Any, ) -> CompressResult: ...几个参数值得留意：messages：一批消息，兼容 Anthropic 和 OpenAI 两种格式。model 和 model_limit：用于 token 计数和上下文窗口大小的判断，默认按 Claude Sonnet 的 20 万 token 上限算。optimize：是否真的压缩。传 False 会原样返回，方便做 A/B 对照（同一批消息，一边压一边不压，比较效果）。config 和 **kwargs：压缩选项，后者是前者的字段简写。先用传进来的 config（没有就取默认值），再拿 **kwargs 里对得上字段名的键去覆盖它。也就是说 compress(messages, protect_recent=0) 这种写法，等价于构造一个 CompressConfig(protect_recent=0)。CompressConfig：压缩到什么程度CompressConfig 是面向用户的压缩选项，控制「压什么、压多狠、用哪个模型」。它是一个 dataclass，几个默认值透露了 Headroom 的取向：@dataclass class CompressConfig: compress_user_messages: bool = False # 默认不压用户消息 compress_system_messages: bool = True # 默认压系统消息 protect_recent: int = 4 # 最近 4 条消息不动 protect_analysis_context: bool = True # 检测到分析/评审意图时保护代码 target_ratio: float | None = None # 保留比例，None = 模型自己决定 min_tokens_to_compress: int = 250 # 短于 250 token 的消息跳过 kompress_model: str | None = None # 文本压缩模型 ID savings_profile: str | None = None # 命名的高压缩档位dataclass 是 Python 标准库提供的一种类写法，专门用来定义「主要用来装数据」的类。只要在类上加一个 @dataclass 装饰器，它就会按你声明的字段自动生成 __init__、__repr__ 等方法，字段还能直接写默认值，省去手写字段赋值的样板代码。CompressConfig 这种纯配置类用它正合适。这些字段大致分三组。第一组决定压什么：compress_user_messages 和 compress_system_messages 分别控制用户消息、系统消息要不要参与压缩；protect_recent 把最后 N 条消息保护起来不动，因为它们是当前对话的活跃部分；protect_analysis_context 更进一步，一旦识别出「分析」「评审」这类意图，就会把相关代码保护起来不压。第二组决定压多狠：target_ratio 是 Kompress 的保留比例，设 None 就由模型自己定（默认激进，大约只留 15%）；min_tokens_to_compress 是参与压缩的最小 token 数，消息短于这个值就直接跳过，压缩本身有开销，太短不值得压。第三组决定用哪个模型、用哪套档位：kompress_model 是文本压缩用的模型 ID，默认是作者训练的 chopratejas/kompress-v2-base，可以换成 HuggingFace 上其它针对特定领域的模型，设为 'disabled' 则彻底关掉 ML 压缩；savings_profile 则是一套预设好的高压缩档位，它定义在 headroom/agent_savings.py，一共四个：档位保留比例压用户/系统消息protect_recent特点agent-9010%都压2最激进，强制走 Kompress，目标省 90%，面向 Codex/Claude/Cursor 这类 agentbalanced30%都不压4折中档，保护好用户和系统消息，目标省 70%coding模型自定都不压2代码负载档，不钉死比例，靠无损压缩和相关性出节省general模型自定都不压0通用档，代码少，没什么位置性内容要保护这四个档位分两路：agent-90 和 balanced 显式钉死了 target_ratio，直接规定保留多少；coding 和 general 不设这个比例，交给 Kompress 自己定，省多少主要看无损压缩和相关性过滤实际压掉多少。从这些默认值能看出，Headroom 出厂时是按「包裹编程 agent」这个场景调的：用户自己的消息（compress_user_messages=False）和最近的 4 条对话（protect_recent=4）都保护起来不压，因为它们是当前正在处理的活跃上下文；真正拿来开刀的是那些塞了大段工具输出、日志、检索结果的历史消息。CompressResult：结果长什么样压缩结果是另一个 dataclass，CompressResult：@dataclass class CompressResult: messages: list[dict[str, Any]] # 压缩后的消息，格式和输入一致 tokens_before: int = 0 # 压缩前 token 数 tokens_after: int = 0 # 压缩后 token 数 tokens_saved: int = 0 # 省下的 token 数 compression_ratio: float = 0.0 # 省下的比例，0.35 表示省了 35% transforms_applied: list[str] = field(default_factory=list) # 用过哪些变换关键的一点是 messages 的格式和输入完全一样，你可以直接把它塞回原来的 LLM 客户端调用里，无需改任何别的代码。transforms_applied 记录了这次实际跑过哪些变换（transform），后面排查「为什么没压」的时候很有用。拿到管线并执行配置就绪后，compress() 做的核心动作只有几行：pipeline = _get_pipeline() pipeline_extensions = PipelineExtensionManager(hooks=hooks, discover=False) # ... 发出 INPUT_RECEIVED 事件、抽取用户查询 ... result = pipeline.apply( messages=messages, model=model, model_limit=model_limit, context=context, biases=biases, compress_user_messages=cfg.compress_user_messages, compress_system_messages=cfg.compress_system_messages, target_ratio=cfg.target_ratio, protect_recent=cfg.protect_recent, # ... 把 CompressConfig 的字段透传给各个变换 ... )CompressConfig 里的字段在这里被摊平成一个个关键字参数，透传给管线，再由管线传给每个变换。这样每个变换都能看到「用户要不要压系统消息」「保护最近几条」这类全局意图。执行完之后还有一道 膨胀防护栏（inflation guard） 值得注意：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"], )如果「压缩」之后 token 数反而变多了（比如插入的标记比省下的还多），就直接回退到原始消息，并在 transforms_applied 里打上 inflation_guard:reverted 标记。压缩层的底线是绝不能帮倒忙。整个 apply() 外面还包了一层 try/except，任何异常都会记一次失败指标，然后原样返回输入消息。_get_pipeline：懒加载的单例_get_pipeline() 负责把管线装出来，它用的是单例（singleton，全进程只建一个实例）加线程锁的经典写法：def _get_pipeline() -> Any: global _pipeline if _pipeline is not None: return _pipeline with _pipeline_lock: if _pipeline is not None: return _pipeline from headroom.transforms import TransformPipeline # Default pipeline: CacheAligner → ContentRouter _pipeline = TransformPipeline() return _pipeline管线只在第一次调用时创建，之后复用同一个实例。这样管线里的压缩器只需加载一次，比如初始化 Rust 核心和 ML 模型，后面的每次 compress() 调用都直接复用，不用重复付出加载开销。TransformPipeline 的编排顺序管线的真身是 headroom/transforms/pipeline.py 里的 TransformPipeline。它的职责总结成一句话就是：按正确的顺序，把一串变换依次作用到消息上。默认装配哪些变换、什么顺序，由 _build_default_transforms() 决定：def _build_default_transforms(self) -> list[Transform]: transforms: list[Transform] = [] # 0. 工具结果拦截器（默认关闭，需 opt-in） if getattr(self.config, "intercept_tool_results", False) or \ os.environ.get("HEADROOM_INTERCEPT_ENABLED"): transforms.append(ToolResultInterceptorTransform()) # 1. Cache Aligner（前缀稳定，用于缓存命中） if self.config.cache_aligner.enabled: transforms.append(CacheAligner(self.config.cache_aligner)) # 2. 内容感知压缩：ContentRouter 处理所有内容类型 transforms.append(ContentRouter()) return transforms默认顺序就是注释里写的两步：先 CacheAligner，再 ContentRouter。顺序不能乱，CacheAligner 必须在前，因为它要在内容被改动之前先检查前缀的稳定性。从代码里可以看到，管道最前面还有一个工具结果拦截器，默认关着，要靠环境变量或配置显式打开，它是专门针对「工具结果」的一类可插拔重写器。它和后面那些通用压缩器思路不同：通用压缩器拿到什么内容就压什么，拦截器则是按「这是哪个工具的返回」来匹配，命中了才对这个工具的结果做定制化的改写。目前仓库里只有一个具体实现：ast-grep 拦截器（astgrep.py），它匹配 Claude Code 的 Read 工具，当读出的是个代码文件且足够大时，调用 ast-grep 把整份文件内容替换成一份函数级大纲，只留每个顶层函数和类的签名、将函数体省略掉。apply：逐个变换跑一遍TransformPipeline.apply() 是真正干活的地方。剥掉计时、追踪、日志之后，主循环很简洁：for transform in self.transforms: if not transform.should_apply(current_messages, tokenizer, **kwargs): continue try: result = transform.apply(current_messages, tokenizer, **kwargs) except Exception: self._breaker_record_failure() raise current_messages = result.messages all_transforms.extend(result.transforms_applied) # ... 累积标记、警告、计时 ...每个变换调用前先被问一句 should_apply()，如果条件不满足就跳过，满足了才 apply()，输出的消息再喂给下一个变换。所有变换用过的记录都累积到 all_transforms 里，最后进 CompressResult.transforms_applied。这里还藏着一个熔断器（circuit breaker）：self._breaker_threshold = _breaker_env("HEADROOM_PIPELINE_BREAKER_THRESHOLD", 3, int) self._breaker_cooldown_s = _breaker_env("HEADROOM_PIPELINE_BREAKER_COOLDOWN_S", 60.0, float)如果管线连续失败达到阈值（默认 3 次），熔断器就打开，在冷却窗口（默认 60 秒）内所有请求都原样透传、不再尝试压缩，避免每个请求都去重跑一遍注定失败的变换。窗口过后再自动恢复。有一次成功就把连续失败计数清零。ContentRouter：路由决策ContentRouter 是整套压缩的核心分流器，代码在 content_router.py。它的任务是：分析一段内容，判断它是什么类型，然后交给最合适的压缩器。类型判断的第一步是内容检测：mixed = is_mixed_content(content) detection = _detect_content(content) strategy = self._determine_strategy(content)_detect_content() 优先走 Rust 核心里的原生检测链（基于 Magika，Google 开源的一个用机器学习识别文件类型的小模型），在 Windows 上因为原生检测可能卡死，默认降级到纯 Python 的正则检测。检测的结果是一个内容类型，比如 JSON 数组、源代码、搜索结果、构建日志等等。拿到类型之后，_strategy_from_detection() 用一张映射表把类型翻译成压缩策略：mapping = { ContentType.SOURCE_CODE: CompressionStrategy.CODE_AWARE, ContentType.JSON_ARRAY: CompressionStrategy.SMART_CRUSHER, ContentType.SEARCH_RESULTS: CompressionStrategy.SEARCH, ContentType.BUILD_OUTPUT: CompressionStrategy.LOG, ContentType.GIT_DIFF: CompressionStrategy.DIFF, ContentType.HTML: CompressionStrategy.HTML, ContentType.TABULAR: CompressionStrategy.TABULAR, ContentType.PLAIN_TEXT: CompressionStrategy.TEXT, } strategy = mapping.get(detection.content_type, self.config.fallback_strategy) if (strategy == CompressionStrategy.CODE_AWARE and not self.config.prefer_code_aware_for_code): strategy = CompressionStrategy.KOMPRESS映射表一目了然：JSON 数组走 SmartCrusher，搜索结果走 SearchCompressor，日志走 LogCompressor，git diff 走 DiffCompressor，纯文本走文本压缩。表里没匹配上的走 fallback_strategy，默认是 Kompress（作者训练的文本压缩模型）。最后那个 if 值得注意：源代码本来映射到 CODE_AWARE，但因为默认配置里 prefer_code_aware_for_code=False，代码实际上被改道去了 Kompress。也就是说，默认情况下 Headroom 宁可让代码走通用文本压缩，也不轻易对代码做 AST（抽象语法树） 级别的改写，避免误伤。这一层的取舍我们留到下一篇讲三大压缩器时再展开。上面说的是一段内容只属于一种类型的情况。如果一段内容里既有代码块、又有 JSON、还夹着散文，is_mixed_content() 会判定它是混合内容，走 _compress_mixed() 这条路：先用 split_into_sections() 把内容按代码围栏、JSON 块、搜索结果行拆成一段段带类型的片段，每段各自选策略压缩，最后再拼回去。这样一段图文并茂的工具输出里，代码归代码压、JSON 归 JSON 压，互不干扰。# headroom/transforms/content_router.py（精简） def _compress_mixed(self, content, context, ...): sections = split_into_sections(content) # 拆成带类型的片段 for section in sections: strategy = self._strategy_from_detection_type(section.content_type) # 每段各选策略 compressed_content, ... = self._apply_strategy_to_content( section.content, strategy, context, ...) if section.is_code_fence and section.language: # 保留代码围栏标记，如 ```python compressed_content = f"```{section.language}\n{compressed_content}\n```" compressed_sections.append(compressed_content) return RouterCompressionResult( compressed="\n\n".join(compressed_sections), # 拼回去 strategy_used=CompressionStrategy.MIXED, ...)整个路由决策的流程可以画成这样：CacheAligner：守住缓存前缀ContentRouter 之前那一步是 CacheAligner，代码在 cache_aligner.py。要理解它，得先说清楚它守的是什么。Anthropic、OpenAI 这些服务商都支持 prompt 前缀缓存（prefix cache）：如果两次请求的开头一大段内容一模一样，服务商可以复用上一次算好的 KV cache（注意力机制里键值对的缓存，命中后这段内容几乎不重复计费），省钱又省时间。但缓存命中有个苛刻的前提：前缀必须逐字节稳定。只要系统提示词开头掺了一个会变的值，比如一个时间戳、一个会话 ID，缓存就会整段失效。CacheAligner 的职责就是盯住这个隐患。在 Headroom 的早期版本中，它会去改写系统提示词，把动态内容抽出来重新插到别处。这个思路听上去挺顺：发模型之前先把易变内容摘掉，服务商缓存的就是干净前缀，下次再摘一次，不就命中了？但问题恰恰出在「摘」这个动作上。缓存命中的唯一标准是两次转发出去的前缀逐字节一致，而易变内容的位置、边界、和前后文字的关系并不固定，这次摘成这样、下次可能摘成那样，只要有一字之差缓存照样失效；把内容重新插到别处，插入的位置和格式又成了新的不稳定点。也就是说，越想靠中途改写在热区里制造稳定，越是在热区里不断制造新的字节变化，反而把缓存弄坏。源码里把这叫做违反了「缓存热区（系统提示词）绝不能被改动」的不变式，于是那条改写路径被彻底移除了。现在它是一个纯检测器：只发现问题、发警告，从不改消息。正确的解法不是代理在中途帮你摘，而是从源头就别把易变内容放进系统提示词，把它挪到用户消息之类的地方。这样系统提示词天生就是逐字节稳定的，根本不需要谁来摘。所以 CacheAligner 检测到动态值时，给的是「把这些值挪出系统提示词」的建议，动手的决定权留给使用者。检测的产物是 VolatileFinding（易变内容发现记录）：@dataclass(frozen=True) class VolatileFinding: label: str # 类型标签：uuid / iso8601 / jwt / hex_hash sample: str # 截断后的样本，绝不记录完整内容detect_volatile_content() 会把系统提示词切成 token 逐个分类，识别出四类易变内容：UUID、ISO 8601 时间戳、JWT 令牌、十六进制哈希。检测全程不用正则，而是靠结构化的解析器，比如用标准库的 uuid.UUID 去试解析、用 datetime.fromisoformat 去试时间戳，形状对得上才算数。一旦发现易变内容，就打印出一条警告信息：if all_findings: counts = {} # 统计每类各多少个 # ... msg_text = ( f"CacheAligner: detected volatile content in system prompt " f"({counts_str}); cache prefix unstable. " "Move dynamic values out of the system prompt to recover cache hits." ) warnings.append(msg_text) logger.warning(msg_text)小结这一篇我们跟着一次 compress() 调用，把 Headroom 压缩管线的主干走了一遍：入口 compress()：解析 CompressConfig（默认按编程 agent 场景调，保护用户消息和最近 4 条），跑完管线拿到 CompressResult，中间有膨胀防护栏和异常兜底，绝不帮倒忙。_get_pipeline() 与 TransformPipeline：懒加载的单例管线，默认顺序是 CacheAligner → ContentRouter。管线里还带连续失败熔断。ContentRouter 路由决策：先检测内容类型（原生 Magika 链，Windows 降级到正则），再查映射表选压缩器；代码默认改道走 Kompress；混合内容拆片段分别压。CacheAligner：一个纯检测器，用结构化解析（非正则）找出系统提示词里的 UUID、时间戳、JWT、哈希这类易变内容，发警告提示缓存前缀不稳，但从不改写提示词。整体链路还是比较清晰的，至此，我们已经了解了「一段内容被送到哪个压缩器」这条路由主线，但每个压缩器内部到底怎么把 token 压下来，还没拆开。下一篇我们就深入三大压缩器：处理 JSON 的 SmartCrusher、AST 感知的 CodeAwareCompressor，以及跑在 Rust 核心里的 Kompress 文本压缩模型。参考Headroom GitHub 仓库Headroom 架构文档Headroom 压缩原理文档SmartCrusher 文档代码压缩文档Kompress-v2-base 模型卡
在上一篇里，我们看了 Headroom 的 CCR（Compress-Cache-Retrieve，可逆压缩）：压缩时把原文按哈希缓存在本地，模型觉得信息不够，就拿着哈希把原文取回来。它管的是「一次会话内」的信息不丢。这一篇我们看另外两块和「记住事情」有关的能力：一个是跨会话、跨 agent 的共享记忆，你在 Claude Code 里定过的偏好、积累的经验，能不能让 Codex、Gemini 下次也用上；另一个是 headroom learn，它会翻你过去的编程会话记录，自动找出反复踩的坑，把纠正写进各 agent 的上下文文件里。这两块的源码分别在 headroom/memory/ 和 headroom/learn/ 目录下。headroom memory：跨 agent 的共享记忆在第二篇的学习里，我们其实已经见过记忆的命令了，运行 headroom wrap claude --memory，代理会在流量里自动注入和提取记忆，什么都不用改。想在自己的代码里用，库提供了一个包装函数 with_memory()：from openai import OpenAI from headroom import with_memory # 一行套上，之后照常使用 client = with_memory(OpenAI(), user_id="alice") # 第一个会话：随口告诉它你的偏好 client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "I prefer Python for backend work"}] ) # 之后换一个全新会话： client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": "What language should I use?"}] ) # 回答会引用上个会话记下的 Python 偏好效果就是：第一个会话里你只随口说了句偏好，新会话里模型就能据此回答。这背后 with_memory() 在每次调用里做了三件事：第一件是注入相关记忆：按当前的用户消息做语义检索，把查到的相关记忆拼进消息发给模型。拼的位置有讲究，不是塞进系统提示词，而是拼到第一条用户消息里。系统提示词是缓存热区，动了会让 prompt cache 整段失效，这和之前的 CacheAligner 是同一个考量。第二件是指导模型如何记忆：Headroom 会在系统提示词里加一段固定指令，大意是，如果这轮对话里有值得长期记住的事实（用户偏好、身份、当前目标这类可复用的信息），就在回答之后输出一个 <memory> 块；寒暄、一次性问题、已经知道的信息不要记。格式是 XML 标签包一段 JSON：<memory>{"memories": [{"content": "..."}]}</memory>，没什么可记的，模型就输出空的 {"memories": []}。第三件是解析并保存记忆：Headroom 拿到响应后把这个块解析出来，逐条调 add() 存进记忆库里（下一节介绍）；然后把块从响应里剥掉再返回，你看到的回答是干净的。这整个提取内联在同一次调用里完成，没有额外的 API 调用。攒下的记忆可以用 headroom memory 系列命令直接管理：headroom memory list # 看存了哪些记忆 headroom memory list --scope USER # 只看 user 级（跨会话持久）的 headroom memory list --since 7d # 最近 7 天的 headroom memory stats # 统计 headroom memory export --output backup.json # 导出备份 headroom memory prune --older-than 30d # 清理 30 天前的记忆的使用比较简单，下面看它内部是怎么组织的。分层记忆上面说到，Headroom 从响应里解析出 <memory> 块之后，会逐条调 add() 把事实存进记忆库。这个 add() 就是记忆系统的核心入口，它所在的类是 core.py 里的 HierarchicalMemory。它把持久化存储、向量索引、全文索引、Embedding、缓存这些组件拼成一套统一的记忆 API，外界只用跟它打交道。看它的实现，一条记忆是怎么进来的：async def add(self, content, user_id, session_id=None, agent_id=None, turn_id=None, importance=0.5, ..., auto_bubble=None): memory = Memory(content=content, user_id=user_id, session_id=session_id, agent_id=agent_id, turn_id=turn_id, importance=importance, ...) if auto_embed: memory.embedding = await self._embedder.embed(content) # 生成向量 await self._store.save(memory) # 落库 if memory.embedding is not None: await self._vector_index.index(memory) # 建向量索引 await self._index_for_text_search(memory) # 建全文索引 should_bubble = auto_bubble if auto_bubble is not None else self._config.auto_bubble if should_bubble: await self._maybe_bubble(memory) # 重要记忆上浮 return memory可以看到，一条记忆加进来，流程分四步：先给内容生成向量，然后把记忆本体落库，最后同时建两个索引，向量索引管语义检索，全文索引管关键词精确检索。这四步各自对应 headroom/memory/adapters/ 下的适配器，而且每一步都留了可替换的后端：步骤适配器后端选择生成向量embedders.pysentence-transformers（默认，需要 torch 较重）、ONNX（推荐的轻量项，无需 torch、约 86MB）、OpenAI、Ollama落库存储sqlite.py内置只有 SQLite，第三方存储可走 entry point 插件接入向量索引sqlite_vector.py / hnsw.py默认自动选：有 sqlite-vec 就用它，否则回退 hnswlib全文索引fts5.pySQLite FTS5（BM25 排序 + Porter 词干），也可走外部插件向量索引管语义，全文索引管字面，两个配合起来就是常说的混合检索。生成向量：sentence-transformers 是 HuggingFace 生态里的老牌向量库，把一段文本编码成向量，效果好但要带 torch，体积较大；ONNX 是跨框架的模型推理格式，同一个模型改用它跑就不用装 torch，体积小很多，所以被标为推荐。OpenAI 和 Ollama 则是把向量生成外包给 API 或本地 Ollama 服务。落库存储：SQLite 是嵌在进程里的本地数据库，不用单独起服务，记忆这种单机数据用它正合适。Entry points 是 Python 的插件发现机制，第三方包注册后能被自动找到，存储、索引这些后端都靠它对外开放。向量索引：sqlite-vec 是 SQLite 的向量检索扩展，向量存在数据库文件里，查询走页缓存，记忆条数再多内存占用也不涨。hnswlib 是 HNSW 算法的实现，HNSW（Hierarchical Navigable Small World）是一种近似最近邻算法，能在大量向量里快速找到语义最接近的几条，检索快，但整个图索引都在内存里，随条数涨。全文索引：FTS5（Full-Text Search 5） 是 SQLite 自带的全文检索扩展，直接在数据库里建全文索引，不用额外部署搜索引擎。BM25 是信息检索里经典的相关性打分算法，关键词命中越多、越稀有，分越高；Porter 词干提取则把 running、runs、ran 这类英文变形归到同一个词干 run，搜索按词干匹配，命中更全。细心的读者会发现，add() 这个函数的签名里有 user_id、session_id、agent_id、turn_id 这一串参数，这是 Headroom 记忆系统的另一个特点 —— 分层作用域（scope）。一条记忆可以挂在四层里的任何一层：user → session → agent → turn，从宽到窄。注意这里的 agent 不是指「某个 agent 应用」，而是当前会话里的一个 agent 实例，一个会话里可以有多个 agent（比如主 agent 和它派生的 subagent），每个实例的存活期都在会话内部，所以 agent 比 session 窄。四层按存活期理解：user 跨所有会话，session 只管当前这一个会话，agent 管会话里某一个实例，turn 是单次 LLM 调用。函数末尾的 _maybe_bubble 是配合分层的另一个重要机制：记忆上浮。它解决的问题是：一条记忆是在某个具体会话里产生的（session 级），会话一结束它就跟着没用了，可有些记忆明明值得长期留下。上浮的判断很直接，当重要性 importance 达到阈值（bubble_threshold，默认 0.7）的，就复制一份提到 user 级：副本的 session_id、agent_id、turn_id 全部清空，从此对这个用户的所有会话可见，同时记下 promoted_from（从哪条记忆升上来的）和 promotion_chain（提升链条），方便溯源；原记忆还留在原处不动。这样普通记忆随会话消亡，真正重要的少数会自己升上去，越攒越多。记忆的双向同步上面这套语义记忆是 Headroom 自己的存储，但各家编程 agent 其实都有自己的 markdown 记忆文件：Claude Code 的 MEMORY.md、Codex 的 AGENTS.md、Gemini 的 GEMINI.md。Headroom 支持在这两个世界之间做双向同步。一方面是导入侧，本质上也是一次 add() 调用，Headroom 把 markdown 解析成段落，按标题层级算出重要性，打上来源标签后写进来；另一方面是导出侧，就是反过来，把 Headroom 里新增加的记忆回写到 markdown 文件里，这个我们下一节再看。对于 Claude Code，你可能更熟悉它的项目上下文文件 CLAUDE.md，而 Headroom 记忆同步的是它的自动记忆文件 MEMORY.md，放在 ~/.claude/projects/<项目>/memory/ 下，每次启动时前 200 行会常驻进上下文。双向同步的核心逻辑位于 bridge.py 的 MemoryBridge 如下：async def sync(self, paths=None, user_id=None) -> SyncStats: # 阶段 1：把 markdown 里新增 / 改动的段落导入 Headroom 语义记忆 stats.import_stats = await self.import_from_markdown(paths=paths, user_id=user_id) # 阶段 2：把 Headroom 里新增的记忆导出回 markdown new_memories = await self._get_new_organic_memories(user_id, since) if new_memories and paths: count = await self._append_to_markdown(Path(paths[0]).expanduser(), new_memories) self._sync_state["last_sync"] = datetime.now(timezone.utc).isoformat() self._save_sync_state()导入侧靠基于哈希的变更检测，每个文件、每个段落都存了内容哈希，没变的直接跳过，避免重复导入；导出侧只挑「原生记忆」，也就是 Headroom 自己生成的，而不是从 markdown 导进来的那些。具体靠元数据里的 source 标签把导入来的过滤掉，防止同一条内容在两边来回导。async def _get_new_organic_memories(self, user_id, since=None): # ... # 过滤掉 metadata.source == source_tag 的记忆（那些是当初从 md 导入的） if metadata.get("source") == self._config.source_tag: continue各 agent 的写入器导出回 markdown 时，不同 agent 的文件格式不一样，这部分由 writers/ 下的一组写入器分别处理。它们共享一个基类 AgentWriter，通用的处理都放在基类里：按「重要度 × 新近度 × 访问次数」排序、按内容哈希去重、按 token 预算截断、用注释标记包裹自己管理的段落。# writers/base.py MARKER_START = " " MARKER_END = " " def export(self, memories, output_path=None, dry_run=True): ranked = sorted(memories, key=lambda m: m.score, reverse=True) # 排序 # 按 content_hash 去重 # 按 token 预算截断 formatted = self.format_memories(budgeted) # 子类实现格式 section = f"{MARKER_START}\n{formatted}\n{MARKER_END}" # 包进标记 full_content = _merge_section(target, section) # 只替换标记内部标记（marker）在这里划定了回写边界：写入器只碰 和 之间的内容，你自己在文件里手写的部分不会动。子类只需实现 format_memories（怎么排版）和 default_path（写到哪）。以 claude_writer.py 为例，这两个方法长这样：# writers/claude_writer.py（精简） def format_memories(self, memories: list[MemoryEntry]) -> str: """Format as Claude Code MEMORY.md section.""" lines = ["## Headroom Learned Context", "*Auto-maintained by Headroom — do not edit manually*", ""] # 按 category 分组，每组一个 ### 小标题，记忆逐条列成列表项 grouped: dict[str, list[MemoryEntry]] = defaultdict(list) for m in memories: grouped[(m.category or "General").replace("_", " ").title()].append(m) for heading, entries in grouped.items(): lines.append(f"### {heading}") for entry in entries: lines.append(f"- {entry.content}") lines.append("") return "\n".join(lines) def default_path(self) -> Path: """Default: Claude Code project memory directory.""" if self._memory_dir: return self._memory_dir / "MEMORY.md" # ~/.claude/projects/-<sanitized-path>/memory/MEMORY.md sanitized = encode_claude_project_path(self._project_path) return Path.home() / ".claude" / "projects" / sanitized / "memory" / "MEMORY.md"可以看到子类要做的就这两件事：format_memories 负责排版，把记忆按类别分组、每组一个小标题、逐条列成列表项；default_path 负责路径，返回的正是上面说的那个 ~/.claude/projects/<项目>/memory/MEMORY.md。剩下的排序、去重、截断、合并进标记块，都由基类包办了。几个写入器的差异也集中在这两点上：claude_writer.py：写 Claude Code 的 MEMORY.md。它的 token 预算默认 2000，因为 Claude Code 只把 MEMORY.md 的前 200 行常驻上下文；超出的高重要度记忆会被 export_topics 按主题分别写进独立文件，按需加载，不挤占那 200 行的预算。codex_writer.py：写 Codex 的 AGENTS.md，纯 markdown 无 frontmatter，默认预算 3000。cursor_writer.py：写 Cursor 的 .cursor/rules/*.mdc，带 YAML frontmatter（文件头部的元数据块）。generic_writer.py：兜底写入器，输出纯 markdown，任何读 markdown 上下文文件的 agent 都能用。它的 default_path 默认写到项目根目录的 HEADROOM_MEMORY.md，文件名可以传参指定。Gemini 没有专门的写入器，就可以用它，把文件名传成 GEMINI.md 即可。整个记忆同步的数据流如下图所示：headroom learn：从失败会话里学经验记忆是「你告诉它什么，它记什么」。headroom learn 更主动一点：它去翻你过去的编程会话记录，自动找出反复踩的坑，把纠正写进各 agent 的上下文文件里，下次这个坑就不会再踩。headroom learn 默认是空跑（dry-run，只演示不落盘），加 --apply 才真正写文件：headroom learn # 空跑，只看会给出什么建议 headroom learn --apply # 落盘写入 headroom learn --project ~/my-project --apply # 分析指定项目 headroom learn --agent codex --all # 分析所有 Codex 会话 headroom learn --target CLAUDE.md # 改写进团队共享文件不过我第一次在一个项目上直接跑 headroom learn 就失败了，报 403 forbidden：这是因为 learn 对 claude 模型会绕过 ANTHROPIC_BASE_URL、直接请求官方 api.anthropic.com（本意是防止它指向本地代理），而我的 ANTHROPIC_API_KEY 是配给 MiniMax 这类第三方端点的，拿着它调用官方 API 自然被拒。改成 --model claude-cli 走本机 CLI（它会继承第三方端点配置）才跑通：headroom learn --model claude-cli跑通后，真实输出是这样：扫了 207 个会话、8614 次工具调用，其中 486 次失败（5.6%），给出 10 条建议：它给出的不是「Read 失败了 5 次」这种泛泛的统计，而是具体的纠正。官方文档里把这项机制叫做成功关联（Success Correlation）：它不只是记录失败，还会找出模型后来是怎么修好的。比如这次运行里学到的一条经验：失败：在 manager/backend 子目录里执行 git add manager/...，报错 pathspec did not match（路径被拼成了 manager/backend/manager/...）；后来成功：先 cd 到仓库根目录再 add；学到的经验：git 命令一律在仓库根目录执行。注意每条建议后面都跟着一个节省估算，这个数不是 LLM 拍脑袋估的，而是后面要讲到的循环检测实测出来的浪费下界。学到的模式大致分几类：防循环（上面这种）、环境事实（该用哪条命令）、路径纠正、搜索范围、命令模式、已知大文件。下次会话 agent 启动时读到这些，同类错误就不会再犯。写入位置默认是 CLAUDE.local.md（个人的、gitignored），想写进团队共享的 CLAUDE.md 就加 --target 参数。Scanner → Digest → LLM → Recommendations上面这些建议看着简单，背后的问题却不小：207 个会话、8614 次调用，怎么从里面找出值得学的模式？headroom learn 的答案是一条四步流水线，analyzer.py 开头的文档字符串写明了它的立场：# Pipeline: Scanner (events) → Digest Builder → LLM → Recommendations # No regex patterns, no static lookback windows, no hardcoded heuristics. # A single LLM call understands the full conversation context and produces # structured recommendations for CLAUDE.md / MEMORY.md.这段说的是：不用正则、不用固定回看窗口、不用硬编码启发式规则，一次 LLM 调用理解完整对话上下文，产出结构化建议。这条流水线按字面就是四步，每步的职责是：Scanner（扫描）：从磁盘上把 agent 的会话记录读出来。以 Claude Code 为例，读的是 ~/.claude/projects/<项目>/ 下的 JSONL 会话日志，把每一次工具调用（名字、入参、成功还是失败、token 数）和用户消息解析成结构化事件。每种 agent 一个插件（plugins/ 下的 claude、codex、gemini），运行输出里那行 Detected agents 就是这一步探测到的。Digest Builder（摘要）：几百个会话、几千次调用不可能全塞给模型，这一步把它们压成一份 token 预算内（约 8 万 token）的文字摘要：项目概况和总数（多少会话、多少调用、失败率）、检测到的循环放在最前面（最贵的浪费模式，附实测浪费）、之前已学到的模式、以及每个会话精简后的事件流（报错截断、保留成功标记和用户消息）。它就是喂给 LLM 的那份「证据包」。LLM（分析）：只发一次调用。系统提示词把它设定成「分析 coding agent 会话、提取能防止 token 浪费的模式」的专家，并给了明确的优先级，循环最高，往下是环境规则、文件结构事实、用户偏好、失败模式、工作流规则；用户消息就是那份摘要。返回结构化 JSON。Recommendations（建议）：JSON 被解析成一条条 Recommendation，每条带着写到哪个文件（CLAUDE.local.md 还是 MEMORY.md）、具体内容和估计节省的 token。之后 apply_loop_weighting 用循环的实测浪费校准估算值，按节省降序排好，交给 writer.py 落盘。所以判断「哪些是该学的经验」这件事本身，是交给一个 LLM 去做的，而不是用一堆正则去套。扫描和摘要都是确定性的机械工作，只有「从证据里提炼模式」这一步交给模型。SessionAnalyzer.analyze 就是把这四步串起来：def analyze(self, project, sessions) -> AnalysisResult: all_calls = [tc for s in sessions for tc in s.tool_calls] failed_calls = [tc for tc in all_calls if tc.is_error] loops = detect_loops(sessions) # 先检测循环 if not failed_calls and not loops and not any(s.events for s in sessions): return result # 没失败、没循环、没事件，直接返回 digest = _build_digest(project, sessions, loops=loops) # 拼成紧凑摘要 model = self.model or _detect_default_model() # 自动选模型 raw = _call_llm(digest, model) result.recommendations = _parse_llm_response(raw) # 解析成建议 apply_loop_weighting(result.recommendations, loops) # 按实测浪费加权 result.recommendations.sort(key=lambda r: r.estimated_tokens_saved, reverse=True) return result值得注意的是，这里的模型调用走的是 LiteLLM 这个统一接口，不管你使用的是什么模型，一次 completion() 调用即可。如果不用它，就得为每家各写一套 SDK 的直调、各处理一套鉴权和响应格式。具体用哪个模型，由下面这个顺序决定：显式指定 --model 优先级最高，它的取值有两类：任意 litellm 模型名（100 多家 provider 任选），或三个本机 CLI 标识（转给本机对应的 CLI 做分析，支持 claude-cli / gemini-cli / codex-cli）；环境变量里有 API key，按一张写死的映射表来取：ANTHROPIC_API_KEY → claude-sonnet-4-6、OPENAI_API_KEY → gpt-4o、GEMINI_API_KEY → gemini/gemini-flash-latest；一个 key 都没有：看 HEADROOM_LEARN_CLI 环境变量指定的 CLI；如果还没有：自动探测本机装了的 CLI 工具（claude > gemini > codex），让订阅用户不用另配 API key 也能用。检测错误循环headroom learn 把循环列为重点，因为一次性错误只浪费一次，而循环的浪费随重复次数累加。循环在流水线里被处理了两次：模型调用之前，detect_loops 把它检测出来、连着实测浪费一起写进摘要，让模型看得到；模型调用之后，apply_loop_weighting 再拿这份实测浪费去校准建议里的估算。先看 detect_loops，它主要检查两种循环：错误循环：同一个调用失败、重试、又失败。比如反复去读一个根本不存在的路径。RTK 重取循环：RTK（Rust Token Killer，第二篇介绍过的 shell 输出压缩工具）把 grep foo 改写成 grep foo | head -50，结果截断掉了 agent 真正要的内容，agent 只好换个变体再跑一遍（head -100、换偏移量）。每次调用都成功，所以纯看失败的分析根本发现不了它。关键技巧是把这些变体折叠成同一个规范签名（signature），再数重复次数、算实测浪费的 token：def _canonical_signature(tc: ToolCall) -> str: raw = tc.input_summary.strip() if tc.name.lower() in ("bash", "shell"): raw = _PAGINATION_RE.sub(" ", raw) # 去掉 | head -50 / limit 100 这类分页片段 raw = _INT_RE.sub("N", raw) # 裸数字统一替换成 N raw = _WS_RE.sub(" ", raw).strip().lower() return f"{tc.name.lower()}::{raw}"这样 grep foo | head -50 和 grep foo | head -100 就归成了同一个签名。默认要重复满 3 次才算循环，这是能把「循环」和「一次性重试」区分开的最小次数。浪费的 token 是实测下界，不是让 LLM 猜的：错误循环里每次都算浪费，重取循环里第一次是正当工作、只算后面 N-1 次的重取。检测发生在模型调用前，校准发生在模型调用后。模型返回建议之后，apply_loop_weighting 会把与某个循环签名重叠的建议的 estimated_tokens_saved 抬到至少等于该循环实测浪费的 token。因为循环的实测浪费是多次累加的，这一步能可靠地把「防循环」的建议排到「防一次性错误」的建议前面，而不必指望 LLM 自己把权重估对。把纠正写进文件模型给出建议、再经循环的实测浪费校准估算之后，流水线就剩最后一步：把建议写进文件，让下次会话的 agent 能读到。这一步由 writer.py 负责。它同样用标记块（ / ）圈出自己的地盘，只动块内的内容。至于写到哪个文件，ClaudeCodeWriter 的默认目标不是 CLAUDE.md，而是 CLAUDE.local.md：def _resolve_context_path(self, project): if self._context_target is not None: # --target 显式指定的话，它说了算（比如想写进团队共享的 CLAUDE.md） ... if project.project_path == Path.home(): return claude_config_dir() / "CLAUDE.md" # 主目录下的是个人全局记忆 return project.project_path / "CLAUDE.local.md" # 项目级默认写个人文件Claude Code 约定 CLAUDE.md 是团队共享，会提交进 git 仓库，而 CLAUDE.local.md 是个人的，默认被 gitignore 忽略，学到的模式一般都是「个人」的，所以默认写进 CLAUDE.local.md 文件。如果你用的是别的 agent，目标文件也会跟着换：Codex 是 AGENTS.md，Gemini 是 GEMINI.md，这套映射由 headroom/learn/plugins/ 下各自的插件提供，这里不再赘述。小结这一篇我们学习了 Headroom 的两块「记忆」能力：跨 agent 记忆：按 user -> session -> agent -> turn 四层作用域组织，普通记忆随会话消亡，重要的会自动上浮到用户级、跨会话越攒越多；检索同时走向量索引（HNSW）和全文索引（FTS5）两条路；它还能和各 agent 的 markdown 记忆文件双向同步：导入按哈希检测变更、只挑改动的段落，导出只回写自己新增的记忆，并用标记块圈定边界、绝不碰手写的内容。headroom learn：走 Scanner → Digest → LLM → Recommendations 的流水线，判断该学什么这件事交给模型而不是正则；loops.py 把错误循环和 RTK 重取循环折叠成规范签名、按实测浪费加权；writer.py 默认把纠正写进 gitignore 的 CLAUDE.local.md，也支持 AGENTS.md 和 GEMINI.md。到这里，关于 Headroom 模型输入这一侧的内容就基本讲完了。不过省 token 还有另一半没讲：模型输出的那部分。同样一个问题，模型可以啰嗦地复述一大段，也可以简洁作答，输出 token 一样要计费。Headroom 的 Output Shaper 就是冲着这半边来的。我们下一篇看它怎么削减输出 token，也给这个系列收个尾。参考Headroom GitHub 仓库Memory 记忆系统文档SharedContext 跨 agent 共享文档Failure learning 失败学习文档LiteLLM 统一多家 LLM 接口的库
在上一篇里，我们学习了 Headroom 的三大压缩器：处理 JSON 数组的 SmartCrusher、基于 tree-sitter 做 AST（抽象语法树） 感知的 CodeAwareCompressor，以及作者自训、在本地推理的文本压缩模型 Kompress。它们能把工具输出、日志、代码片段压掉一大半，token 随之骤减。不过压缩到这一步，有个绕不开的问题：压缩是会丢信息的。SmartCrusher 把 100 条搜索结果压成 10 条，剩下的 90 条并没有进模型的上下文。万一模型看完这 10 条，发现真正想要的答案在第 47 条上，怎么办？如果没有补救手段，那压缩省下的 token 就是以「模型可能答错」为代价换来的。Headroom 给这个问题的答案叫 CCR（Compress-Cache-Retrieve，可逆压缩）：压缩时把原文在本地缓存起来，同时告诉模型「你要是觉得不够，可以来取」。CCR 原理解析要实现压缩的「可逆」，必须得靠两样东西：一是原文不丢，按一个哈希键存进本地缓存（Python 侧默认是 ~/.headroom/ccr_store.db 这个 SQLite 库，Rust 侧还提供内存和 Redis 后端）；二是压缩产物里带上这个哈希键的标记，模型凭它知道「不够可以来取」。标记的形态不止一种。标准格式是方括号这样的：[100 items compressed to 10. Retrieve more: hash=a1b2c3d4e5f6a1b2c3d4e5f6]上一篇 SmartCrusher 压 JSON 数组时用的是另一种行内标记，作为一个哨兵元素嵌在数组末尾：[ {"ts": "10:00:01", "level": "INFO", "msg": "worker started"}, {"ts": "10:04:59", "level": "ERROR", "msg": "connection refused"}, {"_ccr_dropped": "<<ccr:a1b2c3d4e5f6 2_rows_offloaded>>"} ]两种形态作用都一样：告诉模型原文在哪、怎么取。模型如果只看压缩后的内容就够了，那什么都不用做，省下的 token 落袋为安；只有当它判断信息不够时，才拿着这个 hash 回来取原文。CCR 模块的职责定义在 headroom/ccr/__init__.py 文件里，分成四块：# 1. Tool Injection: 压缩发生时，代理往请求里注入 headroom_retrieve 工具 # 2. Response Handler: 拦截响应，自动处理模型发起的 CCR 工具调用 # 3. Context Tracker: 跨轮追踪被压缩的内容，按需主动展开 # 4. Batch Processing: 处理批量 API 结果里的 CCR 调用我们挨个看这四块，它们合起来覆盖了实时和异步两种场景下的完整取回链路。注入 retrieve 工具模型要能主动取原文，前提是它手里得有这么一个工具可用。这件事由 tool_injection.py 负责。它的核心是一个工具定义 create_ccr_tool_definition：CCR_TOOL_NAME = "headroom_retrieve" # Anthropic 格式（OpenAI / Google 各有一份，字段结构略有不同） { "name": CCR_TOOL_NAME, "description": ( "Retrieve original uncompressed content that was compressed to save tokens. " "Use this when you need more data than what's shown in compressed tool results. " # 取回被压缩掉的原始内容。当压缩结果里的数据不够用时调用它。 ), "input_schema": { "type": "object", "properties": { "hash": {"type": "string", "description": "Hash key from the compression marker"}, }, "required": ["hash"], }, }工具只有一个参数 hash，也就是压缩标记里那串哈希。注入的时机由 CCRToolInjector 控制，它的 scan_for_markers 会扫一遍请求里的所有消息，用一组正则去匹配各个压缩器留下的标记：_marker_patterns = [ # 标准格式: [N items compressed to M. Retrieve more: hash=xxx]（24 位十六进制哈希） re.compile(r"\[(\d+) \w+ compressed to (\d+)\. Retrieve more: hash=([a-f0-9]{24})\]"), # SmartCrusher 的行内标记: <<ccr:HASH ...>>（12~24 位） re.compile(r"<<ccr:([a-f0-9]{12,24})\b"), # ... 省略若干兼容旧格式的正则 ]扫到标记就说明这一轮里有压缩内容，inject_tool_definition 会把 headroom_retrieve 追加进请求的工具列表。这里有个容易忽略的细节：一旦某个会话用过一次 CCR，后续每一轮都会粘性地保留这个工具，哪怕当前轮没有新的压缩标记：def inject_tool_definition(self, tools, *, session_has_done_ccr=False): # session_has_done_ccr=True 时，即使本轮没有新标记也照样注入 if not (session_has_done_ccr or self.has_compressed_content): return tools or [], False # 已经存在（比如来自 MCP server）就不重复注入 for tool in tools or []: if (tool.get("name") or tool.get("function", {}).get("name")) == CCR_TOOL_NAME: return tools, False # ...为什么要粘性保留？因为工具列表的字节一旦在会话中途变来变去，就会打破 Anthropic、OpenAI 的 KV cache，工具列表是缓存前缀的一部分，忽有忽无会让缓存整段失效。这一点和之前讲 CacheAligner 时是同一个考量：稳定前缀，让缓存真正命中。拦截响应、自动取回工具注入进去了，模型也调用了，可这个调用是发给谁的？模型以为它在调一个正常的工具，实际上这个工具由 Headroom 的代理自己兜住。这部分逻辑在 response_handler.py 的 CCRResponseHandler 里。它的入口是 handle_response，拿到模型的响应后先判断里面有没有 CCR 调用，有就进入一个取回循环：async def handle_response(self, response, messages, tools, api_call_fn, provider="anthropic"): current_response = response current_messages = list(messages) rounds = 0 while rounds < self.config.max_retrieval_rounds: # 默认最多 3 轮 ccr_calls, other_calls = self._parse_ccr_tool_calls(current_response, provider) if not ccr_calls: break # 没有 CCR 调用，收工 if other_calls: break # 混了别的工具调用，交回客户端处理 rounds += 1 results = [self._execute_retrieval(call) for call in ccr_calls] # 本地取原文 current_messages.append(self._extract_assistant_message(current_response, provider)) current_messages.append(self._create_tool_result_message(results, provider)) current_response = await api_call_fn(current_messages, tools) # 带着原文续问一轮 return current_response这个循环的意思是：模型说「我要 hash=abc 的原文」，handler 就去本地缓存把原文捞出来（_execute_retrieval），拼成一条工具结果消息，替模型把对话续上，再发一轮 API 请求。模型这轮拿到了完整原文，通常就能给出真正的答案。整个过程对最终客户端是透明的，客户端只会收到最后那条不带 CCR 调用的响应。_execute_retrieval 有两个细节。一是取回按哈希整块取回，返回的是完整原文，不做二次筛选：entry = store.retrieve(ccr_call.hash_key) if entry: content = json.dumps({ "hash": ccr_call.hash_key, "original_content": entry.original_content, "original_item_count": entry.original_item_count, }, indent=2) return CCRToolResult(tool_call_id=..., content=content, success=True, ...)二是缓存有 TTL 存活时长，过期就取不到了。这时 handler 会把失败状态原样返回给模型，让它知道这块内容已经不可用：if entry_status is not None and entry_status["status"] != "available": content = json.dumps({ "error": format_retrieval_miss_detail(entry_status), "hash": ccr_call.hash_key, "status": entry_status["status"], # 比如 expired "ttl_seconds": ..., }, indent=2) return CCRToolResult(..., success=False)整条 CCR 取回流程串起来如下图所示：还有一种边界情况处理得很谨慎：如果模型在同一轮里既调了 headroom_retrieve、又调了别的正常工具，handler 会直接跳出循环、交回客户端。因为一条 assistant 消息里的每个工具调用都要有配对的工具结果，而 handler 只有 CCR 那部分的结果，硬拼一个续问请求会得到非法的消息序列。那交回客户端之后，headroom_retrieve 这个调用谁来兑现呢？其实客户端自己就能兑现。取回工具有两条分发渠道：一条是上面讲的代理注入、由 handler 在代理侧自动兑现；另一条是 MCP server，也就是第二篇里 wrap 注册的那个 headroom MCP 服务，它把 headroom_retrieve 作为真正的工具挂在客户端上。流式响应（streaming）也有对应的 StreamingCCRBuffer，思路是先缓冲、扫到 CCR 调用就切换成非流式处理，取回后再把续写流式吐出去。这里不展开。跨轮主动展开前两块解决的是「模型主动来取」。但还有一种情况：早几轮压掉的内容，模型其实已经忘了它存在。比如第 1 轮搜索返回 100 个文件、压成了 10 个，到第 5 轮用户问「那认证中间件呢」，模型压根不知道 auth_middleware.py 曾经出现在那被压掉的 90 个里。context_tracker.py 就是来补这一环的。ContextTracker 会把每次压缩事件记下来，包括哈希、发生在第几轮、压缩前后的条数、当时的查询上下文，还有一段样本内容用于后面做相关性匹配。等新一轮用户消息进来，analyze_query 会拿这条查询去和历史压缩内容算相关性，够高就主动把对应原文展开：def _calculate_relevance(self, query, context) -> float: query_words = set(self._extract_keywords(query.lower())) score = 0.0 # 和样本内容的关键词重叠 sample_words = set(self._extract_keywords(context.sample_content.lower())) if sample_words: score += len(query_words & sample_words) / len(query_words) * 0.5 for word in query_words: # 长词命中额外加分 if len(word) >= 4 and word in context.sample_content.lower(): score += 0.2 # 和当时查询上下文的关键词重叠 # ... 省略 return min(score, 1.0)相关性算法本身是朴素的关键词重叠，加了几条加权：长词的精确子串命中额外给分、越旧的压缩内容按时间打折（age_factor）、超过 5 分钟的直接不考虑。超过阈值 0.3 的才会进推荐列表，每轮最多主动展开 2 条。批量 API 里的取回前面三块讲的都是实时请求：模型一响应，handler 当场拦截、当场续问。但还有一种调用方式不走这条路 —— 批量 API，比如 Anthropic 的 Message Batches API、OpenAI 的 Batch API、Gemini 的 Batch API 等。批量 API 的玩法是：客户端把 N 个请求打包成一个数组一次性提交，拿到一个 batch ID（类似一个任务号），之后拿它轮询，直到全部跑完，再一次性拉回 N 份结果。批量请求里的内容同样会被 Headroom 压缩，因此模型同样可能会发起 headroom_retrieve 调用。但这条链路和实时调用完全不同，根本不存在「模型响应回来、代理当场拦截」的那一刻，Response Handler 是专为实时调用写的，自然用不上。batch_processor.py 的 BatchResultProcessor 就是补这个场景的，分两步配合。提交批量时，先把每个请求的上下文（消息、工具列表）按 batch ID 存进 BatchContextStore；等客户端拿 batch ID 把 N 份结果全部拉回来时，处理器先按 batch ID 取出存好的上下文，再逐份扫描，发现哪份里有 CCR 调用就对哪份动手：从本地缓存取回原文、拼上工具结果、发起续问调用（最多 3 轮），最后把这份只有工具调用、没有答案的半成品结果，替换成带完整答案的结果。注意这个续问调用不是再发起一次批量，而是普通的实时调用：Anthropic 走 /v1/messages，OpenAI 走 /v1/chat/completions，Gemini 走 generateContent。道理很简单：N 份结果里带 CCR 调用的往往就那么几份，为这几份再排一次异步队列、再等一轮，远不如直接同步调用快。三个服务商的批量结果格式各不相同，但这套"检测 → 取回 → 续问 → 替换"的逻辑是完全一样的。到这里，ccr/__init__.py 里说的四块就齐了：实时请求靠 Tool Injection 和 Response Handler，跨轮遗忘靠 Context Tracker，异步批量靠 Batch Processing。小结这一篇我们把 Headroom 的 CCR 可逆压缩读完了，四块组件都看完了：CCR 的原理：压缩时原文不丢、按哈希缓存在本地，压缩产物带上标记，模型信息不够时凭哈希取回。传输有损，端到端无损。三个组件串成实时取回流程：tool_injection.py 往请求里注入 headroom_retrieve 工具（并粘性保留以护住 KV cache），response_handler.py 拦截并自动兑现模型的取回调用、带着原文续问一轮，context_tracker.py 跨轮追踪压缩内容、按查询相关性主动展开。取回工具有代理注入和 MCP server 两条分发渠道，混合调用时客户端走 MCP 自己兑现。批量 API 里的取回：异步批量请求赶不上实时续问，由 batch_processor.py 在结果回来时补做"检测 → 取回 → 续问 → 替换"。下一篇我们看输入侧的另外两块拼图：跨 agent 的共享记忆，以及从失败会话里学经验的 headroom learn。参考Headroom GitHub 仓库CCR 可逆压缩文档Anthropic Message Batches API 文档OpenAI Batch API 文档Gemini Batch API 文档
在上一篇中，我们跟着 compress() 走完了整条压缩管线。当时看到 ContentRouter 会先识别一段内容到底是 JSON、代码还是普通文本，再把它交给对应的压缩器处理。这三类文本对应的压缩器分别是 SmartCrusher、CodeAwareCompressor 和 Kompress，今天我们就来逐个学习下这三大压缩器。SmartCrusher：统计式压缩 JSON 数组工具调用返回的内容里，最常见的一种形态是一个很长的 JSON 数组，每个元素结构都差不多。比如调用一个 API 接口拉回来的一批记录，或者 Docker 的 JSON 日志一行一个对象，每条都是同一套字段。模型真正需要的往往是头部几条和尾部几条，中间几十条高度雷同的记录既占 token，又没带来新信息。SmartCrusher 做的就是统计这个数组的结构规律，保留最有代表性的若干条，把其余的丢进 CCR（Compress-Cache-Retrieve，可逆压缩）缓存。这个压缩器的 Python 实现已经在最新版本中整个搬到了 Rust。打开 smart_crusher.py，文件顶部有一行注释：所有数组压缩现在都走 headroom._core.SmartCrusher，也就是从 crates/headroom-py 编译出来的 Rust 扩展。Python 侧只留下了配置类和一层薄薄的转发。所以要讲清它的工作原理，得直接看 Rust 那边的实现。按数组分类选压缩策略SmartCrusher 不是一上来就压，它先让一个分析器把这个数组摸清楚。具体的做法是把数组里所有项的字段名合并起来，逐个字段统计一遍，给每个字段算一份 FieldStats。这份统计大致长这样：// crates/headroom-core/src/transforms/smart_crusher/types.rs（精简） pub struct FieldStats { pub field_type: String, // numeric / string / boolean / object / array pub count: usize, // 这个字段出现了多少次 pub unique_count: usize, // 有多少个不同的值 pub unique_ratio: f64, // 不同值占比 = unique_count / count pub is_constant: bool, // 是不是所有项都一样 // 数值字段专有：min_val / max_val / mean_val / variance / change_points // 字符串字段专有：avg_length / top_values（按频率排的高频值） }可以看出统计的维度挺有讲究：unique_ratio 看一个字段的取值有多「重样」，比如日志的 level 字段翻来覆去就 INFO、ERROR 几个值，这个比例就很低；而 message 字段每条都不太一样，比例就高。数值字段额外记方差和变化点（数值突然跳变的位置），字符串字段额外记平均长度和最高频的几个值。统计完每个字段，下一步把这些 FieldStats 翻译成这个数组「是什么类型」：有时间戳字段、数值字段又有方差的，是 time_series；同时存在一个高基数的文本字段（像 message）和一个低基数的级别字段（像 level）的，是 logs；有字段被判定为「像分数」的，是 search_results；都不沾边的，归为 generic认完类型，select_strategy 再据此选对应的压缩策略：// crates/headroom-core/src/transforms/smart_crusher/analyzer.rs（精简） pub fn select_strategy(&self, field_stats, pattern, item_count, ...) -> CompressionStrategy { if item_count < self.config.min_items_to_analyze { return CompressionStrategy::None; // 数组太短，不压 } if pattern == "time_series" && has_change_points { return CompressionStrategy::TimeSeries; // 时间序列，盯变化点 } if pattern == "logs" && message_field.unique_ratio < 0.5 { return CompressionStrategy::ClusterSample; // message 字段大量重复 → 聚类 } if pattern == "search_results" { return CompressionStrategy::TopN; // 搜索结果 → 取 top N } CompressionStrategy::SmartSample // 通用数组 → 智能抽样 }这个函数就是「类型 → 策略」的一张映射表，但有两个细节值得留意。一是它带前置门槛：数组元素太少（少于 min_items_to_analyze）直接返回 None 不压，不值得为一个短数组费这个劲。二是日志那档多加了一道确认：光认出 logs 类型还不够，还要 message 字段的 unique_ratio < 0.5（一半以上的值是重复的）才真的走聚类，因为聚类压的就是重复模板，如果 message 每条都不同，聚类就没意义了。认完类型、过完这些门槛，才知道该用哪种思路去挑要保留的项。按策略挑出要保留的下标归好类，就生成一个压缩计划 CompressionPlan。它的核心是一份 keep_indices，也就是要保留的原数组下标清单，但完整定义里还带了一些执行时要用的信息：// crates/headroom-core/src/transforms/smart_crusher/types.rs pub struct CompressionPlan { pub strategy: CompressionStrategy, // 用哪种策略 pub keep_indices: Vec<usize>, // 要保留的原数组下标 pub constant_fields: BTreeMap<String, Value>,// 所有项都相同的字段，可抽出来只留一份 pub summary_ranges: Vec<(usize, usize, Value)>, // 被归纳成摘要的区间 pub cluster_field: Option<String>, // 日志聚类时按哪个字段分簇 pub sort_field: Option<String>, // 排序/取 top N 时按哪个字段 pub keep_count: usize, // 保留多少条 }keep_indices 是这份计划的主体，挑选它的信号有好几路：搜索结果（TopN）：数组里那个「像分数」的字段（比如每条命中自带的相关度分），按它从高到低排序、取前 N 条。注意这个分数是数据自带的，跟你当下查什么没关系，所以它只适合本身就有排序依据的搜索结果。日志（ClusterSample）：把内容模板相同的行聚成簇，日志往往同一句报错刷几十遍，只有时间戳在变，每个簇只留一条代表，其余的去掉，重复的刷屏就压掉了。通用保底：带报错词的行、取值异常稀有的行，强制保留。报错词是一份写死的清单（error、exception、crash、timeout 等十几个），只要一条里出现其中任何一个就保留。取值稀有针对的是另一类情况：有些异常不带 error 字样，比如一个状态码字段 95 次是 ok、只有 5 次是各种错误码，这些低频值本身就说明不正常，含它们的行也要保留。两道兜底都是为了防止主策略（按相关性取 top N、聚类留代表）把最关键的报错和异常给淘汰掉。查询相关性：这一路会看查询上下文（也就是用户当前的问题），先拿它做确定性的关键词精确匹配，再用一个相关性打分器给每条和这句话的相关程度打个分，超过阈值就保留。这个打分器是「BM25 关键词匹配 + 向量语义相似度」的混合：BM25 看共享多少关键词，向量看语义上有多接近，两者按权重融合，遇到 UUID、ID 这类需要精确匹配的查询还会自动调高关键词的权重。它是叠加在前面所有策略上的，让最终的保留清单向你当前关心的问题倾斜。位置锚点：保留头部、尾部各一小部分，保证开头结尾总有代表。这几路信号挑出来的下标合并去重，就是最终保留的那批。所以 SmartCrusher 不是简单地「掐头去尾、中间全删」，真正的核心是先认类型，再按类型用相关性打分、聚类、异常检测这些手段选出最有代表性的若干条，头尾那部分只是其中一路保底。生成 CCR 标记挑出要丢的行之后，还不能一删了之。crush_array 末尾会把完整的原数组序列化一次、算出哈希、存进 CCR 缓存，然后生成一个指向这个哈希的标记：// crates/headroom-core/src/transforms/smart_crusher/crusher.rs（精简） let dropped_count = items.len() - result.len(); if dropped_count > 0 && self.config.enable_ccr_marker { let canonical = canonical_array_json(items); // 完整原数组 let h = hash_canonical(&canonical); let marker = format!("<<ccr:{h} {dropped_count}_rows_offloaded>>"); if let Some(store) = &self.ccr_store { store.put(&h, &canonical); // 原文按哈希存进缓存 } // marker 会被放进输出，模型凭它把原文取回来 }这个 <<ccr:HASH N_rows_offloaded>> 标记就是可逆的关键：原文按哈希存在本地，模型之后发现信息不够，拿这个哈希就能把丢掉的那 N 行原样取回来。压前 vs 压后举个直观的例子。假设一次日志查询返回了这样一个数组（这里精简到 4 条示意，实际可能是几十上百条）：[ {"ts": "10:00:01", "level": "INFO", "msg": "worker started"}, {"ts": "10:00:02", "level": "INFO", "msg": "worker started"}, {"ts": "10:00:03", "level": "INFO", "msg": "worker started"}, {"ts": "10:04:59", "level": "ERROR", "msg": "connection refused"} ]SmartCrusher 去重、抽样之后，可能得到：[ {"ts": "10:00:01", "level": "INFO", "msg": "worker started"}, {"ts": "10:04:59", "level": "ERROR", "msg": "connection refused"}, {"_ccr_dropped": "<<ccr:a1b2c3d4e5f6 2_rows_offloaded>>"} ]可以看到，输出仍然是原数组里的元素，schema 完全没变。中间两条重复的 INFO 被丢掉了，末尾多出来一个 _ccr_dropped 哨兵对象，它不是真正的记录，只是给模型看的一个提示：这里省了 2 行，需要的话可以用 CCR 取回。文件里还专门提供了 strip_ccr_sentinels 函数，方便下游遍历数组时把这个哨兵过滤掉，免得当成正常记录处理。官方给出的一组真实压缩基准里，代码搜索场景（100 条结果）从 17,765 token 压到 1,408，省了 92%；SRE 事故排查场景从 65,694 压到 5,118，同样是 92%。这类高度结构化、大量重复的数组，正是 SmartCrusher 最擅长的场景。CodeAwareCompressor：AST 感知的代码压缩第二位处理的是源代码。代码和 JSON 数组不一样，它有严格的语法，随便截断几行就可能变成一段无法解析的乱码。CodeAwareCompressor 的核心承诺是：压缩后的代码一定仍然是语法有效的。它的做法是先把代码解析成 AST（抽象语法树）。然后保留 import 语句、函数签名和类型注解这些结构性骨架，只压缩函数体内部的实现细节，最后再拼回一段合法代码。这套思路参考了一篇叫 LongCodeZip 的论文（ASE 2025），它专门解决代码的长上下文压缩。通用的文本剪枝方法（比如 LLMLingua）不理解代码的结构和依赖，效果有限，LongCodeZip 则按代码的结构来压。做法是分两阶段：先做粗粒度，把代码按函数切成块，用「相对你的指令的条件困惑度」给每个函数打分、只留和任务最相关的函数；再做细粒度，把留下的函数体内部再切成更小的块，按 token 预算挑出最相关的子集。论文报告在不掉任务表现的前提下最高能压到 5.6 倍。CodeAwareCompressor 保留签名、按重要性分配函数体预算，和它是同一路数，只不过落地时换成了 tree-sitter 静态解析，不用真的去跑困惑度。tree-sitter 是一个增量式的代码解析库，能把多种语言的源代码解析成统一的语法树结构，很多编辑器用它来做语法高亮和代码折叠。code_compressor.py 的整体流程可以画成一张图：用数据表描述每种语言CodeAwareCompressor 支持不少的编程语言：第一梯队是 Python、JavaScript、TypeScript，第二梯队还有 Go、Rust、Java、C、C++ 和 Perl。要为这么多语言各写一套抽取逻辑，代码会非常臃肿。Headroom 的做法是把每种语言的差异抽成一张数据表 LangConfig，让同一套通用逻辑去查表：@dataclass(frozen=True) class LangConfig: import_nodes: frozenset[str] # 哪些 AST 节点算 import function_nodes: frozenset[str] # 哪些算函数定义 class_nodes: frozenset[str] # 哪些算类定义 type_nodes: frozenset[str] # 哪些算类型定义 body_node_types: frozenset[str] # 哪些算函数体 comment_prefix: str # 注释前缀，Python 是 #，C 系是 // uses_colon_after_signature: bool # Python 签名后跟冒号，C 系跟花括号 # ...以 Python 为例，它的配置就是把 import_statement、function_definition、class_definition 这些 tree-sitter 节点类型分门别类填进去。结构抽取时只有一个通用的 visit 访问器遍历语法树，遇到某个节点就查 LangConfig 判断它属于哪一类，完全不需要为每种语言重复写方法。压缩函数体压缩的关键在 _compress_function_ast。它拿到一个函数节点后，先从 AST 里精确定位函数体，再按分配到的行数预算保留若干条完整语句：# 按 AST 里的语句逐条保留，绝不从表达式中间切断 for start_row, end_row in body_stmts: stmt_lines = code_lines[start_row : end_row + 1] stmt_line_count = len(stmt_lines) # 加上这条就超预算，且已经留了至少一条，就停在这里 if kept_line_count + stmt_line_count > body_limit and stmts_kept > 0: break kept_lines.extend(stmt_lines) kept_line_count += stmt_line_count stmts_kept += 1这里的关键设计是按语句而不是按行截断。它遍历的是 AST 里的语句节点，每一个都是完整、合法的语句，保留到预算用完为止。这样无论砍到哪里，剩下的代码都能正常解析，不会出现半句 if 或者没闭合的括号。每个函数保留多少行，不是平均分配的，而是由 _analyze_symbol_importance 打分决定。这个方法综合了几个信号：一个符号被引用了多少次、它调用了多少别的函数（扇出）、它是不是公开符号、名字是否命中了当前查询的上下文。被引用得多、和查询相关的函数，会分到更多的行数预算，实现细节保留得更完整；边角料函数则可能只剩个签名。docstring（文档字符串）默认走 FIRST_LINE 模式，多行 docstring 只保留第一行摘要并正确闭合引号，剩下的说明文字全部压掉。三道安全阀compress() 方法里有三处保护，任何一处不满足都会原样退回，绝不输出坏代码：语法校验：拼回代码后再用 tree-sitter 解析一遍，只要出现 ERROR 或 MISSING 节点就退回原文。过度压缩保护：如果压缩比低于 0.05（也就是只剩 5% 不到），判定为压得太狠、可能丢了数据，退回原文。异常兜底：AST 压缩过程中抛任何异常，都退回原文或转交 Kompress。压前 vs 压后用一个 Python 函数来感受一下（仿照源码顶部的示例，docstring 扩充成了多行）：import os from typing import List def process_data(items: List[str]) -> List[str]: """Process a list of items. Each item is validated and, when non-empty, normalized by stripping whitespace and lowercasing. Invalid (falsy) items are skipped. The normalized items are collected in order and returned as a new list. """ results = [] for item in items: # Validate item if not item: continue # Process valid item processed = item.strip().lower() results.append(processed) return results压缩后变成：import os from typing import List def process_data(items: List[str]) -> List[str]: """Process a list of items.""" # ... (body compressed: 10 lines → 2 lines) passimport 一行不动，函数签名连同类型注解 List[str] 完整保留，多行 docstring 只留了第一行摘要、后面那段详细说明被压掉了，函数体那一大段实现也只剩一行占位注释。对于一次代码检索返回的多个文件，模型光看签名和类型往往就够判断该看哪个函数了；真要深入某个函数的实现，再通过 CCR 把原文取回来即可。压缩显著时，CodeAwareCompressor 还会在末尾追加一条注释，写明省了多少 token、CCR 的 hash 是多少、多久过期。Kompress：跑在本地的 ModernBERT 压缩模型前两个压缩器面对的都是结构清晰的内容：JSON 有 schema，代码有语法。可现实里还有大量没有明显结构的文本，比如报错栈、RAG 检索回来的文档片段、大段的对话记录。这类内容 SmartCrusher 和 CodeAwareCompressor 都使不上劲，这时就该 Kompress 上场了。Kompress 是作者专门训练的一个文本压缩模型，托管在 HuggingFace 上，模型 ID 是 chopratejas/kompress-v2-base。和前两个基于规则的压缩器不同，它是一个真正的神经网络模型，逐个 token 判断该保留还是丢弃。双头 ModernBERTKompress 是在一个叫 ModernBERT 的开源模型之上实现的，先简单认识一下这个模型。ModernBERT 是 Answer.AI 团队（联合 LightOn 等）在 2024 年底发布的 BERT 现代化版本，论文叫《Smarter, Better, Faster, Longer》。相比 2018 年的原版 BERT，它把后来主流大模型的不少新技术搬了过来：用 RoPE 旋转位置编码替换了老式绝对位置编码，原生支持 8192 token 的上下文（是 BERT 512 的 16 倍），长文本处理速度是同级编码器的两三倍，还首次把代码数据纳入预训练，所以在代码相关任务上格外强。这些特点对 Kompress 很关键：日志、文档动辄几千 token，上下文不够长就放不下；压缩又跑在代理热路径上，速度慢了会拖垮请求。Kompress 用的 base 版有 1.49 亿参数，每个 token 的向量维度是 768。它是一种编码器模型，作用是读完一段文字后，给里面每个 token 都算出一个向量，这个向量捕捉的是这个 token 在上下文里的含义。比如 apple 在「吃了一个 apple」和在「apple 发布了新手机」里，算出来的向量是不一样的，因为模型看了它前后文。这一步解决的是「理解」：模型由此知道每个 token 在当前语境里是什么意思。但理解归理解，这些向量本身只是一堆数字，还没回答「这个 token 要不要留」。要回答这个问题，需要在编码器的输出之上再接一个判断层，也就是所谓的头（head）。头本身不大，就是把 768 维的 token 向量映射成你想要的答案，具体用什么层随任务而定。以 Kompress 的 token 头为例，它是一个 768 → 2 的线性层：把某个 token 的向量乘上一个学好的权重矩阵，输出「该丢」「该留」两个得分，比一下大小就有了去留。训练时，骨干（ModernBERT）和头一起被优化：给模型看成堆标注好「哪些 token 该留」的文本，不断调整权重，直到头的判断越来越准。之所以叫「头」，是相对于「骨干」而言的，骨干负责通用的语言理解、可以原样复用，换任务时往往只需换掉或新训顶上这个小小的头，不必重训整个大模型，这是迁移学习里常见的做法。Kompress 正是拿 ModernBERT 当骨干负责理解，再在顶上接两个这样的头，所以叫「双头」。Kompress 源码解读打开 kompress_compressor.py，模型结构定义在 HeadroomCompressorModel 里：class HeadroomCompressorModel(nn.Module): """Dual-head ModernBERT: token classification + span importance CNN.""" def __init__(self, model_name="answerdotai/ModernBERT-base"): super().__init__() self.encoder = AutoModel.from_pretrained(model_name, ...) # 加载 ModernBERT hidden_size = self.encoder.config.hidden_size # 向量维度 768，两个头都以它为输入 # 头 1：768 → 2 的线性层，逐 token 输出「该丢」「该留」两个得分 self.token_head = nn.Linear(hidden_size, 2) # 头 2：一维卷积，评估一小段连续区域的重要性 self.span_conv = nn.Sequential( # 每次看 5 个相邻 token，提取 256 个局部特征；padding 保证输出与输入等长 nn.Conv1d(hidden_size, 256, kernel_size=5, padding=2), nn.GELU(), # 非线性激活，否则两层卷积等价于一层 # 窗口收窄到 3，把 256 个特征聚成每个位置 1 个重要性分 nn.Conv1d(256, 1, kernel_size=3, padding=1), nn.Sigmoid(), # 压进 0~1，变成可直接跟阈值比较的分数 )这两个头分工不同。token 头看的是单个 token：把某个 token 的 768 维向量映射成两个数，分别代表「该丢」和「该留」的得分，比较一下大小就知道这个 token 去留。span 头看的是一小段连续区域：用两层一维卷积把相邻 token 的向量扫一遍，输出一个 0 到 1 之间的分数，表示「这一整片内容重不重要」。为什么用两层而不是一层？这是因为一层卷积只是对窗口内几个 token 做线性加权，学不会需要组合特征才能识别的复杂模式；先展开成 256 个特征通道、配上非线性再聚合，才有能力先提取局部特征、再据此打分。一维卷积（Conv1d）可以理解成一个在序列上滑动的「小窗口」。它拿一个固定宽度的窗口，从文本开头往结尾一格一格地滑，每滑到一个位置，就把窗口里那几个相邻 token 的向量合在一起算出一个值。和线性层只看单个 token 不同，卷积这个窗口一次看好几个相邻的 token，所以它捕捉的是「这一小片」的局部特征，而不是孤立的某个词。窗口宽度由 kernel_size 决定，Kompress 的 span 头第一层窗口是 5、第二层是 3，两层叠起来，输出每个位置实际覆盖前后共 7 个 token。那为什么要两个头呢？一个不够吗？关键在「拿不准」的情况。模型对很多 token 的去留其实很纠结，单独看它，留也行丢也行。这时候光看单个 token 容易误删，span 头提供的就是一个「看大局」的补救：如果某个 token 处在边界地带（保留概率在 0.3 到 0.5 之间、模棱两可），但它所在的这一片被 span 头判为重要（得分超过 0.5），那就宁可把它留下来。实际的判定逻辑就一行：# kompress_compressor.py（精简） keep = token_keep | (borderline & span_boost) # token_keep：token 头明确说留 # borderline：这个 token 模棱两可（保留概率 0.3~0.5） # span_boost：它所在的片段被判为重要（span 得分 > 0.5）翻译过来就是：token 头明确说留的留；模棱两可、但所在片段重要的，也留。 其余的就丢掉。这样单个 token 判断的「细」和片段判断的「稳」就互补上了，不容易因为某个 token 单独看不显眼就把它误删，从而把一整句关键信息拆断。必留 token 的硬性保险光靠模型打分还不够。有些 token 一旦丢掉，模型没法从剩下的文本里推断出来，比如具体的数字、错误码、文件路径等，这样的值丢了，看上下文是猜不回来的。虽然 CCR 缓存着原文，但模型得先意识到自己缺了关键信息才会去取回，而这类精确值被悄悄丢掉时往往连个缺口都看不出来，所以不能指望 CCR 兜底。kompress_compressor.py 用一条正则定义了这些必留 token：_KOMPRESS_MUST_KEEP_RE = re.compile( r"\b0x[0-9A-Fa-f]+\b" # 十六进制地址：0x7fff2038 r"|(?<![\w.])\d+(?:\.\d+)?(?![\w.])" # 独立数字：42、3.14 r"|[A-Z_]{2,}" # 全大写：SIGILL、EOF、ERROR r"|[a-z_][a-z0-9_]*\.[a-z0-9_]+" # 带点路径：libsystem_kernel.dylib r"|/[a-z0-9/._-]{2,}" # unix 路径：/usr/lib/python3.so r"|--?[a-z][\w-]*" # 命令行 flag：--verbose、-n # ... )不管模型给这些 token 打多低的分，_add_kompress_must_keep_words 都会强行把它们保留下来。这是一道防止模型误伤关键信息的硬性保险。压前 vs 压后Kompress 的压缩是词级别的丢弃。比如这样一句啰嗦的报错描述：The application process crashed with signal SIGILL at address 0x7fff2038 in libsystem_kernel.dylib # 应用进程因信号 SIGILL 在地址 0x7fff2038 处崩溃，位置在 libsystem_kernel.dylib模型判断后可能压成：crashed signal SIGILL address 0x7fff2038 libsystem_kernel.dylib # 崩溃 信号 SIGILL 地址 0x7fff2038 libsystem_kernel.dylibThe、with、at、in 这些没有信息量的虚词被丢掉了，而 SIGILL（全大写命中必留规则）、0x7fff2038（十六进制地址）、libsystem_kernel.dylib（带点路径）这些排查问题真正要用的 token 一个都没少。虽然读起来不再是通顺的句子，但模型需要的语义信息完整保留。压缩显著时，Kompress 同样会追加一条 CCR 提示，注明原文可以取回。其余几个压缩器SmartCrusher、CodeAwareCompressor、Kompress 是分工最重的三个压缩器，但 ContentRouter 的映射表里还挂着另外几个针对特定内容形态的压缩器，同样在 headroom/transforms/ 下。它们都是各自领域的一套专门规则，简单认识一下：LogCompressor（log_compressor.py）：对付运行日志和构建输出。日志的特点是大量重复模板和刷屏的 INFO，它按行解析、识别日志级别和格式，把重复的堆栈、刷屏的常规行压掉，留下报错和关键状态。DiffCompressor（diff_compressor.py）：对付 git diff 的输出。diff 里真正要紧的是改了哪些文件、增删了哪些关键行，它把无关的大段上下文折叠，保留变更的骨架。SearchCompressor（search_compressor.py）：对付 grep、ripgrep 这类命令行搜索的纯文本结果，把命中按文件聚合、去掉冗余，而不是按 JSON 数组处理。TextCrusher（text_crusher.py）：大段纯文本的快速确定性压缩。它是 Kompress 之外的另一条路，不走神经网络，而是用 BM25 相关性给句子打分、再去掉近似重复的片段，抽取式地保留原句（不改写），毫秒级就能跑完，适合请求路径上不能等 Kompress 那种大模型推理的场合。TabularIngest（tabular_ingest.py）：CSV、TSV、markdown 表格这类文本本身没有对应的压缩器，直接走 Kompress 会破坏它的行列结构，所以它先把表格文本解析成 JSON 记录数组，再交给现成的 SmartCrusher 处理。HTMLExtractor（html_extractor.py）：处理网页抓取回来的 HTML。严格说它做的是「抽取」而非「压缩」，把正文从导航、页脚、脚本这些结构性噪音里剥离出来，丢掉的是不相关的整块，而不是逐个 token。小结今天我们详细地学习了 Headroom 的三大压缩器：SmartCrusher：面向 JSON 数组的统计式压缩器，去重加头尾抽样，输出严格保持原 schema，计算已整个搬进 Rust，对高度重复的结构化数据能压到 90% 以上。CodeAwareCompressor：基于 tree-sitter 把代码解析成 AST，用一张数据表适配九种语言，保留 import、函数签名和类型注解，按语句压缩函数体，并用语法校验、过度压缩保护、异常兜底三道安全阀保证绝不输出坏代码。Kompress：作者训练的双头 ModernBERT 模型，逐 token 判断保留与丢弃，用必留正则守住数字、错误码、路径等关键信息，首次使用时后台下载、本地推理。其余压缩器：LogCompressor 压日志、DiffCompressor 压 git diff、SearchCompressor 压命令行搜索结果、TextCrusher 用 BM25 快速压纯文本、TabularIngest 把表格文本桥接给 SmartCrusher、HTMLExtractor 抽取网页正文，各自守着一种特定的内容形态。所有的压缩器共用同一套路由和 CCR 机制：由 ContentRouter 按内容类型分发到对应的一个，丢掉的原文一样进 CCR 缓存、可以取回，保证压缩不是有去无回。关于 CCR 的完整机制，也就是原文怎么缓存、模型怎么用 headroom_retrieve 把它取回来，以及 Headroom 怎么在 Claude、Codex、Gemini 这些不同 agent 之间共享一份压缩过的记忆，我们下一篇继续。参考Headroom GitHub 仓库SmartCrusher 文档代码压缩文档文本与日志压缩文档Headroom 架构文档Kompress-v2-base 模型卡ModernBERT GitHub 仓库ModernBERT 论文：Smarter, Better, Faster, Longer论文 LongCodeZip：面向代码语言模型的长上下文压缩tree-sitter 代码解析库
在前两篇里，我们先认识了 Headroom 是什么：一个替 AI agent 压缩上下文的中间层，在工具输出、日志、代码、对话历史进入大模型之前先把它们压小，token 能减少 60% 到 95%。然后又花了一整篇学习它怎么用，重点是 wrap 和 proxy 两种接入方式：headroom wrap claude 把它套在编程 agent 前面，headroom proxy 起一个本地代理，再用 headroom doctor、headroom dashboard 看健康状况和实时节省。用起来之后，自然会想知道它内部是怎么组织的。Headroom 表面上给了四种完全不同的用法（库、代理、agent 包裹、MCP server），但它们实际上用的都是同一套压缩逻辑。它还同时用了 Python 和 Rust 两种语言，这两边又是怎么拼在一起的。今天我们就从全局视角把这些理清楚，为后面几篇的源码拆解打个底。数据流总览先看最外层。不管你用哪种方式接入，Headroom 做的事情本质上只有一件：在请求发给大模型之前拦一道，把里面的内容压缩，再原样转发出去；响应回来时再看要不要处理。整条链路如下图所示：这里有两点值得注意。第一，Headroom 站在 agent 和大模型中间，对两边都尽量透明，agent 以为自己直接在跟模型说话，模型收到的则是已经压过的内容。第二，压缩是可逆的。图里最后那段 需要时取回原文 的交互，对应的是 Headroom 的 CCR（Compress-Cache-Retrieve，可逆压缩）机制：原文缓存在本地，模型真需要细节时可以通过一个专门的工具把它取回来。CCR 的细节我们留到后面单开一篇讲，这里只需知道压缩掉的东西并没有真的丢。四个入口，一条管线Headroom 对外暴露了四种用法，它们的调用姿势差别很大：用法接入方式典型命令 / 代码库在代码里直接调函数from headroom import compress代理起一个本地 HTTP 代理headroom proxy --port 8787agent 包裹把编程 agent 套进代理headroom wrap claudeMCP server作为工具挂给模型headroom mcp install看起来是四条路，但顺着源码往下走，它们最终都汇到同一个地方。库这条路最直白。compress() 定义在 headroom/compress.py 里，它内部通过一个懒加载的单例函数 _get_pipeline() 拿到压缩管线：def _get_pipeline() -> Any: """Get or create the singleton compression pipeline.""" global _pipeline if _pipeline is not None: return _pipeline with _pipeline_lock: # ... from headroom.transforms import TransformPipeline # 默认管线：CacheAligner → ContentRouter _pipeline = TransformPipeline() return _pipeline可以看到，compress() 拿到的是一个 TransformPipeline 实例。代理这条路呢？headroom proxy 启动的服务定义在 headroom/proxy/server.py，它的核心类 HeadroomProxy 在初始化时同样构造了 TransformPipeline，而且按不同厂商各建了一个：# headroom/proxy/server.py self.anthropic_pipeline = TransformPipeline(...) self.openai_pipeline = TransformPipeline(...)headroom wrap 又是什么？我们可以看下 headroom/cli/wrap.py 开头的说明，它做的是先起一个代理，再把编程 agent 的流量指过去。也就是说 wrap 是 proxy 的上层封装，它并没有另写一套压缩逻辑，压缩仍然发生在代理里的 TransformPipeline。MCP 这条路也一样。Headroom 的 MCP server 定义在 headroom/ccr/mcp_server.py，它对外暴露一个 headroom_compress 工具，模型主动调用这个工具时，底层走的还是同一套压缩入口。所以四种用法在架构上是一个漏斗：这个设计的好处很明显：新增一种接入方式，或者修一个压缩上的 bug，都只需要动 TransformPipeline 这一处即可。生命周期契约与扩展点上面说的 TransformPipeline 是实际执行压缩的编排器。但 Headroom 还在它之上定义了一层更抽象的生命周期契约，专门用来描述一次请求从进来到出去要经过哪些阶段。这层契约放在 headroom/pipeline.py 里，核心是一个枚举和一个固定顺序的元组：class PipelineStage(str, Enum): """Stable lifecycle stages for the canonical Headroom pipeline.""" SETUP = "setup" PRE_START = "pre_start" POST_START = "post_start" INPUT_RECEIVED = "input_received" INPUT_CACHED = "input_cached" INPUT_ROUTED = "input_routed" INPUT_COMPRESSED = "input_compressed" INPUT_REMEMBERED = "input_remembered" PRE_SEND = "pre_send" POST_SEND = "post_send" RESPONSE_RECEIVED = "response_received" CANONICAL_PIPELINE_STAGES: tuple[PipelineStage, ...] = ( PipelineStage.SETUP, # ... 顺序与上面的枚举一致 PipelineStage.RESPONSE_RECEIVED, )一共十一个阶段，从初始化一直排到模型响应回来。它们的顺序被写成一个不可变的元组 CANONICAL_PIPELINE_STAGES，注释里管这叫 canonical（规范的、权威的） 顺序，意思是全项目以这份顺序为准。每个阶段做的事情如下：阶段英文名这一步发生了什么初始化SETUP加载配置、准备好各个压缩器和扩展启动前PRE_START服务启动前的钩子，扩展可以在此介入启动后POST_START服务已就绪，做启动后的一次性动作收到输入INPUT_RECEIVED拿到一次请求的消息和工具定义输入已缓存INPUT_CACHEDCacheAligner 稳定前缀，让厂商的缓存能命中输入已路由INPUT_ROUTEDContentRouter 判断每段内容是什么类型输入已压缩INPUT_COMPRESSED对应的压缩器把内容压小输入已记忆INPUT_REMEMBERED需要跨会话记忆时，把内容写进记忆层发送前PRE_SEND消息定稿，转发给大模型之前的最后一道发送后POST_SEND请求已发出，等待响应收到响应RESPONSE_RECEIVED模型返回，做统计、必要时触发取回把它画成一条时间线更直观：那么这套阶段是给谁用的呢？答案在同一个文件里的 PipelineEvent 和 PipelineExtension 上。每到一个阶段，Headroom 会发出一个 PipelineEvent 事件，里面带着当前的消息、工具、请求头等信息。第三方可以实现 PipelineExtension 这个协议接口，注册进来后就能在任意阶段插手，改写消息或者读取统计：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 point（入口点，一种让第三方包在安装后被自动发现的机制）来注册的，discover_pipeline_extensions() 会在启动时把它们都找出来。也就是说，这十一个阶段不只是内部注释，它是 Headroom 对外开放的一个稳定扩展点。要区分两个「pipeline」：headroom/pipeline.py 定义的是生命周期契约（有哪些阶段、扩展怎么挂），而 headroom/transforms/pipeline.py 里的 TransformPipeline 是真正跑压缩的编排器。前者描述流程骨架，后者是骨架里 INPUT_CACHED → INPUT_ROUTED → INPUT_COMPRESSED 这几步的具体实现。三层结构从代码语言和职责上看，Headroom 大致分成三层。第一层是 Python 编排层。上面看到的 TransformPipeline、_get_pipeline()、生命周期契约都在这一层。它负责的是流程调度：按什么顺序跑、每步计多少 token、出错了怎么办。headroom/transforms/pipeline.py 里的 _build_default_transforms() 就是在拼装这条流水线：def _build_default_transforms(self) -> list[Transform]: """Build default transform pipeline from config.""" transforms: list[Transform] = [] # 1. Cache Aligner（前缀稳定） if self.config.cache_aligner.enabled: transforms.append(CacheAligner(self.config.cache_aligner)) # 2. 内容感知压缩，ContentRouter 按类型分发 transforms.append(ContentRouter()) return transforms顺序是固定的：先 CacheAligner 再 ContentRouter。CacheAligner 的作用是稳定提示词的前缀，让 Anthropic、OpenAI 这些厂商的 KV cache（复用已算过的前缀、省下重复计算）能真正命中；ContentRouter 则是分发中枢，判断每一段内容是 JSON、代码、日志还是普通文本，交给对应的压缩器。这一层还塞了不少工程上的容错逻辑。比如 TransformPipeline.apply() 里有一个熔断器（circuit breaker）：连续失败若干次后，接下来一段时间直接放行不压缩，避免每个请求都去重跑一遍必然失败的压缩。# 熔断器打开时直接透传，不再尝试压缩 if self._breaker_is_open(): passthrough_tokens = tokenizer.count_messages(messages) return TransformResult( messages=messages, tokens_before=passthrough_tokens, tokens_after=passthrough_tokens, transforms_applied=["pipeline:circuit_open"], )第二层是内容感知的压缩器。这是真正干压缩活的地方，都放在 headroom/transforms/ 目录下。ContentRouter 会根据内容类型路由到不同的压缩器：SmartCrusher：统计式的 JSON 和数组压缩器，主要对付工具返回的大段结构化数据，实现在 smart_crusher.pyCodeAwareCompressor：基于 AST（抽象语法树） 的代码压缩器，保留 import、函数签名和类型，实现在 code_compressor.pyKompressCompressor：调用作者自己训练的 Kompress 文本压缩模型，实现在 kompress_compressor.py此外还有日志、diff、搜索结果、HTML 等各自的压缩器（log_compressor.py、diff_compressor.py、search_compressor.py、html_extractor.py）这些压缩器具体怎么工作，我们后面详细学习，今天先认个脸熟。第三层是 Rust 热路径。上面那些压缩器里最吃计算的部分，其实并不是纯 Python 跑的。SmartCrusher、CodeAwareCompressor 里的解析和统计，底层调的是 Rust 写的核心。压缩发生在代理的关键路径上，每一次请求都要过一遍，性能敏感，所以作者把热点逻辑挪到了 Rust。这也就引出了下一个问题：Python 和 Rust 是怎么连起来的。Rust 与 Python 怎么连Headroom 的 Rust 代码集中在仓库的 crates/ 目录，一共四个 crate（Rust 里对一个包的称呼），职责各不相同。这部分的说明在项目根目录的 RUST_DEV.md 里写得比较全，它列出的工作区布局是这样：crates/ headroom-core/ # 库：共享类型 + 各压缩器的 Rust 实现 headroom-proxy/ # 二进制：基于 axum 的透明反向代理 headroom-py/ # PyO3 cdylib，向 Python 暴露 headroom._core headroom-parity/ # 库 + CLI：跑 Rust 与 Python 的一致性对拍逐个看它们干什么：headroom-core 是地基。压缩器、分词器、CCR 存储、相关性打分、ONNX 推理这些底层能力都在它的 src/ 下（transforms/、tokenizer/、ccr/、signals/、onnx_cpu.rs 等）。它不关心 Python，也不关心网络，就是一堆纯 Rust 库。headroom-py 是桥。它是一个 PyO3 cdylib。PyO3 是把 Rust 写成 Python 扩展模块的绑定库，cdylib 则是编译产物类型（C 动态库）。这个 crate 把 headroom-core 里的能力包一层，暴露成 Python 能 import 的模块，名字叫 headroom._core。headroom-proxy 是一个独立的代理二进制，基于 axum 和 tokio（Rust 的 Web 框架与异步运行时）实现透明反向代理。按 RUST_DEV.md 的说法，它的定位是逐步接管原来 Python 代理承担的转发工作，运维时让 Rust 代理站在公网端口、Python 代理退到私有端口，对终端用户无感。headroom-parity 是对拍工具。Rust 端口是从 Python 一点点迁过来的，怎么保证两边算出来的结果一模一样？靠它。它把 Python 的输出录成 JSON 固定样本（fixture），再拿 Rust 实现去比对，一旦出现偏差就报出来。关键在 headroom-py 这座桥的连法。看它 src/lib.rs 顶部的注释：//! PyO3 bindings for headroom-core. Exposed to Python as `headroom._core`. //! //! Why in-process: ContentRouter compresses on the proxy's hot path. Any //! IPC / subprocess / RPC bridge would dominate the cost we're trying to //! save. PyO3 calls cost ~microseconds; staying in-process is ~free.这里作者把设计意图说得很清楚：Python 调 Rust 是进程内直接调用，不是通过 IPC（进程间通信）、子进程或者 RPC。原因是压缩发生在代理的热路径上，如果每次压缩都要跨进程通信，那点通信开销反而会盖过压缩省下来的成本。PyO3 的进程内调用是微秒级的，几乎免费。lib.rs 末尾用 PyO3 的 #[pymodule] 宏把一批 Rust 类型和函数注册进 headroom._core 模块，Python 那边 from headroom._core import ... 拿到的就是它们：m.add_function(wrap_pyfunction!(hello, m)?)?; m.add_class::<PyDiffCompressor>()?; m.add_class::<PySmartCrusher>()?; m.add_class::<PySearchCompressor>()?; m.add_function(wrap_pyfunction!(detect_content_type, m)?)?; # ... 还有一批压缩器类型和检测函数那这个 Rust 扩展是怎么装进用户机器的？答案是 maturin。maturin 是专门打包 Rust 编写的 Python 扩展的构建工具，它把 Rust 代码编译好，连同 Python 代码一起打进 wheel（Python 的预编译安装包格式）。所以你 pip install headroom-ai 时，装下来的 wheel 里已经带了编译好的 Rust 二进制，不需要本地装 Rust 工具链。RUST_DEV.md 里 make build-wheel 那条命令做的就是这件事。代理启动时还会主动检查这座桥通不通。headroom/proxy/server.py 里有个 _check_rust_core()，它会尝试 from headroom._core import hello 并调用一次，确认 Rust 扩展确实加载成功，否则默认直接以错误退出。顶层目录结构把上面几层对到仓库目录上，整体是这样一棵树：headroom/ ├── headroom/ # Python 主包 │ ├── __init__.py # 对外导出面 │ ├── compress.py # compress() 一函数入口 + _get_pipeline │ ├── pipeline.py # 生命周期契约：11 个阶段 + 扩展协议 │ ├── config.py # 配置定义 │ ├── transforms/ # 压缩器与编排 │ │ ├── pipeline.py # TransformPipeline 编排器 │ │ ├── content_router.py # ContentRouter 内容路由 │ │ ├── cache_aligner.py # CacheAligner 前缀稳定 │ │ ├── smart_crusher.py # SmartCrusher，JSON 压缩 │ │ ├── code_compressor.py # CodeAwareCompressor，AST 感知 │ │ └── kompress_compressor.py # KompressCompressor，文本模型 │ ├── proxy/ # 本地代理服务 │ │ ├── server.py # HeadroomProxy、create_app │ │ ├── handlers/ # 各厂商请求处理器 │ │ └── output_shaper.py # 输出侧 token 优化 │ ├── ccr/ # 可逆压缩 + MCP server │ ├── memory/ # 跨 agent 记忆 │ ├── learn/ # 失败会话离线学习 │ ├── providers/ # 各厂商与各 agent 的适配 │ └── cli/ # 命令行子命令 wrap/proxy/doctor 等 ├── crates/ # Rust 工作区 │ ├── headroom-core/ # 核心库：压缩器实现、分词、CCR │ ├── headroom-py/ # PyO3 绑定，暴露为 headroom._core │ ├── headroom-proxy/ # axum 透明反向代理二进制 │ └── headroom-parity/ # Rust 与 Python 一致性对拍 ├── sdk/typescript/ # TypeScript SDK，只有库没有 CLI ├── docs/ # 官方文档源 ├── benchmarks/ # 压缩基准 ├── RUST_DEV.md # Rust 部分开发指南 └── pyproject.toml # Python 包定义从这棵树能看出几件事。Python 主包 headroom/ 是重心，功能几乎都在这里，transforms/ 管压缩、proxy/ 管代理、ccr/ 和 memory/ 管可逆与记忆、cli/ 管命令行。crates/ 是 Rust 那一侧，通过 headroom-py 这座桥挂进 Python。sdk/typescript/ 是给 TypeScript 用户的库，注意它只是 SDK，没有 CLI，命令行能力是 Python 包独有的。小结今天我们从全局看了一遍 Headroom 的架构：数据流：Headroom 站在 agent 和大模型中间，压缩进入模型的内容，压缩可逆，需要时能把原文取回四个入口一条管线：库、代理、包裹、MCP 四种用法最终都汇到同一个 TransformPipeline，wrap 是 proxy 的上层封装，改一处四处受益生命周期契约：headroom/pipeline.py 定义了从 SETUP 到 RESPONSE_RECEIVED 的十一个规范阶段，配合 PipelineExtension 对外开放扩展点三层结构：Python 编排层调度流程，内容感知压缩器干压缩的活，Rust 热路径承担吃计算的部分Rust 与 Python 的连法：maturin 把 Rust 编译进 wheel，PyO3 把 headroom-core 暴露成 headroom._core，Python 进程内直接调用而非 IPC，四个 crate 分别负责核心库、绑定桥、代理二进制和一致性对拍有了这张全局地图，接下来就可以往里钻了。第四篇我们进入压缩管线的源码，顺着一次 compress() 调用往下追：_get_pipeline() 怎么建管线、create_pipeline() 怎么装配、ContentRouter 又是凭什么判断一段内容该交给哪个压缩器。我们下一篇继续。参考Headroom GitHub 仓库Headroom 架构文档Headroom 压缩原理文档CCR 可逆压缩文档Headroom 代理文档Headroom MCP 文档PyO3 官方文档maturin 官方仓库
在上一篇中，我们认识了 Headroom 这个项目。它是一个面向 AI agent 的上下文压缩层：在工具输出、日志、RAG 片段、文件这些内容进入大模型之前先压缩一遍，官方称 token 能减少 60% 到 95%，而答案基本不变。我们也提到了它的四种用法：当库直接调用、当中间代理（proxy）使用、把编程 agent 整个包裹（wrap）起来、以及作为 MCP server 接入。其中 wrap 和 proxy 这两条路还有不少细节没展开，这一篇我们就拿它俩练手。动手之前先分清两者的区别：它们本质上都是让流量过一遍本地代理进程，差别只在谁把客户端指向代理。wrap 面向 Claude Code 这类现成的编程 agent，一条命令就把环境和配置替你设好，不想用了再用 unwrap 一键还原；proxy 只管起代理，把客户端指过来这一步交给你，适合任意 OpenAI 或 Anthropic 兼容的客户端。下面分别来看。headroom wrap claude 实战如果你日常就是用 Claude Code 写代码，wrap 是门槛最低的一条路。它的逻辑在 headroom/cli/wrap.py 里，wrap 是一个 Click 命令组，claude、codex、copilot、cursor、aider、opencode、cline、continue、goose 等一长串编程工具各是它下面的子命令。Click 是 Python 里最常用的命令行框架之一，用装饰器的方式定义命令、子命令、选项和参数，帮你处理参数解析、类型校验、帮助信息生成这些琐事。headroom 这个 CLI 就是用它搭起来的。直接运行：$ headroom wrap claude启动过程输出如下：这一条命令背后其实做了一连串事。顺着 wrap.py 里 claude() 函数读，主要步骤是这样的：检查 claude 在 PATH 里；启动本地代理子进程，等它就绪；配置上下文工具 rtk；注册 MCP 取回工具；设置代码图压缩器；把代理地址写进环境变量和 Claude Code 的配置文件；启动 Claude Code；会话退出时还原 base URL、停掉代理。我们逐一看几个关键动作。启动本地代理claude() 会调用 _ensure_proxy 起一个后台代理。实际上就是把自己作为一个子进程再拉起来：python -m headroom.cli proxy --port 8787，并把 --learn、--memory、--backend 等参数透传过去。起完之后它会轮询端口，等代理真正能接受连接了才继续：# headroom/cli/wrap.py（精简） for _i in range(timeout_seconds): time.sleep(1) if _check_proxy(port): click.echo(f" Logs: {log_path}") return proc if proc.poll() is not None: # ... 进程提前退出，读日志尾部报错 raise RuntimeError(...)代理启动时会把已经缓存好的机器学习组件加载进内存，然后 uvicorn（Python 的 ASGI 服务器）才绑定端口。模型都是延迟加载的：没下载过的模型启动时不会去拉，而是推迟到第一次真正用到时再下载，免得卡住启动。即便只加载已缓存的组件，初始化本身也要点时间，源码里默认给了 45 秒超时，检测到装了 torch 这类重型 ML 依赖时放宽到 90 秒，不够可以用环境变量 HEADROOM_WRAP_PROXY_TIMEOUT 调大。Headroom 在压缩链路上会用到几个模型，分工和加载时机各不相同：Magika（google/magika）：Google 开源的 AI 文件类型检测工具，用一个很小的深度学习模型读内容开头、中间、结尾的若干字节，就能判断这段内容到底是 JSON、源代码、日志还是纯文本，官方称在 200 多种类型上的准确率约 99%，Gmail、Google Drive 都用它做文件类型识别。在 Headroom 里，它负责给后面的 ContentRouter 做类型判断，好把每段路由到对的压缩器。它的模型（约 3 MB）随 pip 包自带，不走 HuggingFace，所以启动时就加载好了，不需要联网下载。Kompress（chopratejas/kompress-v2-base）：作者自己训练的文本压缩模型，专门处理纯文本，给每个 token 打重要性分、丢掉低价值的部分。它以 ONNX 格式发布，跑在本地 Rust 核心里推理。这个模型托管在 HuggingFace，首次真正压缩文本时才下载一次，之后走本地缓存；想完全离线可以预下载好再设 HF_HUB_OFFLINE=1。图片与相关性模型：这些是按功能触发的懒加载模型，纯做文本压缩时一般用不上。压缩图片时会用到 technique-router（判断该用哪种图片压缩手法）和 SigLIP 图像编码器（Google 图文编码模型的架构，把图片编码成向量）；开启相关性过滤、跨会话记忆时会用到 BGE、MiniLM 这类轻量文本向量模型，用来给内容算语义相似度。配置上下文工具 rtk代理起来后，claude() 会调用 _setup_rtk 配置一个上下文工具。这里的上下文工具，指的是在 shell 命令输出进入模型之前先把它精简、替你省 token 的一类命令行工具；Headroom 默认用的是 rtk（Rust Token Killer）。它其实是一个独立的开源项目，用 Rust 写成，Apache 2.0 许可，不需要 API Key、也没有遥测；Headroom 只是把它的二进制下载到 ~/.headroom/bin 并接进 Claude Code。rtk 针对的是另一类 token 浪费：shell 命令的输出。agent 干活时会大量执行 git diff、ls、pytest、cargo build 这些命令，它们的原始输出又长又杂，满屏的样板、空行、无关的通过项。rtk 的做法是把命令包一层，你不直接跑 git status，而是跑 rtk git status，rtk 拿到原始输出后按几种策略精简再交给模型：删掉不影响判断的样板和空行、把同类项归组（比如按目录聚合文件列表、按规则聚合 lint 告警）、测试和构建只留失败项和报错。rtk 官方在 2900 多条真实命令上测下来，平均能砍掉约 89% 的输出噪声，具体因命令而异：rtk git status 省 ~81% rtk cargo test 省 ~92%（只显示失败） rtk find 省 ~78% rtk grep 省 ~50%如果某个命令 rtk 没有对应的过滤规则，它就原样放行，所以加不加都安全。Headroom 把它接进来的方式，是调用 rtk init --global 命令，往 Claude Code 的全局配置 ~/.claude/settings.json 里注册一个 PreToolUse 钩子，如下所示：{ "hooks": { "PreToolUse": [ { "matcher": "Bash", "hooks": [ { "type": "command", "command": "rtk hook claude" } ] } ] } }钩子指 Claude Code 在特定时机自动调用的外部脚本，PreToolUse 表示在工具调用之前触发，其中 matcher 为 Bash 表示只在执行 Bash 工具前触发。命令 rtk hook claude 是 rtk 内置的钩子处理器：Claude Code 每次要执行 Bash 前，会把这次调用的信息（含即将执行的命令）以 JSON 从标准输入传给它，它据此把命令改写成走 rtk 过滤的版本再交回去。rtk 给不同 agent 各准备了一个这样的处理器（rtk hook claude / cursor / gemini / copilot），因为各家钩子的 JSON 格式不一样。这一步是幂等的，重复 wrap 不会重复注册。rtk 的二进制装在 ~/.headroom/bin/rtk，而这个目录默认不在 PATH 里。所以如果你在终端直接敲 rtk、或者模型照 RTK.md 的说明去跑 rtk gain 这类命令，可能会撞上 command not found: rtk。把 ~/.headroom/bin 加进 PATH 即可解决：export PATH="$HOME/.headroom/bin:$PATH"。除此之外，它还会在全局目录放一份 ~/.claude/RTK.md 命令速查表，并在全局的 ~/.claude/CLAUDE.md 里用 @RTK.md 把它引入，让每个会话的模型都知道 rtk 有哪些命令可用。除了 rtk，Headroom 还支持另一个上下文工具 LeanCTX（同样是本地 Rust 二进制，除了压缩 shell 输出，还带缓存文件读取、跨会话记忆等能力），把环境变量 HEADROOM_CONTEXT_TOOL 设成 lean-ctx 就能换过去。如果不想让 wrap 碰你的上下文工具配置，可以使用 --no-context-tool（旧名 --no-rtk）跳过这一步。注册 MCP 取回工具接着 _setup_headroom_mcp 会给 Claude Code 注册一个名为 headroom 的 MCP 服务器。注册就是往 Claude Code 的 ~/.claude.json 的 mcpServers 里写一条这样的配置："headroom": { "type": "stdio", "command": "/opt/homebrew/bin/headroom", "args": ["mcp", "serve"], "env": {} }type 为 stdio 表示这个 MCP 服务器通过标准输入输出跟 Claude Code 通信（本地 MCP 最常见的传输方式）；command 加 args 合起来就是启动命令 headroom mcp serve，这里 command 被解析成了 headroom 的绝对路径。env 在这里是空的，因为代理跑在默认端口 8787；只有当代理端口不是默认值时，Headroom 才会往 env 里补一个 HEADROOM_PROXY_URL，好告诉这个 MCP 服务器去连哪个代理。为什么要有这一步，得回到上一篇讲的 CCR（Compress-Cache-Retrieve，可逆压缩）。代理压缩工具输出时并不会把原文一删了之，而是把原文缓存在本地（就是 ~/.headroom/ccr_store.db 这个 SQLite 库，默认存活 30 分钟），同时在被压掉的位置留下一个形如 [Retrieve more: hash=…] 的标记。如果模型后面发现自己需要被折叠掉的细节，就靠这个 hash 去把原文取回来。而取回这个动作，需要客户端侧真有一个 headroom_retrieve 工具可调，这正是这个 MCP 服务器提供的。在 Claude Code 里敲一下 /mcp，就能看到这个 headroom 服务器实际挂了三个工具，headroom_retrieve 只是其中之一：headroom_retrieve：按 hash 取回被压掉的原文，也就是上面说的取回工具，hash 来自压缩标记 [… hash=abc123] 或 headroom_compress 的返回；headroom_compress：反过来，让模型主动把一段内容（大段工具输出、文件内容、搜索结果等）先压一遍再拿去推理，返回压缩后的文本加一个可供日后取回的 hash；headroom_stats：查看本会话的压缩统计——压了多少次、省了多少 token、估算省了多少钱，以及最近的压缩事件。可以点击每个工具查看详情：这一步同样也是幂等的，重复 wrap 不会重复注册。加 --no-mcp 可以跳过，但那样压缩标记就变得不可逆。还要注意：如果 Claude Code 在 wrap 之前已经在运行，得重启一次才能加载到这几个新注册的工具。设置代码图压缩器最后 wrap 还会配一个代码图压缩器，解决的是代码任务里的一类浪费：模型为了搞清楚一个函数在哪定义、被谁调用，往往会把整个文件甚至几个文件读进上下文，而它真正需要的只是符号定义、调用链、引用位置这些结构信息。代码图压缩器把项目预先索引成一张「代码图」，让模型按需查询这些结构，而不必整文件塞进来。默认用的是 tokensave：一个用 Rust 写的开源代码智能 MCP 服务器，把代码库预先索引成一张本地的语义知识图谱（存在 SQLite 里），再以一批 MCP 工具的形式提供符号查询、调用链遍历、影响分析等能力，官方称能省掉约 60% 到 80% 的结构探查开销，全程本地不外传。wrap 会把它的二进制拉下来，首次 tokensave init 建索引、之后 tokensave sync 增量更新（索引是惰性的，第一次真正查询时才补齐，不阻塞 wrap）。如果 tokensave 不可用，则回退到 Serena：oraios 开源的编程 agent 工具包，靠 LSP（Language Server Protocol，语言服务器协议）来理解代码，操作的对象是函数、类这样的符号而不是行号，支持 20 多种语言、同样纯本地运行，通过 uvx 启动。要注意这两个工具是二选一的：tokensave 是主用，只有它不可用时才轮到 Serena，正常情况下你只会启用其中一个。它们注册成的是独立的 MCP 服务器，名字就叫 tokensave 和 serena，和前面的 headroom 平级，都写在 ~/.claude.json 的 mcpServers 里："tokensave": { "type": "stdio", "command": "~/.local/bin/tokensave", "args": [ "serve" ], "env": {} }有几个开关可以调整这些行为：--no-tokensave 彻底不用 tokensave；--serena 反过来强制启用 Serena，哪怕 tokensave 是好的；--no-serena 则连兜底也省掉，完全禁用代码图压缩的功能。写入代理地址并启动 Claude Code最后一步才是真正把 Claude Code 指向代理。它把 ANTHROPIC_BASE_URL 指向本地代理，既写进即将启动的子进程环境，也写进项目本地的 .claude/settings.local.json：{ "env": { "ANTHROPIC_BASE_URL": "http://127.0.0.1:8787" } }之所以要写进 settings.local.json，是因为 Claude Code 的守护进程 fork 出的会话 worker 是重新读配置文件、而不是继承父进程环境的，只塞环境变量不够。值得注意的是，这里还牵扯到 Claude Code 的工具延迟加载（tool deferral），它是 Claude Code 用来控制 MCP token 开销的机制。一个工具的定义（名称、描述、参数 schema）本身就要占上下文，接的 MCP server 一多，光是工具定义就可能吃掉几万甚至十几万 token，而且每轮对话都得原样重发一遍；延迟加载的做法是默认不把所有工具的完整定义塞进请求，只给模型一个内置的「工具搜索」工具，等模型真需要某个工具时，再按需把对应的定义取进上下文，官方称这一招能省下约 85% 的工具定义 token。控制它的开关是 ENABLE_TOOL_SEARCH，可以取 true（始终延迟） 或 auto（定义能塞进上下文窗口 10% 以内就照常加载、超了才延迟）。这个行为平时默认开启，可一旦 ANTHROPIC_BASE_URL 指向自定义地址（也就是代理）就会被 Claude Code 关掉，转而把每个工具的 schema 都提前塞进上下文，白白多占几万 token，所以 wrap 会顺手把 ENABLE_TOOL_SEARCH 设成 true，让延迟加载保持开启。设置完成后，claude() 就用 subprocess.run 把真正的 claude 命令拉起来，你在命令行上传给它的额外参数原样透传，比如 headroom wrap claude --resume <id> 或 headroom wrap claude -- --model opus。此后你就照常用 Claude Code 即可，压缩在幕后进行。另外，这里还有一点需要特别注意，如果你的 Claude Code 平时是走第三方 Anthropic 兼容端点的，比如把 ANTHROPIC_BASE_URL 指向 MiniMax、智谱、Kimi 这类服务，再用 ANTHROPIC_AUTH_TOKEN 鉴权，那么一 wrap 很可能直接报 401 Invalid bearer token。原因是 wrap 只把 ANTHROPIC_BASE_URL 改成了本地代理，但代理并不知道你原来的上游是谁，它不会自动帮你从 ANTHROPIC_BASE_URL 推断。它转发 Anthropic 流量的默认目标是官方的 https://api.anthropic.com，需要使用另一个环境变量 ANTHROPIC_TARGET_API_URL 显式告诉它：$ export ANTHROPIC_TARGET_API_URL=https://api.minimaxi.com/anthropic $ headroom wrap claude设好之后，代理就会把请求连同你的 ANTHROPIC_AUTH_TOKEN 一起转发到真正的上游。几个常用开关顺带记一下：--memory：开启跨会话的持久记忆，启动前会先同步 Headroom 的记忆库和 Claude 的记忆文件--learn：开启实时流量学习，把错误到修复的模式写进 agent 的记忆文件（隐含开启 --memory）--1m：保留 100 万 token 的上下文窗口。因为走了自定义 base URL 后 Claude Code 会丢掉 context-1m 这个 beta 请求头、退回到 20 万，这个开关通过给启动进程设 ANTHROPIC_MODEL 加 [1m] 后缀把 1M 窗口重新激活--no-proxy：不自己起代理，复用一个已经在跑的unwrap 撤销通过上面的学习, 我们知道在运行 wrap 命令时, 会往磁盘写不少东西, 这些大体可以分成「随会话消失」和「留在磁盘上」两类。随会话消失的是本地代理和 settings.local.json 配置。claude() 用了一个 finally 块，只要你从 Claude Code 里 /exit（或者进程因别的原因退出），它就会把写进 settings.local.json 的 ANTHROPIC_BASE_URL 撤掉，然后再把本地代理停掉。撤 BASE URL 时，如果这个文件本来就是 wrap 新建的，删掉之后就变成空文件了，它会把整个 settings.local.json 一并删除；如果你项目里原本就有自己的 BASE URL，则还原成原值而不是删掉。留在磁盘上、跨会话存在的是那些注册类的改动：~/.claude.json 里的几个 MCP 服务器（headroom、tokensave 或 serena）和 ~/.claude/settings.json 里的 rtk 钩子。这些 wrap 退出时不会动，好让你下次直接用，要彻底清掉需要使用 unwrap 命令：$ headroom unwrap claude它的逻辑在 wrap.py 的 unwrap_claude() 里，做的正是把上面这些注册注销掉。至于 BASE URL 和本地代理，正常退出时已经清干净了，unwrap 时会再兜底扫描一遍，保证彻底清理干净。移除 rtk 钩子这一步有个坑。unwrap 是靠钩子命令里是否含 rtk-rewrite、headroom-init-claude 这两个标记来识别的，前提是 rtk 把钩子注册成了指向 ~/.claude/hooks/rtk-rewrite.sh 脚本的形式。但某些 rtk 版本并不生成这个脚本，正如前文所述，而是把钩子写成 rtk hook claude 这样的命令，于是 unwrap 匹配不到，会提示 No rtk Claude hook found in settings.json。遇到这种情况只能手动删下了。headroom proxy 独立模式wrap 虽然很省心，但它只认识那一批预置的编程 agent。如果你用的是别的客户端，比如自己写的脚本、某个 OpenAI 兼容的 GUI、或者一个第三方 IDE 插件，那就直接起独立代理，把客户端指过来。这种方案通用性最强。$ headroom proxy --port 8787启动后会打印一段横幅：横幅里的路由表说明了这个代理认识哪些接口、分别转发到哪个上游：Routing: /v1/messages → https://api.anthropic.com /v1/chat/completions → (OpenAI 上游) /v1/responses → (OpenAI 上游) (HTTP + WebSocket) /v1internal:streamGenerateContent → (Cloud Code 上游) /v1/projects/.../publishers/... → (Vertex 上游)也就是说，它同时兼容 Anthropic 的 /v1/messages 和 OpenAI 的 /v1/chat/completions 等格式，来什么格式就按什么格式转发。接入方式就是改一个环境变量，让客户端把请求发到代理而不是官方地址：# Claude Code / Anthropic SDK $ ANTHROPIC_BASE_URL=http://127.0.0.1:8787 claude # 任意 OpenAI 兼容客户端 $ OPENAI_BASE_URL=http://127.0.0.1:8787/v1 your-appheadroom proxy 的命令定义在 headroom/cli/proxy.py，它是整个项目里参数最多的命令，光 Click 选项就几十个，这里挑几个上手阶段经常用得到的介绍下：--port / -p：监听端口，默认 8787，也可用环境变量 HEADROOM_PORT--mode：优化模式，默认 token（优先压缩，允许改写历史轮次换取最大节省）；另一个是 cache（冻结历史轮次，尽量命中服务商的前缀缓存）--no-optimize：透传模式，只转发不压缩，用来对照--memory / --learn：与 wrap 里同名开关一致，开启记忆和流量学习--budget：给这个代理设一个花费上限（美元），超了就返回 429代理默认只绑定回环地址 127.0.0.1。如果你把它绑到非回环地址对外提供服务，务必设 HEADROOM_PROXY_TOKEN 加上入站鉴权，否则接口就是裸奔的，启动横幅里也会用醒目的 WARNING 提示这一点。除了业务接口，代理还暴露了一组运维端点：GET /livez 进程存活 GET /readyz 是否可接流量 GET /health 聚合健康状态 GET /stats 详细统计 GET /metrics Prometheus 指标后面几个命令都靠它们工作。headroom doctor：健康检查$ headroom doctor它的实现在 headroom/cli/doctor.py，对 Headroom 的各个状态进行检查：代理进程是否在跑并应答 /livez运行中的代理版本和已安装包版本是否一致（版本漂移会让你跑着旧代码却不自知）Claude Code / Codex 是否配置成走了代理当前 shell 的环境变量是否指向代理节省数据是否在累积有没有设预算每一项给出 pass / warn / fail 三种状态，输出是一张带颜色的表格：从这张表能一眼看出问题出在哪。还可以加一个 --json 参数，拿到机器可读的结果，方便接进脚本。headroom perf：性能与节省报告$ headroom perf该命令读取日志文件 ~/.headroom/logs/proxy.log 做聚合分析，包括：token 节省与压缩效果、缓存命中率与前缀稳定性、各类转换（transform）和路由的分布、以及一些可操作的建议等。输出结果如下：默认看最近 7 天，可以用 --hours 24 缩窄窗口。它支持三种输出格式：$ headroom perf --hours 24 # 最近 24 小时 $ headroom perf --format json # 聚合报告输出成 JSON $ headroom perf --format csv --hours 24 > last-24h.csv--format json 和 --format csv 会吐出机器可读的数据，适合接进你自己的看板或做长期趋势图。另外还有一个更聚焦的 headroom savings 命令，专门展示随时间累积的持久压缩节省。headroom dashboard：实时面板如果你更喜欢看图，代理内置了一个 Web 面板：$ headroom dashboard Dashboard: http://127.0.0.1:8787/dashboarddashboard 命令很简单，就是把 http://127.0.0.1:<port>/dashboard 在浏览器里打开（加 --no-open 则只打印地址不开浏览器）。要注意它依赖一个正在运行的代理，面板数据来自代理的统计端点，所以得先有 headroom proxy 或 headroom wrap 在跑。打开后是这样一个实时节省面板：小结这一篇我们把 Headroom 在本机跑了起来，走通了 wrap 和 proxy 两条命令行接入方式：wrap 包裹编程 agent：headroom wrap claude 命令背后做了一堆事，起本地代理、配 rtk 上下文工具、注册 headroom 的 MCP 取回工具、设 tokensave 或 Serena 代码图压缩器，最后把 Claude Code 指向代理再启动。这些动作分两种寿命：本地代理和 settings.local.json 里的 BASE URL 随会话退出自动收回，而 MCP、rtk 钩子这些留在磁盘上的注册要用 unwrap 才清得掉；proxy 独立模式：对于 wrap 不支持的客户端，可以起一个独立代理，通过 ANTHROPIC_BASE_URL 或 OPENAI_BASE_URL 把请求指过来即可，通用性最强；同时它还会暴露一组运维端点，供命令行工具使用，比如 doctor 对代理做健康检查，perf 读日志生成聚合报告，dashboard 看实时节省面板；到这里，Headroom 的 wrap 和 proxy 都只是我们眼中的黑盒：请求进去、压缩后的请求出来。但它到底是靠什么识别内容类型、又是怎么在不丢信息的前提下把 token 砍掉一大半的？下一篇我们就掀开盖子，从 ContentRouter 到三大压缩器，再到管线的完整生命周期，把 Headroom 的整体架构理一遍。参考Headroom GitHub 仓库Headroom 官方文档 QuickstartHeadroom Proxy 文档Headroom MCP 文档Headroom CCR 可逆压缩文档Headroom PyPI 页面rtk（Rust Token Killer）仓库lean-ctx（LeanCTX）仓库tokensave 仓库Serena（oraios/serena）仓库Click 官方文档Claude 官方文档：Tool search toolAnthropic 工程博客：Introducing advanced tool useKompress-v2-base 模型卡Magika（google/magika）仓库technique-router 模型仓库SigLIP 图像编码器模型仓库BGE 向量模型（FlagEmbedding）仓库all-MiniLM-L6-v2 模型卡
用过 Claude Code、Codex 这类 coding agent 的人，大概都有过盯着它烧 token 的经历。你让它排查一个线上问题，它先 grep 一遍代码，再 cat 几个文件，跑一轮测试看输出，翻几屏日志，最后去 GitHub 上查相关 issue。这一连串动作里，每一步的输出都会被原封不动塞回对话，成为下一步推理的上下文，而这些内容大多又长又重复，真正有用的可能就那么几行。一个任务还没跑完，token 就已经哗哗地流出去了。这些 token 你既要按输入计费为它们付钱，它们还会挤占本就有限的上下文窗口。Headroom 就是冲着这个场景来的。它是 Netflix 工程师 Tejas Chopra 在 2026 年 1 月开源的一个工具，做的事情说穿了很简单：在一段内容进入大模型之前，先替你压缩一遍，把那些又长又重复的部分挤掉，官方称 token 能减少 60% 到 95%，而答案基本不变。这个系列我们就来好好研究下 Headroom。Headroom 介绍Headroom 是一个面向 AI agent 的上下文压缩层。它的定位很直接：在一段内容进入大模型之前，先把它压缩一遍，在不改变语义的前提下把喂给模型的 token 数砍下来。那为什么要专门做这么一层压缩？关键在于喂给 agent 的内容里有大量冗余。agent 干活靠的是不停调用各种工具，比如搜代码、读文件、执行命令、查日志，每次调用的返回都会被塞回对话里，作为下一步推理的上下文。而这些工具输出往往又长又啰嗦：一次代码搜索命中上百处，每处还带出成片的源代码，可真正有用的往往只是命中的函数签名和它所在的位置，大段函数体的实现细节模型多半用不上；一份线上事故的日志动辄几万行，真正有用的可能就那么几段异常堆栈；RAG 拉回来的文档片段，彼此之间常有大量冗余；一个大 JSON 数组里，几百个元素结构完全一样，模型其实看几个样本就够了。这些内容如果一股脑全部塞进上下文，带来两个后果。一是贵：输入 token 越多，每次调用越贵，而 agent 一个任务里可能要调用几十上百次模型。二是挤：上下文窗口是有上限的，垃圾内容占多了，真正重要的信息反而被挤出去或者被稀释，模型的表现还会下降。Headroom 的思路是在这些内容进模型之前先做一道压缩。它给出的官方压缩基准是这样的：场景压缩前 token压缩后 token节省代码搜索（100 条结果）17,7651,40892%SRE 事故排查65,6945,11892%GitHub issue 分诊54,17414,76173%代码库探索78,50241,25447%压缩幅度跟内容类型强相关：结构高度重复的工具输出（代码搜索、日志）能压掉九成，而信息密度本来就高的代码库探索只能压掉不到一半。这也符合直觉，冗余越多的地方越好压。光压得狠没用，关键是压完之后模型答得对不对。作者跑了几组标准评测集，给出的准确率对比是：评测集压缩前压缩后备注GSM8K（数学推理）0.8700.870持平TruthfulQA（事实性）0.5300.560+0.030SQuAD v2（阅读理解）97%97%压缩 19%BFCL（函数调用）97%97%压缩 32%从这组数字看，压缩在保住准确率的前提下把 token 砍了下来，个别评测集甚至略有提升。把无关的冗余去掉、让相关信息更突出，本来就有可能帮到模型。核心特性除了压缩比这个硬指标，Headroom 还有几个设计上的特点。内容感知路由：Headroom 不是拿一套算法压所有东西，而是先判断内容类型，再挑对应的压缩器。这个分发的活由 ContentRouter（内容路由器）来干，它能识别 JSON、源代码、纯文本、日志、diff 等类型，把每一段路由到最合适的压缩器上，混合内容还会被拆开、分段路由再拼回去。CCR 可逆压缩：压缩最让人担心的是丢信息，万一模型后面正好需要被压掉的那段原文怎么办？Headroom 的答案是 CCR（Compress-Cache-Retrieve，压缩-缓存-取回）。它在压缩的同时把原文缓存在本地，并给模型提供一个 headroom_retrieve 工具，模型需要更细的内容时可以主动调它，在缓存的存活时间（TTL）内把原文取回来。压缩因此是可逆的，而不是一刀切地删掉。跨 agent 记忆：现在很多人同时用 Claude、Codex、Gemini 几个 agent 干活，但它们之间的上下文是割裂的。Headroom 提供了一层跨 agent 的共享记忆（SharedContext），能在多个 agent 之间传递压缩后的上下文，还带来源标注和自动去重。本地优先：压缩、缓存、记忆都在本机完成，数据不出机器。它内置的文本压缩模型 chopratejas/kompress-v2-base 也是跑在本地 Rust 核心里的，遥测默认关闭，需要手动设 HEADROOM_TELEMETRY=on 才会开启。对于不能把代码和日志外发的团队，这一点挺关键。四种用法Headroom 以一个 Python 包的形式发布，PyPI 包名叫 headroom-ai，许可证是 Apache 2.0，要求 Python 3.10 及以上版本。虽然是个 Python 包，但它的压缩核心是用 Rust 写的，通过 maturin 打成预编译的 wheel，安装时不需要你本地有 Rust 工具链。maturin 是把 Rust 代码打包成标准 wheel（Python 的二进制安装包格式）、直接发布到 PyPI 的构建工具，通常和 Rust 的 Python 绑定库 PyO3 搭配使用。之前写 turbovec 快速入门时我们就见过这套 Rust + PyO3 + maturin 的工具链，pydantic-core、polars 这些明星项目用的也是它。一条 pip 命令就能装上。headroom 命令行工具随基础包一起提供，而 [all] 会把代理、MCP、压缩模型等可选依赖一并装上，下面几种用法都要用到它们：$ pip install "headroom-ai[all]"装好之后，Headroom 在接入方式上给了四条路径，你可以按自己的使用形态挑一条。第一种是当库用。直接在代码里调 compress() 函数，把消息列表压缩一遍再发给模型。接口就是一个 compress() 函数：from headroom import compress # messages 是标准的 Anthropic 或 OpenAI 消息格式 result = compress(messages, model="claude-sonnet-4-5-20250929") result.messages # 压缩后的消息，格式不变，token 更少 result.tokens_saved # 省下的 token 数 result.compression_ratio # 比如 0.65 表示省了 65%返回的 CompressResult 里除了压缩后的消息，还带上了压缩前后的 token 数和实际用到的压缩器列表，方便你核对效果。这个函数对 Anthropic、OpenAI、LiteLLM 乃至任意 HTTP 客户端都适用，因为它只负责压缩消息内容，至于请求怎么发、发给谁，完全由你自己决定。compress() 内部还有个保护逻辑：如果压缩后 token 反而比压缩前还多（比如短消息本来就没啥可压的），它会自动回退到原始消息，不会帮倒忙。第二种是当代理用。用 headroom proxy 命令启动一个本地代理，把 LLM 请求的地址指向它，它压缩完再转发给真正的 API，全程零代码改动。第三种是包裹 agent。用 headroom wrap claude 这样的命令，把一个现成的 coding agent 包起来，让它的所有请求自动走压缩。除了 Claude，它还支持 codex、copilot、cursor、aider、opencode、cline、continue、goose、openhands、openclaw、vibe 等一长串工具。第四种是当 MCP server 用。MCP 作为大模型接入外部工具和数据源的标准协议，现在被越来越多的 agent 支持。Headroom 可以作为一个 MCP server 挂上去，把压缩、取回等能力以工具的形式暴露给模型。60 秒上手预览四种用法里最省事的就是包裹 agent，我们就拿它开个头，先感受下压缩真正跑起来是什么样，更完整的实战留到下一篇。我们在上一节已经用 pip install 安装好了 headroom-ai，直接包裹你在用的 coding agent 即可。比如包裹 Claude：$ headroom wrap claude这一条命令会在本地起一个代理，把 Claude Code 的请求都指过去，再照常把 claude 拉起来。之后你正常写代码，所有请求都会先经 Headroom 压缩一遍再发出去，全程零代码改动。想还原就用 headroom unwrap claude。如果你用的不是这批预置的 agent，也可以起一个独立代理，让任意兼容的客户端把请求指过来：$ headroom proxy --port 8787然后把你的 LLM 客户端的 base URL 指到 http://localhost:8787 即可，零代码改动。代理跑起来后，还能开一个实时面板看省了多少：$ headroom dashboard面板效果大致如下：小结这篇我们从整体上认识了 Headroom：定位：一个面向 AI agent 的上下文压缩层，在内容进入大模型之前先压缩一遍，官方称 token 可减少 60% 到 95% 而答案基本不变；要解决的问题：coding agent 反复读工具输出、日志、RAG 片段和文件，token 成本高、上下文窗口被冗余挤占；四种用法：库（compress()）、代理（headroom proxy）、包裹 agent（headroom wrap）、MCP server，按自己的使用形态任选；核心特性：ContentRouter 按内容类型分发到不同压缩器，CCR 把原文缓存在本地让压缩可逆，此外还有跨 agent 记忆，以及压缩、缓存、记忆全在本机完成、数据不出机器的本地优先设计。上一节只是浅尝辄止地跑了两条命令，很多细节都还没有展开。下一篇我们就顺着 wrap 和 proxy 这两条路深入下去，把每一步讲清楚，看看其效果究竟如何。参考Headroom GitHub 仓库Headroom 官方文档站Headroom 快速上手文档Headroom 架构文档Headroom 压缩基准文档Headroom CCR 可逆压缩文档Kompress-v2-base 模型卡PyPI 上的 headroom-ai 包maturin GitHub 仓库