Fork me on GitHub

分类 Headroom 下的文章

学习Headroom 的输出 token 优化

在前面几篇里,我们把 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-input-output-cost.jpg

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 计费的。

output-config-effort.png

努力档位路由要做的,就是识别出这种机械续写的轮次,把它的思考档位调低;而对新问题、对报错,保持全力。判断哪种轮次靠的是 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) 上的一个伪随机小数 fracfrac 落在 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 优化:

  1. 为什么输出也值得压:输出 token 的单价约为输入的 5 倍,浪费集中在三处:寒暄与收尾语、复述已有上下文、对机械步骤过度思考。代理本身不生成输出 token,只能通过改写请求去引导模型少写。
  2. 两根杠杆:详略引导往系统提示词末尾追加简洁指令,五个等级层层加码(1 级管寒暄、2 级管复述、3 级管理由和改写幅度、4 级电报体),默认 2 级;努力档位路由按消息结构识别机械续写、把它的思考档位调低,同时守住两条安全规则:只降不加、绝不动思考开关。
  3. 学习偏好headroom learn --verbosity 从打断率、快速跳过率等行为信号反推你想要的级别,启发式之外还能加 LLM 裁判,运行时还有 AIMD 控制器根据实时信号继续微调。
  4. 诚实的度量:输出节省是反事实问题,Headroom 把它分成三层:合成对照给出的估计(带置信区间、只标注为估计)、留出对照组的 A/B 测量(HEADROOM_OUTPUT_HOLDOUT),以及不需要反事实的直接浪费(复读率)。

至此,Headroom 的两个方向就凑齐了。把输入侧压缩和输出侧削减放在一起对比,正好作为整个系列的总结:

维度输入侧压缩输出侧削减
作用对象发给模型的内容模型写回来的内容
手法直接缩小文本体积改写请求去影响模型行为
节省是否可直接观测是,压前压后两个数一减否,是反事实,需估计或 A/B 测量
单价权重输入 token,单价较低输出 token,Opus 级约为输入的 5 倍
相关模块transforms/proxy/output_shaper.pyoutput_savings.py

输入侧是把已经确定的文本压小,是个纯函数,干净利落。输出侧动不了模型本身,只能通过改写请求去引导它少写,效果天生带不确定性,所以配了一整套诚实的度量。

写到这里,这个系列也差不多要告一段落了。回头看这条路线:第一篇介绍 Headroom 是什么、它要解决什么问题;第二篇把 headroom wrapheadroom proxy 跑起来;第三篇俯瞰架构,把 ContentRouter、三大压缩器、CacheAligner、CCR 和管线生命周期串成一张图;第四篇钻进 compress() 入口和管线编排的源码;第五篇细看 SmartCrusher、CodeAwareCompressor 和在本地推理的 Kompress;第六篇看 CCR 可逆压缩;第七篇看跨 agent 记忆和 headroom learn;到这最后一篇,把方向从输入侧翻到了输出侧。

从压缩内容到压缩生成,从纯函数式的确定节省到反事实的诚实估计,Headroom 这套东西的价值不只在省钱的数字,更在它对待「省了多少」这件事的态度。希望这个系列能帮你把 Headroom 从一个命令行工具,理解成一套可以借鉴的工程思路。

参考


学习 Headroom 的压缩管线

在上一篇里,我们从架构的角度把 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 两种格式。
  • modelmodel_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_messagescompress_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 这类 agent
balanced30%都不压4折中档,保护好用户和系统消息,目标省 70%
coding模型自定都不压2代码负载档,不钉死比例,靠无损压缩和相关性出节省
general模型自定都不压0通用档,代码少,没什么位置性内容要保护

这四个档位分两路:agent-90balanced 显式钉死了 target_ratio,直接规定保留多少;codinggeneral 不设这个比例,交给 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,
        ...)

整个路由决策的流程可以画成这样:

compressors.png

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 压缩管线的主干走了一遍:

  1. 入口 compress():解析 CompressConfig(默认按编程 agent 场景调,保护用户消息和最近 4 条),跑完管线拿到 CompressResult,中间有膨胀防护栏和异常兜底,绝不帮倒忙。
  2. _get_pipeline()TransformPipeline:懒加载的单例管线,默认顺序是 CacheAligner → ContentRouter。管线里还带连续失败熔断。
  3. ContentRouter 路由决策:先检测内容类型(原生 Magika 链,Windows 降级到正则),再查映射表选压缩器;代码默认改道走 Kompress;混合内容拆片段分别压。
  4. CacheAligner:一个纯检测器,用结构化解析(非正则)找出系统提示词里的 UUID、时间戳、JWT、哈希这类易变内容,发警告提示缓存前缀不稳,但从不改写提示词。

