Headroom SDK 实战指南：用 HeadroomClient 透明压缩 LLM 上下文并度量 Token 节省

Headroom SDK 实战指南：用 HeadroomClient 透明压缩 LLM 上下文并度量 Token 节省

【免费下载链接】headroomCompress tool outputs, logs, files, and RAG chunks before they reach the LLM. 20% fewer tokens for coding agents, 60-95% fewer tokens for JSON, same answers. Library, proxy, MCP server.项目地址: https://gitcode.com/GitHub_Trending/head/headroom

本文围绕 Headroom 官方 SDK 指南展开：Headroom SDK 通过HeadroomClient包装你现有的 OpenAI / Anthropic / Google 客户端，在请求发出前对消息（尤其是大型工具输出）执行确定性压缩与缓存优化，同时保持调用方式与原客户端完全一致。读完本文，你将掌握 SDK 的安装接入、三种运行模式（optimize / audit / simulate）的差异、按请求级别的覆盖参数、会话与历史指标的查询方式、错误处理体系，以及底层_create调用链中「解析 → 变换 → 缓存优化 → 统计」的实际工作原理，并了解 Python SDK 与 TypeScript SDK、独立代理（Proxy）三种接入形态的选型边界。

一、安装与定位：SDK 在 Headroom 中的角色

Headroom 的核心卖点是「在内容到达 LLM 之前压缩它」——压缩工具输出、日志、文件与 RAG 分块，从而降低每次请求的输入 Token。它提供三种接入形态：

1. SDK（本文主题）：在你的应用进程内包装 LLM 客户端，控制粒度最细，指标在进程内可查；
2. Proxy：把工具的 API 地址指向 Headroom 代理，适合无法改代码的现成工具（Claude Code、Cursor 等）；
3. MCP Server：以工具形式挂到支持 MCP 的 Agent 上。

官方指南给出的 SDK 选型定位（wiki/sdk.md）：

也就是说：需要细粒度控制用 SDK，管理现成工具用 Proxy。

安装命令（指南原文）：

pip install headroom-ai openai

二、快速上手：HeadroomClient 如何包装现有客户端

HeadroomClient的构造函数签名为（源码见 headroom/client.py）：

HeadroomClient( original_client, # 底层 LLM 客户端（OpenAI 风格） provider, # Provider 实例，提供模型名/上下文限制/token 计数 store_url=None, # 指标存储 URL（sqlite:// 或 jsonl://），默认临时目录 default_mode="audit", # 默认模式（"audit" | "optimize"） model_context_limits=None, cache_optimizer=None, enable_cache_optimizer=True, enable_semantic_cache=False, config=None, # 完整的 HeadroomConfig，优先级高于上面的独立参数 )

指南中的快速上手示例：

from headroom import HeadroomClient, OpenAIProvider from openai import OpenAI # 创建包装客户端 client = HeadroomClient( original_client=OpenAI(), provider=OpenAIProvider(), default_mode="optimize", ) # 用法与原客户端完全一致 response = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "user", "content": "Hello!"}, ], ) print(response.choices[0].message.content)

从源码结构看，__init__内部依次完成四件事（headroom/client.py）：

1. 初始化存储：create_storage(store_url)。若未指定store_url，默认落到sqlite:///<tempdir>/headroom.db，开箱即用不污染项目目录；
2. 构建变换管线：TransformPipeline(self._config, provider=self._provider)，这是真正执行压缩的引擎；
3. 初始化缓存优化器：enable_cache_optimizer=True且未显式传入cache_optimizer时，会通过CacheOptimizerRegistry按 provider 名称自动探测（OpenAI / Anthropic / Google 各有注册实现，见 headroom/cache 模块 的导出列表）；enable_semantic_cache=True时再用SemanticCacheLayer包一层查询级语义缓存；
4. 暴露两套 API 门面：self.chat.completions（OpenAI 风格）和self.messages（Anthropic 风格），二者最终都汇入同一个内部_create方法，只以api_style="openai" / "anthropic"区分。

三、一次请求的完整处理链（_create 源码剖析）

理解 SDK 行为的关键是HeadroomClient._create（headroom/client.py），其内部流程为：

请求进入 │ ├─ 1. parse_messages(messages, tokenizer) # 把消息切成 Block（system/user/assistant/tool_call/tool_result/rag…） ├─ 2. tokenizer.count_messages(messages) # 统计压缩前 tokens_before ├─ 3. CacheAligner.get_alignment_score(...) # 计算前缀缓存对齐分数 ├─ 4. compute_prefix_hash(messages) # 稳定前缀哈希 │ ├─ 5. 若 mode == OPTIMIZE： │ TransformPipeline.apply(messages, model, model_limit, output_buffer, tool_profiles) │ → 得到 optimized_messages / tokens_after / transforms_applied │ ├─ 6. 缓存优化： │ 若启用语义缓存 → SemanticCacheLayer.process(...)，命中则直接返回缓存响应 │ 否则 → provider 专属 cache optimizer 处理断点/前缀对齐 │ ├─ 7. 通过 provider registry 把优化后的请求转发给 original_client ├─ 8. 写入 RequestMetrics 到存储（SQLite），并更新内存会话统计 └─ 9. 返回原始格式的响应对象（对调用方透明）

几个要点：

• Block 解析：parse_messages将消息切分为带类型的原子块（Block.kind包括system/user/assistant/tool_call/tool_result/rag/unknown，定义见 headroom/config.py），后续变换针对块级内容而非整条消息，这保证了 tool_calls、角色顺序等结构不被破坏；
• 输出缓冲：output_buffer（配置项output_buffer_tokens，默认 4000，见 headroom/config.py）为模型输出预留空间，压缩预算 = 模型上下文上限 − 输出缓冲；
• 上下文上限解析顺序：用户model_context_limits覆盖（支持版本化名称的前缀匹配）→ Provider 提供值，见_get_context_limit（headroom/client.py）与HeadroomConfig.get_context_limit（headroom/config.py）。

四、真正的价值点：工具输出压缩

指南强调「实际节省发生在工具输出上」。这是一个 500 条搜索结果被压缩的完整示例（指南原文）：

import json # 包含大型工具输出的对话 messages = [ {"role": "user", "content": "Search for Python tutorials"}, { "role": "assistant", "content": None, "tool_calls": [ { "id": "call_123", "type": "function", "function": {"name": "search", "arguments": '{"q": "python"}'}, } ], }, { "role": "tool", "tool_call_id": "call_123", "content": json.dumps( {"results": [{"title": f"Tutorial {i}", "score": 100 - i} for i in range(500)]} ), }, {"role": "user", "content": "What are the top 3?"}, ] # Headroom 将 500 条结果压缩到约 15 条，保留得分最高的条目 response = client.chat.completions.create(model="gpt-4o-mini", messages=messages) # 查看节省 stats = client.get_stats() print(f"Tokens saved: {stats['session']['tokens_saved_total']}") # 典型输出: "Tokens saved: 3500"

背后的执行者是SmartCrusher变换器：它识别 JSON 列表/键值对结构，按相关性评分（BM25 始终可用，embeddings 需要可选依赖 sentence-transformers，见 headroom/init.py 的导出注释）保留高价值条目、丢弃冗余长尾，并保留错误项与异常值。管线完成时会输出日志如SmartCrusher: kept 15 of 1000 items（见第七节日志）。

需要说明的是，指南中「500 条 → 约 15 条」「3500 tokens」为示例性典型值，实际保留数量取决于评分阈值与内容分布；仓库中的 SmartCrusher 单测（tests/test_smart_crusher.py）覆盖了其保留/丢弃策略的回归行为。

五、受支持的 Provider

OpenAI

from headroom import HeadroomClient, OpenAIProvider from openai import OpenAI client = HeadroomClient( original_client=OpenAI(), provider=OpenAIProvider(), )

Anthropic

Anthropic 走client.messages门面（对应_create的api_style="anthropic"分支，headroom/client.py）：

from headroom import HeadroomClient, AnthropicProvider from anthropic import Anthropic client = HeadroomClient( original_client=Anthropic(), provider=AnthropicProvider(), ) response = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=1024, messages=[{"role": "user", "content": "Hello!"}], )

Google

from headroom import HeadroomClient from headroom.providers import GoogleProvider import google.generativeai as genai client = HeadroomClient( original_client=genai, provider=GoogleProvider(), )