整体链路还是比较清晰的,至此,我们已经了解了「一段内容被送到哪个压缩器」这条路由主线,但每个压缩器内部到底怎么把 token 压下来,还没拆开。下一篇我们就深入三大压缩器:处理 JSON 的 SmartCrusher、AST 感知的 CodeAwareCompressor,以及跑在 Rust 核心里的 Kompress 文本压缩模型。

参考


学习 Headroom 的跨 agent 记忆与失败学习

在上一篇里,我们看了 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 调用。

with-memory-three-things.jpg

攒下的记忆可以用 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

可以看到,一条记忆加进来,流程分四步:先给内容生成向量,然后把记忆本体落库,最后同时建两个索引,向量索引管语义检索,全文索引管关键词精确检索。

memory-add-four-step-pipeline.jpg

这四步各自对应 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_idsession_idagent_idturn_id 这一串参数,这是 Headroom 记忆系统的另一个特点 —— 分层作用域(scope)。一条记忆可以挂在四层里的任何一层:user → session → agent → turn,从宽到窄。注意这里的 agent 不是指「某个 agent 应用」,而是当前会话里的一个 agent 实例,一个会话里可以有多个 agent(比如主 agent 和它派生的 subagent),每个实例的存活期都在会话内部,所以 agent 比 session 窄。四层按存活期理解:user 跨所有会话,session 只管当前这一个会话,agent 管会话里某一个实例,turn 是单次 LLM 调用。

hierarchical-memory-scopes-and-bubbling.jpg

函数末尾的 _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.pyMemoryBridge 如下:

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 = "<!-- headroom:memory:start -->"
MARKER_END = "<!-- headroom:memory: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)在这里划定了回写边界:写入器只碰 <!-- headroom:memory:start --><!-- headroom:memory:end --> 之间的内容,你自己在文件里手写的部分不会动。子类只需实现 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 即可。

整个记忆同步的数据流如下图所示:

cross-agent-memory-sync-flow.jpg

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 条建议:

headroom-learn.png

它给出的不是「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 调用理解完整对话上下文,产出结构化建议。这条流水线按字面就是四步,每步的职责是:

  1. Scanner(扫描):从磁盘上把 agent 的会话记录读出来。以 Claude Code 为例,读的是 ~/.claude/projects/<项目>/ 下的 JSONL 会话日志,把每一次工具调用(名字、入参、成功还是失败、token 数)和用户消息解析成结构化事件。每种 agent 一个插件(plugins/ 下的 claude、codex、gemini),运行输出里那行 Detected agents 就是这一步探测到的。
  2. Digest Builder(摘要):几百个会话、几千次调用不可能全塞给模型,这一步把它们压成一份 token 预算内(约 8 万 token)的文字摘要:项目概况和总数(多少会话、多少调用、失败率)、检测到的循环放在最前面(最贵的浪费模式,附实测浪费)、之前已学到的模式、以及每个会话精简后的事件流(报错截断、保留成功标记和用户消息)。它就是喂给 LLM 的那份「证据包」。
  3. LLM(分析):只发一次调用。系统提示词把它设定成「分析 coding agent 会话、提取能防止 token 浪费的模式」的专家,并给了明确的优先级,循环最高,往下是环境规则、文件结构事实、用户偏好、失败模式、工作流规则;用户消息就是那份摘要。返回结构化 JSON。
  4. Recommendations(建议):JSON 被解析成一条条 Recommendation,每条带着写到哪个文件(CLAUDE.local.md 还是 MEMORY.md)、具体内容和估计节省的 token。之后 apply_loop_weighting 用循环的实测浪费校准估算值,按节省降序排好,交给 writer.py 落盘。

learn-scanner-digest-llm-recommendations.jpg

所以判断「哪些是该学的经验」这件事本身,是交给一个 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 的直调、各处理一套鉴权和响应格式。

具体用哪个模型,由下面这个顺序决定:

  1. 显式指定 --model 优先级最高,它的取值有两类:任意 litellm 模型名(100 多家 provider 任选),或三个本机 CLI 标识(转给本机对应的 CLI 做分析,支持 claude-cli / gemini-cli / codex-cli);
  2. 环境变量里有 API key,按一张写死的映射表来取:ANTHROPIC_API_KEYclaude-sonnet-4-6OPENAI_API_KEYgpt-4oGEMINI_API_KEYgemini/gemini-flash-latest
  3. 一个 key 都没有:看 HEADROOM_LEARN_CLI 环境变量指定的 CLI;
  4. 如果还没有:自动探测本机装了的 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 -50grep foo | head -100 就归成了同一个签名。默认要重复满 3 次才算循环,这是能把「循环」和「一次性重试」区分开的最小次数。浪费的 token 是实测下界,不是让 LLM 猜的:错误循环里每次都算浪费,重取循环里第一次是正当工作、只算后面 N-1 次的重取。