三个 Provider 均为懒加载导出（headroom/providers/init.py 中按名称解析到headroom.providers.openai / anthropic / google子模块），避免import headroom时提前加载各家 SDK。Provider 除提供上下文限制外，还提供对应模型的 token 计数能力，这直接决定tokens_before / tokens_after统计的精度。

六、三种运行模式：Optimize / Audit / Simulate

模式由HeadroomMode枚举定义（headroom/config.py）：AUDIT（只观察不修改）、OPTIMIZE（应用确定性变换）、SIMULATE（只返回变换计划）。注意构造函数默认值为default_mode="audit"，即默认是安全审计模式，需要显式传default_mode="optimize"才实际压缩（headroom/client.py）。

Optimize（指南推荐的默认写法）

应用全部安全变换：

client = HeadroomClient( original_client=OpenAI(), provider=OpenAIProvider(), default_mode="optimize", )

Audit

只观察和记录、不修改请求。适合上线前量化「如果开启压缩能省多少」：

client = HeadroomClient( original_client=OpenAI(), provider=OpenAIProvider(), default_mode="audit", )

在_create源码中，audit 模式会跳过第 5 步的pipeline.apply，但依然完成解析、token 统计、前缀哈希与指标落库——所以即使不改一个字节，你也能持续拿到浪费信号（waste signals）与对齐分数。

Simulate

不发 API 请求、直接返回压缩计划，适合在 CI 或成本预估脚本中离线评估：

plan = client.chat.completions.simulate( model="gpt-4o", messages=large_conversation, ) print(f"Would save {plan.tokens_saved} tokens") print(f"Transforms: {plan.transforms}")

simulate的返回对象是SimulationResult，源码_simulate（headroom/client.py）中可见其完整字段：tokens_before / tokens_after / tokens_saved / transforms / estimated_savings（按 provider 价格估算的单请求美元节省）、messages_optimized（可直接打印查看压缩后内容）、block_breakdown、waste_signals、cache_alignment_score。它是真实变换管线的pipeline.simulate分支，与 optimize 走同一套变换逻辑，只是不落库、不转发。

七、按请求覆盖参数（headroom_* 系列）

create/messages.create/simulate均接受一组headroom_前缀的关键字参数（headroom/client.py）：

指南示例：