检测发生在模型调用前,校准发生在模型调用后。模型返回建议之后,apply_loop_weighting 会把与某个循环签名重叠的建议的 estimated_tokens_saved 抬到至少等于该循环实测浪费的 token。因为循环的实测浪费是多次累加的,这一步能可靠地把「防循环」的建议排到「防一次性错误」的建议前面,而不必指望 LLM 自己把权重估对。

把纠正写进文件

模型给出建议、再经循环的实测浪费校准估算之后,流水线就剩最后一步:把建议写进文件,让下次会话的 agent 能读到。这一步由 writer.py 负责。它同样用标记块(<!-- headroom:learn:start --> / <!-- headroom:learn:end -->)圈出自己的地盘,只动块内的内容。至于写到哪个文件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 的两块「记忆」能力:

  1. 跨 agent 记忆:按 user -> session -> agent -> turn 四层作用域组织,普通记忆随会话消亡,重要的会自动上浮到用户级、跨会话越攒越多;检索同时走向量索引(HNSW)和全文索引(FTS5)两条路;它还能和各 agent 的 markdown 记忆文件双向同步:导入按哈希检测变更、只挑改动的段落,导出只回写自己新增的记忆,并用标记块圈定边界、绝不碰手写的内容。
  2. headroom learn:走 Scanner → Digest → LLM → Recommendations 的流水线,判断该学什么这件事交给模型而不是正则;loops.py 把错误循环和 RTK 重取循环折叠成规范签名、按实测浪费加权;writer.py 默认把纠正写进 gitignore 的 CLAUDE.local.md,也支持 AGENTS.mdGEMINI.md

到这里,关于 Headroom 模型输入这一侧的内容就基本讲完了。不过省 token 还有另一半没讲:模型输出的那部分。同样一个问题,模型可以啰嗦地复述一大段,也可以简洁作答,输出 token 一样要计费。Headroom 的 Output Shaper 就是冲着这半边来的。我们下一篇看它怎么削减输出 token,也给这个系列收个尾。

参考


学习 Headroom 的 CCR 可逆压缩

在上一篇里,我们学习了 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 调用

ccr-four-components.jpg

我们挨个看这四块,它们合起来覆盖了实时和异步两种场景下的完整取回链路。

注入 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.pyCCRResponseHandler 里。

它的入口是 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 取回流程串起来如下图所示:

ccr-workflow.png

还有一种边界情况处理得很谨慎:如果模型在同一轮里既调了 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 等。

batch-processing.png

批量 API 的玩法是:客户端把 N 个请求打包成一个数组一次性提交,拿到一个 batch ID(类似一个任务号),之后拿它轮询,直到全部跑完,再一次性拉回 N 份结果。批量请求里的内容同样会被 Headroom 压缩,因此模型同样可能会发起 headroom_retrieve 调用。但这条链路和实时调用完全不同,根本不存在「模型响应回来、代理当场拦截」的那一刻,Response Handler 是专为实时调用写的,自然用不上。

batch_processor.pyBatchResultProcessor 就是补这个场景的,分两步配合。提交批量时,先把每个请求的上下文(消息、工具列表)按 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 可逆压缩读完了,四块组件都看完了:

  1. CCR 的原理:压缩时原文不丢、按哈希缓存在本地,压缩产物带上标记,模型信息不够时凭哈希取回。传输有损,端到端无损。
  2. 三个组件串成实时取回流程tool_injection.py 往请求里注入 headroom_retrieve 工具(并粘性保留以护住 KV cache),response_handler.py 拦截并自动兑现模型的取回调用、带着原文续问一轮,context_tracker.py 跨轮追踪压缩内容、按查询相关性主动展开。取回工具有代理注入和 MCP server 两条分发渠道,混合调用时客户端走 MCP 自己兑现。
  3. 批量 API 里的取回:异步批量请求赶不上实时续问,由 batch_processor.py 在结果回来时补做"检测 → 取回 → 续问 → 替换"。

下一篇我们看输入侧的另外两块拼图:跨 agent 的共享记忆,以及从失败会话里学经验的 headroom learn

参考


学习 Headroom 的三大压缩器

在上一篇中,我们跟着 compress() 走完了整条压缩管线。当时看到 ContentRouter 会先识别一段内容到底是 JSON、代码还是普通文本,再把它交给对应的压缩器处理。这三类文本对应的压缩器分别是 SmartCrusherCodeAwareCompressorKompress,今天我们就来逐个学习下这三大压缩器。

compressors.jpg

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):把内容模板相同的行聚成簇,日志往往同一句报错刷几十遍,只有时间戳在变,每个簇只留一条代表,其余的去掉,重复的刷屏就压掉了。
  • 通用保底:带报错词的行、取值异常稀有的行,强制保留。报错词是一份写死的清单(errorexceptioncrashtimeout 等十几个),只要一条里出现其中任何一个就保留。取值稀有针对的是另一类情况:有些异常不带 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 是一个增量式的代码解析库,能把多种语言的源代码解析成统一的语法树结构,很多编辑器用它来做语法高亮和代码折叠。

tree-sitter.png

code_compressor.py 的整体流程可以画成一张图:

code-compressor.png

用数据表描述每种语言

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_statementfunction_definitionclass_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)
    pass

import 一行不动,函数签名连同类型注解 List[str] 完整保留,多行 docstring 只留了第一行摘要、后面那段详细说明被压掉了,函数体那一大段实现也只剩一行占位注释。对于一次代码检索返回的多个文件,模型光看签名和类型往往就够判断该看哪个函数了;真要深入某个函数的实现,再通过 CCR 把原文取回来即可。压缩显著时,CodeAwareCompressor 还会在末尾追加一条注释,写明省了多少 token、CCR 的 hash 是多少、多久过期。

Kompress:跑在本地的 ModernBERT 压缩模型

前两个压缩器面对的都是结构清晰的内容:JSON 有 schema,代码有语法。可现实里还有大量没有明显结构的文本,比如报错栈、RAG 检索回来的文档片段、大段的对话记录。这类内容 SmartCrusherCodeAwareCompressor 都使不上劲,这时就该 Kompress 上场了。

Kompress 是作者专门训练的一个文本压缩模型,托管在 HuggingFace 上,模型 ID 是 chopratejas/kompress-v2-base。和前两个基于规则的压缩器不同,它是一个真正的神经网络模型,逐个 token 判断该保留还是丢弃。

双头 ModernBERT

Kompress 是在一个叫 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.dylib

Thewithatin 这些没有信息量的虚词被丢掉了,而 SIGILL(全大写命中必留规则)、0x7fff2038(十六进制地址)、libsystem_kernel.dylib(带点路径)这些排查问题真正要用的 token 一个都没少。虽然读起来不再是通顺的句子,但模型需要的语义信息完整保留。压缩显著时,Kompress 同样会追加一条 CCR 提示,注明原文可以取回。

其余几个压缩器

SmartCrusherCodeAwareCompressorKompress 是分工最重的三个压缩器,但 ContentRouter 的映射表里还挂着另外几个针对特定内容形态的压缩器,同样在 headroom/transforms/ 下。它们都是各自领域的一套专门规则,简单认识一下:

  • LogCompressorlog_compressor.py):对付运行日志和构建输出。日志的特点是大量重复模板和刷屏的 INFO,它按行解析、识别日志级别和格式,把重复的堆栈、刷屏的常规行压掉,留下报错和关键状态。
  • DiffCompressordiff_compressor.py):对付 git diff 的输出。diff 里真正要紧的是改了哪些文件、增删了哪些关键行,它把无关的大段上下文折叠,保留变更的骨架。
  • SearchCompressorsearch_compressor.py):对付 grepripgrep 这类命令行搜索的纯文本结果,把命中按文件聚合、去掉冗余,而不是按 JSON 数组处理。
  • TextCrushertext_crusher.py):大段纯文本的快速确定性压缩。它是 Kompress 之外的另一条路,不走神经网络,而是用 BM25 相关性给句子打分、再去掉近似重复的片段,抽取式地保留原句(不改写),毫秒级就能跑完,适合请求路径上不能等 Kompress 那种大模型推理的场合。
  • TabularIngesttabular_ingest.py):CSV、TSV、markdown 表格这类文本本身没有对应的压缩器,直接走 Kompress 会破坏它的行列结构,所以它先把表格文本解析成 JSON 记录数组,再交给现成的 SmartCrusher 处理。
  • HTMLExtractorhtml_extractor.py):处理网页抓取回来的 HTML。严格说它做的是「抽取」而非「压缩」,把正文从导航、页脚、脚本这些结构性噪音里剥离出来,丢掉的是不相关的整块,而不是逐个 token。

小结