response = client.chat.completions.create( model="gpt-4o", messages=[...], # 覆盖本次请求的模式 headroom_mode="audit", # 为输出预留更多 token headroom_output_buffer_tokens=8000, # 保留最后 5 轮 headroom_keep_turns=5, )

其余**kwargs原样透传给底层客户端，因此temperature、tools等官方参数不受影响。

八、验证与观测：validate_setup、get_stats、日志

验证安装配置

result = client.validate_setup() if not result["valid"]: print("Setup issues:", result["issues"])

validate_setup的检查项在源码中有明确清单（headroom/client.py）：provider 能否计数 token、存储是否可读、default_mode是否为合法枚举、缓存优化器（如启用）是否加载成功。返回结构为{"valid", "provider", "storage", "config", "cache_optimizer"}四组ok/error字段。

会话统计（无数据库查询）

stats = client.get_stats() print(stats) # { # "session": {"requests_total": 10, "tokens_saved_total": 5000, ...}, # "config": {"mode": "optimize", "provider": "openai", ...}, # "transforms": {"smart_crusher_enabled": True, ...} # }

从实现看（headroom/client.py），get_stats读的是纯内存字典_session_stats，零 I/O：requests_total / requests_optimized / requests_audit / tokens_saved_total / cache_hits五项计数器在每次_create结束时由_update_session_stats累加，其中tokens_saved_total只累计 optimize 模式下的max(0, before - after)。配置段则暴露当前模式、provider 名、缓存优化器名与语义缓存开关；transforms 段暴露smart_crusher_enabled与cache_aligner_enabled。

开启日志

import logging logging.basicConfig(level=logging.INFO) # 现在你将看到： # INFO:headroom.transforms.pipeline:Pipeline complete: 45000 -> 4500 tokens # INFO:headroom.transforms.smart_crusher:SmartCrusher: kept 15 of 1000 items

日志前缀对应源码模块：headroom.transforms.pipeline报告每次管线执行的压缩前后 token 数，headroom.transforms.smart_crusher报告保留条数，是排查「压缩是否真的生效」的最直接手段。

九、流式与错误处理

流式

流式对 SDK 是透明的——压缩发生在请求侧（发出之前），响应侧原样透传：

stream = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello!"}], stream=True, ) for chunk in stream: if chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="")

异常体系

Headroom 定义了统一的异常树，全部继承自HeadroomError，并携带机器可读的details字典（headroom/exceptions.py）：

指南的推荐捕获顺序（子类在前、基类兜底）：

from headroom import ( HeadroomClient, HeadroomError, ConfigurationError, ProviderError, ) try: response = client.chat.completions.create(...) except ConfigurationError as e: print(f"Config issue: {e}") except ProviderError as e: print(f"Provider issue: {e}") except HeadroomError as e: print(f"Headroom error: {e}")

由于__str__会把details拼进消息（如Invalid mode 'foo' (valid_modes=...)），直接打印异常即可拿到诊断细节，无需额外解析。

十、历史指标与高级配置

历史指标查询

与get_stats的内存统计不同，get_metrics走的是持久化存储（默认 SQLite），支持时间/模型/模式过滤（headroom/client.py）：

from datetime import datetime, timedelta metrics = client.get_metrics( start_time=datetime.utcnow() - timedelta(hours=1), limit=100, ) for m in metrics: print(f"{m.timestamp}: {m.tokens_input_before} -> {m.tokens_input_after}")

返回的RequestMetrics数据模型定义在 headroom/config.py，除 token 前后值外还包含变换列表、缓存命中情况等字段。另有get_summary(start_time, end_time)直接返回聚合统计。HeadroomClient同时支持上下文管理器（with HeadroomClient(...) as client:退出时自动close()存储连接）。

高级配置

指南指向完整配置文档 wiki/configuration.md，并给出典型组合：

client = HeadroomClient( original_client=OpenAI(), provider=OpenAIProvider(), default_mode="optimize", enable_cache_optimizer=True, enable_semantic_cache=False, model_context_limits={ "gpt-4o": 128000, "gpt-4o-mini": 128000, }, )

从源码补充几个容易踩的点：

• model_context_limits只做用户覆盖：DEFAULT_MODEL_CONTEXT_LIMITS是空字典（headroom/config.py），未覆盖的模型走 Provider 提供的上下文限制；
• enable_semantic_cache默认关闭：语义缓存叠加在 provider 缓存优化器之上，带相似度阈值/TTL/条目上限（由CacheOptimizerConfig控制），开启后命中会直接返回缓存响应而不打上游；
• store_url决定指标落盘位置：sqlite://或jsonl://，默认在系统临时目录，多实例共享建议显式指定，避免get_metrics读到的是临时库。

十一、延伸：TypeScript SDK 与生态

除 Python SDK 外，仓库在 sdk/typescript 提供了 TypeScript 实现，与 Python 版能力对齐：src/client.ts是核心包装客户端，src/adapters/下提供 OpenAI、Anthropic、Gemini、Vercel AI 四家适配器，另有simulate.ts（干跑模拟）、shared-context.ts（多 Agent 共享上下文）与 hooks 机制；sdk/typescript/examples 目录含工具调用 Agent、流式聊天、结构化输出、多 provider 等可直接参考的示例，测试覆盖见 sdk/typescript/test。Python 侧的演示脚本可参考 examples/context_compression_demo.py 与 examples/README.md。

十二、小结：什么时候用 SDK

SDK 的最小接入面只有三行：pip install→HeadroomClient(original_client=..., provider=..., default_mode="optimize")→ 原样调用。压缩在请求侧完成、响应侧透明透传，配合get_stats()/get_metrics()/ INFO 日志，你可以在不改业务代码的前提下持续观测每次请求省了多少 token、应用了哪些变换，从而把「上下文压缩」变成可验证、可回退（切回 audit）的工程能力。

【免费下载链接】headroomCompress tool outputs, logs, files, and RAG chunks before they reach the LLM. 20% fewer tokens for coding agents, 60-95% fewer tokens for JSON, same answers. Library, proxy, MCP server.项目地址: https://gitcode.com/GitHub_Trending/head/headroom