今天我们详细地学习了 Headroom 的三大压缩器:

  1. SmartCrusher:面向 JSON 数组的统计式压缩器,去重加头尾抽样,输出严格保持原 schema,计算已整个搬进 Rust,对高度重复的结构化数据能压到 90% 以上。
  2. CodeAwareCompressor:基于 tree-sitter 把代码解析成 AST,用一张数据表适配九种语言,保留 import、函数签名和类型注解,按语句压缩函数体,并用语法校验、过度压缩保护、异常兜底三道安全阀保证绝不输出坏代码。
  3. Kompress:作者训练的双头 ModernBERT 模型,逐 token 判断保留与丢弃,用必留正则守住数字、错误码、路径等关键信息,首次使用时后台下载、本地推理。
  4. 其余压缩器:LogCompressor 压日志、DiffCompressor 压 git diff、SearchCompressor 压命令行搜索结果、TextCrusher 用 BM25 快速压纯文本、TabularIngest 把表格文本桥接给 SmartCrusher、HTMLExtractor 抽取网页正文,各自守着一种特定的内容形态。

所有的压缩器共用同一套路由和 CCR 机制:由 ContentRouter 按内容类型分发到对应的一个,丢掉的原文一样进 CCR 缓存、可以取回,保证压缩不是有去无回。

关于 CCR 的完整机制,也就是原文怎么缓存、模型怎么用 headroom_retrieve 把它取回来,以及 Headroom 怎么在 Claude、Codex、Gemini 这些不同 agent 之间共享一份压缩过的记忆,我们下一篇继续。

参考


学习 Headroom 的整体架构

在前两篇里,我们先认识了 Headroom 是什么:一个替 AI agent 压缩上下文的中间层,在工具输出、日志、代码、对话历史进入大模型之前先把它们压小,token 能减少 60% 到 95%。然后又花了一整篇学习它怎么用,重点是 wrapproxy 两种接入方式:headroom wrap claude 把它套在编程 agent 前面,headroom proxy 起一个本地代理,再用 headroom doctorheadroom dashboard 看健康状况和实时节省。

用起来之后,自然会想知道它内部是怎么组织的。Headroom 表面上给了四种完全不同的用法(库、代理、agent 包裹、MCP server),但它们实际上用的都是同一套压缩逻辑。它还同时用了 Python 和 Rust 两种语言,这两边又是怎么拼在一起的。今天我们就从全局视角把这些理清楚,为后面几篇的源码拆解打个底。

数据流总览

先看最外层。不管你用哪种方式接入,Headroom 做的事情本质上只有一件:在请求发给大模型之前拦一道,把里面的内容压缩,再原样转发出去;响应回来时再看要不要处理。整条链路如下图所示:

headroom-workflow.png

这里有两点值得注意。第一,Headroom 站在 agent 和大模型中间,对两边都尽量透明,agent 以为自己直接在跟模型说话,模型收到的则是已经压过的内容。第二,压缩是可逆的。图里最后那段 需要时取回原文 的交互,对应的是 Headroom 的 CCR(Compress-Cache-Retrieve,可逆压缩)机制:原文缓存在本地,模型真需要细节时可以通过一个专门的工具把它取回来。CCR 的细节我们留到后面单开一篇讲,这里只需知道压缩掉的东西并没有真的丢。

四个入口,一条管线

Headroom 对外暴露了四种用法,它们的调用姿势差别很大:

用法接入方式典型命令 / 代码
在代码里直接调函数from headroom import compress
代理起一个本地 HTTP 代理headroom proxy --port 8787
agent 包裹把编程 agent 套进代理headroom wrap claude
MCP 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 工具,模型主动调用这个工具时,底层走的还是同一套压缩入口。

所以四种用法在架构上是一个漏斗:

4-ways.png

这个设计的好处很明显:新增一种接入方式,或者修一个压缩上的 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模型返回,做统计、必要时触发取回

把它画成一条时间线更直观:

pipeline-stages.png

那么这套阶段是给谁用的呢?答案在同一个文件里的 PipelineEventPipelineExtension 上。每到一个阶段,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

顺序是固定的:先 CacheAlignerContentRouterCacheAligner 的作用是稳定提示词的前缀,让 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.py
  • CodeAwareCompressor:基于 AST(抽象语法树) 的代码压缩器,保留 import、函数签名和类型,实现在 code_compressor.py
  • KompressCompressor:调用作者自己训练的 Kompress 文本压缩模型,实现在 kompress_compressor.py
  • 此外还有日志、diff、搜索结果、HTML 等各自的压缩器(log_compressor.pydiff_compressor.pysearch_compressor.pyhtml_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.mdmake 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 的架构:

  1. 数据流:Headroom 站在 agent 和大模型中间,压缩进入模型的内容,压缩可逆,需要时能把原文取回
  2. 四个入口一条管线:库、代理、包裹、MCP 四种用法最终都汇到同一个 TransformPipeline,wrap 是 proxy 的上层封装,改一处四处受益
  3. 生命周期契约headroom/pipeline.py 定义了从 SETUPRESPONSE_RECEIVED 的十一个规范阶段,配合 PipelineExtension 对外开放扩展点
  4. 三层结构:Python 编排层调度流程,内容感知压缩器干压缩的活,Rust 热路径承担吃计算的部分
  5. Rust 与 Python 的连法:maturin 把 Rust 编译进 wheel,PyO3 把 headroom-core 暴露成 headroom._core,Python 进程内直接调用而非 IPC,四个 crate 分别负责核心库、绑定桥、代理二进制和一致性对拍

有了这张全局地图,接下来就可以往里钻了。第四篇我们进入压缩管线的源码,顺着一次 compress() 调用往下追:_get_pipeline() 怎么建管线、create_pipeline() 怎么装配、ContentRouter 又是凭什么判断一段内容该交给哪个压缩器。我们下一篇继续。

参考


Headroom 上手:wrap 与 proxy 两种接入方式实战

在上一篇中,我们认识了 Headroom 这个项目。它是一个面向 AI agent 的上下文压缩层:在工具输出、日志、RAG 片段、文件这些内容进入大模型之前先压缩一遍,官方称 token 能减少 60% 到 95%,而答案基本不变。我们也提到了它的四种用法:当库直接调用、当中间代理(proxy)使用、把编程 agent 整个包裹(wrap)起来、以及作为 MCP server 接入。

headroom-quickstart.png

其中 wrapproxy 这两条路还有不少细节没展开,这一篇我们就拿它俩练手。动手之前先分清两者的区别:它们本质上都是让流量过一遍本地代理进程,差别只在谁把客户端指向代理。wrap 面向 Claude Code 这类现成的编程 agent,一条命令就把环境和配置替你设好,不想用了再用 unwrap 一键还原;proxy 只管起代理,把客户端指过来这一步交给你,适合任意 OpenAI 或 Anthropic 兼容的客户端。下面分别来看。

headroom wrap claude 实战

如果你日常就是用 Claude Code 写代码,wrap 是门槛最低的一条路。它的逻辑在 headroom/cli/wrap.py 里,wrap 是一个 Click 命令组,claudecodexcopilotcursoraideropencodeclinecontinuegoose 等一长串编程工具各是它下面的子命令。

Click 是 Python 里最常用的命令行框架之一,用装饰器的方式定义命令、子命令、选项和参数,帮你处理参数解析、类型校验、帮助信息生成这些琐事。headroom 这个 CLI 就是用它搭起来的。

直接运行:

$ headroom wrap claude

启动过程输出如下:

headroom-wrap-claude-startup.png

这一条命令背后其实做了一连串事。顺着 wrap.pyclaude() 函数读,主要步骤是这样的:

  1. 检查 claude 在 PATH 里;
  2. 启动本地代理子进程,等它就绪;
  3. 配置上下文工具 rtk;
  4. 注册 MCP 取回工具;
  5. 设置代码图压缩器;
  6. 把代理地址写进环境变量和 Claude Code 的配置文件;
  7. 启动 Claude Code;
  8. 会话退出时还原 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 在压缩链路上会用到几个模型,分工和加载时机各不相同:

  • Magikagoogle/magika):Google 开源的 AI 文件类型检测工具,用一个很小的深度学习模型读内容开头、中间、结尾的若干字节,就能判断这段内容到底是 JSON、源代码、日志还是纯文本,官方称在 200 多种类型上的准确率约 99%,Gmail、Google Drive 都用它做文件类型识别。在 Headroom 里,它负责给后面的 ContentRouter 做类型判断,好把每段路由到对的压缩器。它的模型(约 3 MB)随 pip 包自带,不走 HuggingFace,所以启动时就加载好了,不需要联网下载。
  • Kompresschopratejas/kompress-v2-base):作者自己训练的文本压缩模型,专门处理纯文本,给每个 token 打重要性分、丢掉低价值的部分。它以 ONNX 格式发布,跑在本地 Rust 核心里推理。这个模型托管在 HuggingFace,首次真正压缩文本时才下载一次,之后走本地缓存;想完全离线可以预下载好再设 HF_HUB_OFFLINE=1
  • 图片与相关性模型:这些是按功能触发的懒加载模型,纯做文本压缩时一般用不上。压缩图片时会用到 technique-router(判断该用哪种图片压缩手法)和 SigLIP 图像编码器(Google 图文编码模型的架构,把图片编码成向量);开启相关性过滤、跨会话记忆时会用到 BGEMiniLM 这类轻量文本向量模型,用来给内容算语义相似度。

配置上下文工具 rtk

代理起来后,claude() 会调用 _setup_rtk 配置一个上下文工具。这里的上下文工具,指的是在 shell 命令输出进入模型之前先把它精简、替你省 token 的一类命令行工具;Headroom 默认用的是 rtk(Rust Token Killer。它其实是一个独立的开源项目,用 Rust 写成,Apache 2.0 许可,不需要 API Key、也没有遥测;Headroom 只是把它的二进制下载到 ~/.headroom/bin 并接进 Claude Code。

rtk-home.jpg

rtk 针对的是另一类 token 浪费:shell 命令的输出。agent 干活时会大量执行 git difflspytestcargo 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 表示在工具调用之前触发,其中 matcherBash 表示只在执行 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 注册一个名为 headroomMCP 服务器。注册就是往 Claude Code 的 ~/.claude.jsonmcpServers 里写一条这样的配置:

"headroom": {
  "type": "stdio",
  "command": "/opt/homebrew/bin/headroom",
  "args": ["mcp", "serve"],
  "env": {}
}

typestdio 表示这个 MCP 服务器通过标准输入输出跟 Claude Code 通信(本地 MCP 最常见的传输方式);commandargs 合起来就是启动命令 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、估算省了多少钱,以及最近的压缩事件。

可以点击每个工具查看详情:

headroom-retrieve.png

这一步同样也是幂等的,重复 wrap 不会重复注册。加 --no-mcp 可以跳过,但那样压缩标记就变得不可逆。还要注意:如果 Claude Code 在 wrap 之前已经在运行,得重启一次才能加载到这几个新注册的工具。

设置代码图压缩器

最后 wrap 还会配一个代码图压缩器,解决的是代码任务里的一类浪费:模型为了搞清楚一个函数在哪定义、被谁调用,往往会把整个文件甚至几个文件读进上下文,而它真正需要的只是符号定义、调用链、引用位置这些结构信息。代码图压缩器把项目预先索引成一张「代码图」,让模型按需查询这些结构,而不必整文件塞进来。

默认用的是 tokensave:一个用 Rust 写的开源代码智能 MCP 服务器,把代码库预先索引成一张本地的语义知识图谱(存在 SQLite 里),再以一批 MCP 工具的形式提供符号查询、调用链遍历、影响分析等能力,官方称能省掉约 60% 到 80% 的结构探查开销,全程本地不外传。wrap 会把它的二进制拉下来,首次 tokensave init 建索引、之后 tokensave sync 增量更新(索引是惰性的,第一次真正查询时才补齐,不阻塞 wrap)。

tokensave.png

如果 tokensave 不可用,则回退到 Serena:oraios 开源的编程 agent 工具包,靠 LSP(Language Server Protocol,语言服务器协议)来理解代码,操作的对象是函数、类这样的符号而不是行号,支持 20 多种语言、同样纯本地运行,通过 uvx 启动。

serena.png

要注意这两个工具是二选一的:tokensave 是主用,只有它不可用时才轮到 Serena,正常情况下你只会启用其中一个。它们注册成的是独立的 MCP 服务器,名字就叫 tokensaveserena,和前面的 headroom 平级,都写在 ~/.claude.jsonmcpServers 里:

"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.jsonANTHROPIC_BASE_URL 撤掉,然后再把本地代理停掉。撤 BASE URL 时,如果这个文件本来就是 wrap 新建的,删掉之后就变成空文件了,它会把整个 settings.local.json 一并删除;如果你项目里原本就有自己的 BASE URL,则还原成原值而不是删掉。

留在磁盘上、跨会话存在的是那些注册类的改动:~/.claude.json 里的几个 MCP 服务器(headroomtokensaveserena)和 ~/.claude/settings.json 里的 rtk 钩子。这些 wrap 退出时不会动,好让你下次直接用,要彻底清掉需要使用 unwrap 命令:

$ headroom unwrap claude

它的逻辑在 wrap.pyunwrap_claude() 里,做的正是把上面这些注册注销掉。至于 BASE URL 和本地代理,正常退出时已经清干净了,unwrap 时会再兜底扫描一遍,保证彻底清理干净。

移除 rtk 钩子这一步有个坑。unwrap 是靠钩子命令里是否含 rtk-rewriteheadroom-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

启动后会打印一段横幅:

headroom-proxy-startup.png

横幅里的路由表说明了这个代理认识哪些接口、分别转发到哪个上游:

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-app

headroom 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 三种状态,输出是一张带颜色的表格:

headroom-doctor.png

从这张表能一眼看出问题出在哪。还可以加一个 --json 参数,拿到机器可读的结果,方便接进脚本。

headroom perf:性能与节省报告

$ headroom perf

该命令读取日志文件 ~/.headroom/logs/proxy.log 做聚合分析,包括:token 节省与压缩效果、缓存命中率与前缀稳定性、各类转换(transform)和路由的分布、以及一些可操作的建议等。输出结果如下:

headroom-perf.png

默认看最近 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/dashboard

dashboard 命令很简单,就是把 http://127.0.0.1:<port>/dashboard 在浏览器里打开(加 --no-open 则只打印地址不开浏览器)。要注意它依赖一个正在运行的代理,面板数据来自代理的统计端点,所以得先有 headroom proxyheadroom wrap 在跑。打开后是这样一个实时节省面板:

headroom-dashboard-panel.png

小结

这一篇我们把 Headroom 在本机跑了起来,走通了 wrapproxy 两条命令行接入方式:

  1. wrap 包裹编程 agentheadroom wrap claude 命令背后做了一堆事,起本地代理、配 rtk 上下文工具、注册 headroom 的 MCP 取回工具、设 tokensave 或 Serena 代码图压缩器,最后把 Claude Code 指向代理再启动。这些动作分两种寿命:本地代理和 settings.local.json 里的 BASE URL 随会话退出自动收回,而 MCP、rtk 钩子这些留在磁盘上的注册要用 unwrap 才清得掉;
  2. proxy 独立模式:对于 wrap 不支持的客户端,可以起一个独立代理,通过 ANTHROPIC_BASE_URLOPENAI_BASE_URL 把请求指过来即可,通用性最强;同时它还会暴露一组运维端点,供命令行工具使用,比如 doctor 对代理做健康检查,perf 读日志生成聚合报告,dashboard 看实时节省面板;

到这里,Headroom 的 wrap 和 proxy 都只是我们眼中的黑盒:请求进去、压缩后的请求出来。但它到底是靠什么识别内容类型、又是怎么在不丢信息的前提下把 token 砍掉一大半的?下一篇我们就掀开盖子,从 ContentRouter 到三大压缩器,再到管线的完整生命周期,把 Headroom 的整体架构理一遍。

参考


Headroom 介绍:给 AI agent 装一层可逆的上下文压缩

用过 Claude Code、Codex 这类 coding agent 的人,大概都有过盯着它烧 token 的经历。你让它排查一个线上问题,它先 grep 一遍代码,再 cat 几个文件,跑一轮测试看输出,翻几屏日志,最后去 GitHub 上查相关 issue。这一连串动作里,每一步的输出都会被原封不动塞回对话,成为下一步推理的上下文,而这些内容大多又长又重复,真正有用的可能就那么几行。一个任务还没跑完,token 就已经哗哗地流出去了。这些 token 你既要按输入计费为它们付钱,它们还会挤占本就有限的上下文窗口。

Headroom 就是冲着这个场景来的。它是 Netflix 工程师 Tejas Chopra 在 2026 年 1 月开源的一个工具,做的事情说穿了很简单:在一段内容进入大模型之前,先替你压缩一遍,把那些又长又重复的部分挤掉,官方称 token 能减少 60% 到 95%,而答案基本不变。

headroom-github.png

这个系列我们就来好好研究下 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.030
SQuAD 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-dashboard.png

小结

这篇我们从整体上认识了 Headroom:

  1. 定位:一个面向 AI agent 的上下文压缩层,在内容进入大模型之前先压缩一遍,官方称 token 可减少 60% 到 95% 而答案基本不变;
  2. 要解决的问题:coding agent 反复读工具输出、日志、RAG 片段和文件,token 成本高、上下文窗口被冗余挤占;
  3. 四种用法:库(compress())、代理(headroom proxy)、包裹 agent(headroom wrap)、MCP server,按自己的使用形态任选;
  4. 核心特性:ContentRouter 按内容类型分发到不同压缩器,CCR 把原文缓存在本地让压缩可逆,此外还有跨 agent 记忆,以及压缩、缓存、记忆全在本机完成、数据不出机器的本地优先设计。

上一节只是浅尝辄止地跑了两条命令,很多细节都还没有展开。下一篇我们就顺着 wrapproxy 这两条路深入下去,把每一步讲清楚,看看其效果究竟如何。

参考