Headroom 集成指南：compress() 函数、LiteLLM 回调、ASGI 中间件与代理的八条接入路径详解

Headroom 是一个 LLM 上下文压缩库，可以在工具输出、日志、文件和 RAG 分块到达模型之前将其压缩，让相同的回答消耗更少的 token。你并不一定要运行 Headroom 的代理服务器——它可以作为库嵌入任何 LLM 客户端、代理或框架。本文基于 wiki/integration-guide.md 并结合仓库源码，完整讲解八条接入路径（compress() 函数、LiteLLM 回调、ASGI 中间件、独立代理、Agno、LangChain、TypeScript SDK、OpenClaw 插件）以及压缩钩子（Hooks）的定制机制，并结合 headroom/compress.py、headroom/integrations/asgi.py、headroom/hooks.py 等源码解析每条路径的底层调用链与安全护栏。读完本文，你可以针对任意技术栈选择最省事的接入方式，并理解压缩发生在请求链路的哪个位置、返回了哪些可观测指标。

一、路径选择总览

不同技术栈对应的最简接入方式如下（完整继承自原文档的选择矩阵）：

你的场景	使用方式	接入成本
任意 Python 应用	compress() 函数	2 行代码
LiteLLM	LiteLLM 回调	1 行代码
Python 代理（FastAPI、自研）	ASGI 中间件	1 行代码
Claude Code / Cursor / Copilot CLI	Headroom 代理	1 条命令或环境变量
Agno 智能体	Agno 集成	包装模型
LangChain	LangChain 集成	包装模型
非 Python 应用	Headroom 代理	HTTP
TypeScript SDK	compress()	npm install headroom-ai
Vercel AI SDK	headroomMiddleware()	中间件适配器
OpenAI Node SDK / Anthropic TS SDK	withHeadroom()	客户端包装器

选择逻辑很清晰：能改代码就用库（compress/回调/中间件），改不了代码就用代理（改 base URL 即可）。下面逐条展开。

二、compress() 函数：最简单的集成方式

这是原文档强调的"最简集成"：不依赖代理、不依赖任何配置，拿到消息、压缩、再发送。

from headroom import compress

# Before sending to your LLM:
result = compress(messages, model="claude-sonnet-4-5-20250929")
response = your_client.create(messages=result.messages)  # Fewer tokens, same answer

print(f"Saved {result.tokens_saved} tokens ({result.compression_ratio:.0%})")

2.1 配合 Anthropic SDK

from anthropic import Anthropic
from headroom import compress

client = Anthropic()
messages = [
    {"role": "user", "content": "What went wrong?"},
    {"role": "assistant", "content": "Let me check.", "tool_use": [...]},
    {"role": "user", "content": [{"type": "tool_result", "content": huge_json}]},
]

compressed = compress(messages, model="claude-sonnet-4-5-20250929")
response = client.messages.create(
    model="claude-sonnet-4-5-20250929",
    messages=compressed.messages,
    max_tokens=1000,
)

2.2 配合 OpenAI SDK

from openai import OpenAI
from headroom import compress

client = OpenAI()
messages = [
    {"role": "user", "content": "Analyze these results"},
    {"role": "tool", "content": big_json_output, "tool_call_id": "call_1"},
]

compressed = compress(messages, model="gpt-4o")
response = client.chat.completions.create(
    model="gpt-4o",
    messages=compressed.messages,
)

2.3 配合 LiteLLM（直接调用）

import litellm
from headroom import compress

messages = [...]
compressed = compress(messages, model="bedrock/claude-sonnet")
response = litellm.completion(model="bedrock/claude-sonnet", messages=compressed.messages)

2.4 配合任意 HTTP 客户端

import httpx
from headroom import compress

compressed = compress(messages, model="claude-sonnet-4-5-20250929")
httpx.post(
    "https://api.anthropic.com/v1/messages",
    json={
        "model": "claude-sonnet-4-5-20250929",
        "messages": compressed.messages,
    },
    headers={"X-Api-Key": api_key, "anthropic-version": "2023-06-01"},
)

2.5 compress() 的返回值

result = compress(messages, model="gpt-4o")
result.messages  # list[dict] — compressed messages, same format as input
result.tokens_before  # int — original token count
result.tokens_after  # int — compressed token count
result.tokens_saved  # int — tokens removed
result.compression_ratio  # float — 0.0 (no savings) to 1.0 (100% removed)
result.transforms_applied  # list[str] — what ran (e.g., ["router:smart_crusher:0.35"])

2.6 源码剖析：compress() 内部做了什么

对照 headroom/compress.py 的实现，可以确认几个关键事实：

1. 签名与默认值（见 headroom/compress.py#L171-L179）：compress(messages, model="claude-sonnet-4-5-20250929", model_limit=200000, optimize=True, hooks=None, config=None, **kwargs)。model 只用于 token 计数和上下文上限，不参与转发；optimize=False 时直接透传，可用于 A/B 对照。
2. 内部流水线：compress() 通过 _get_pipeline() 懒加载单例 TransformPipeline，默认顺序为 CacheAligner → ContentRouter——前者稳定前缀以提升 provider KV 缓存命中率，后者按内容类型路由到对应压缩器（JSON 走 SmartCrusher、代码走 CodeCompressor、文本走 Kompress），见 headroom/compress.py#L409-L418 的注释。
3. 膨胀护栏：如果"压缩"后 token 数反而变大（tokens_after > tokens_before），函数会整体回退到原始消息并标记 transforms_applied=["inflation_guard:reverted"]，见 headroom/compress.py#L275-L291。这意味着库调用方永远不会收到比输入更大的 payload。
4. 异常安全：压缩过程中任何异常都会被捕获，记录 OTel 指标后原样返回原始消息（见 headroom/compress.py#L349-L362），保证压缩层故障不会打断业务链路。
5. 用户查询提取：调用前会用 _extract_user_query(messages) 提取用户查询作为相关性上下文，使 SmartCrusher 按相关性而非纯统计量挑选保留项（见 headroom/compress.py#L249-L253）。

2.7 可配置项：CompressConfig

compress() 支持通过 config=CompressConfig(...) 或 kwargs 简写传入压缩选项。headroom/compress.py#L77-L147 定义了完整的字段：

字段	默认值	含义
compress_user_messages	False	是否压缩 user 消息。编码代理场景默认跳过；文档压缩/RAG 管线或 user 消息含大工具输出时设为 True
compress_system_messages	True	是否压缩 system 消息。语音代理等必须原样保留指令的场景设为 False
protect_recent	4	不压缩最后 N 条消息（活跃对话）。设 0 表示全部可压缩
protect_analysis_context	True	检测 "analyze"/"review" 意图并保护代码不被压缩
frozen_message_count	0	已被 provider prompt cache 锚定的前缀消息数，转换不会改写该前缀，避免把 0.1x 的缓存前缀读变成全价重写。代理处理器会自动计算并传入；库模式下自管会话循环的调用方应传上一请求的消息数
target_ratio	None	Kompress 的保留比例：None=模型自决（约保留 15%，较激进），0.5=保留 50%（文档安全），0.7=保守。只影响文本压缩，SmartCrusher（JSON）走数组去重逻辑
min_tokens_to_compress	250	单条消息低于此 token 数则不压缩
kompress_model	None	Kompress 模型 ID，None=默认模型；设 'disabled' 完全跳过 ML 压缩（只剩 SmartCrusher + CacheAligner）
savings_profile	None	命名的高节省档位，例如 Codex/Claude/Cursor 用的 'agent-90'

典型调参组合（源自源码 docstring）：

# 金融文档：全部压缩、保留 50%
compress(messages, model="claude-opus-4-20250514",
    compress_user_messages=True,
    target_ratio=0.5,
    protect_recent=0,
)

# 日志/搜索结果：激进压缩
compress(messages, model="gpt-4o", target_ratio=0.2)

另外，headroom/compress.py#L365-L395 还提供了 compress_spreadsheet(path, ...)：把 .xlsx/.xls 每个 sheet 渲染为 CSV 并逐 sheet 走表格压缩器（无优先、有损 CCR 兜底），需要 pip install headroom-ai[spreadsheet] 扩展。

三、LiteLLM 集成：一行回调

如果你已经用 LiteLLM 作为 LLM 网关，把 Headroom 注册为回调即可，所有调用自动压缩：

import litellm
from headroom.integrations.litellm_callback import HeadroomCallback

litellm.callbacks = [HeadroomCallback()]

# All calls now compressed automatically
response = litellm.completion(model="gpt-4o", messages=[...])
response = litellm.completion(model="bedrock/claude-sonnet", messages=[...])
response = litellm.completion(model="azure/gpt-4o", messages=[...])

3.1 回调的底层机制

查看 headroom/integrations/litellm_callback.py 可知：

• HeadroomCallback 继承 litellm.integrations.custom_logger.CustomLogger，实现 async_pre_call_hook——LiteLLM 在每次 API 调用前触发该钩子，对 call_type 为 completion/acompletion 的请求压缩 data["messages"]，压缩成功（tokens_saved > 0）才替换消息（见 headroom/integrations/litellm_callback.py#L109-L156）。继承 CustomLogger 而非裸对象，是为了兼容 LiteLLM 未来新增钩子的 no-op 默认实现（源码注释提及 #1114 缺陷修复）。
• 回调累计 total_tokens_saved 属性（headroom/integrations/litellm_callback.py#L87-L90），可直接用于日志与报表。
• 支持云模式：构造时传 api_key="hdr_xxx"（或设置环境变量 HEADROOM_API_KEY），压缩将走 Headroom Cloud 的 /v1/saas/compress 端点，获得组织级 CCR、TOIN 学习与分析面板；云模式依赖 httpx。
• 压缩失败时仅记录 warning 并使用原始消息，绝不阻断请求（见 headroom/integrations/litellm_callback.py#L153-L154）。

3.2 配合 LiteLLM Proxy

如果 LiteLLM 以代理服务器形态运行，有两种方式：

方式一——ASGI 中间件（推荐，见下节）：

# In your LiteLLM proxy startup
from litellm.proxy.proxy_server import app
from headroom.integrations.asgi import CompressionMiddleware

app.add_middleware(CompressionMiddleware)

方式二——在 LiteLLM 配置文件中注册回调：

# litellm_config.yaml
litellm_settings:
  callbacks: ["headroom.integrations.litellm_callback.HeadroomCallback"]

（源码 docstring 补充：云模式下可同时声明 environment_variables: HEADROOM_API_KEY: "hdr_xxx"，见 headroom/integrations/litellm_callback.py#L54-L61。）

四、ASGI 中间件：给任何 ASGI 应用加压缩

CompressionMiddleware 是面向 FastAPI、Starlette、LiteLLM proxy 或任何自研 ASGI 代理的即插即用中间件：

from headroom.integrations.asgi import CompressionMiddleware

# FastAPI
app = FastAPI()
app.add_middleware(CompressionMiddleware)

# Starlette
app = Starlette(routes=[...])
app.add_middleware(CompressionMiddleware)

# LiteLLM proxy
from litellm.proxy.proxy_server import app

app.add_middleware(CompressionMiddleware)

4.1 拦截范围与透传规则

对照 headroom/integrations/asgi.py#L41-L47 的 _LLM_PATHS 常量，中间件只拦截对以下四个路径的 POST 请求，其余请求原样透传：

• /v1/messages（Anthropic）
• /v1/chat/completions（OpenAI）
• /v1/responses（OpenAI Responses API）
• /chat/completions（LiteLLM 无 /v1 前缀的路径）

匹配采用 path.endswith(p) or path == p（headroom/integrations/asgi.py#L106），因此带前缀的路径同样可命中。请求体在 ASGI 层被完整缓冲后再解析 messages 字段，压缩后重新序列化为新的 receive 回调传给下游；非 JSON 请求只记 debug 日志并跳过。

4.2 响应头指标

压缩发生时，中间件在 http.response.start 上追加以下响应头（headroom/integrations/asgi.py#L176-L184）：

• x-headroom-compressed: true —— 已应用压缩
• x-headroom-tokens-before: <int> —— 压缩前 token 数
• x-headroom-tokens-after: <int> —— 压缩后 token 数
• x-headroom-tokens-saved: <int> —— 减少的 token 数

4.3 本地模式与云模式

中间件构造参数（headroom/integrations/asgi.py#L65-L84）：min_tokens=500、model_limit=200000、hooks=None、api_key=None、api_url=None。

• 本地模式（默认）：进程内调用 headroom.compress() 完成压缩；
• 云模式（app.add_middleware(CompressionMiddleware, api_key="hdr_xxx") 或设置 HEADROOM_API_KEY）：POST 到 {api_url}/v1/saas/compress（默认 https://api.headroomlabs.ai，可用 HEADROOM_API_URL 覆盖），获得组织级 CCR、TOIN 学习与分析面板；此模式要求安装 httpx。

五、Headroom 代理：非 Python 应用与只支持改 Base URL 的工具

Headroom 代理是一个独立 HTTP 服务器，最适合非 Python 应用，或只能配置 base URL 的工具（Claude Code、Cursor、GitHub Copilot CLI）：

pip install "headroom-ai[all]"
headroom proxy --port 8787

# Claude Code
ANTHROPIC_BASE_URL=http://localhost:8787 claude

# GitHub Copilot CLI
headroom wrap copilot -- --model claude-sonnet-4-20250514

# Cursor / Any OpenAI client
OPENAI_BASE_URL=http://localhost:8787/v1 cursor

对于翻译类后端（translated backends），Copilot 包装器可切换到 Headroom 的 OpenAI 兼容路由：

headroom wrap copilot --backend anyllm --anyllm-provider groq -- --model gpt-4o

关于 Copilot 托管 API（--subscription 及隐式 OAuth 路径），Headroom 路由到通用主机 https://api.githubcopilot.com（提供完整模型集）；企业/数据驻留租户的专属 Copilot 主机通过 GITHUB_COPILOT_API_URL 固定下来，例如 export GITHUB_COPILOT_API_URL=https://api.<your-host>.githubcopilot.com，该覆盖值会透传到上游请求。这一行为在源码中得到印证：GITHUB_COPILOT_API_URL 出现在 headroom/copilot_auth.py、headroom/providers/proxy_targets.py 与 headroom/cli/wrap.py 中。详见 TESTING-copilot-subscription.md。

5.1 云厂商后端

代理可通过 --backend 直连云厂商（--backend、--region、--port 参数定义于 headroom/cli/proxy.py）：

# AWS Bedrock
headroom proxy --backend bedrock --region us-east-1

# Google Vertex AI
headroom proxy --backend vertex_ai --region us-central1

# Azure OpenAI
headroom proxy --backend azure

# OpenRouter (400+ models)
OPENROUTER_API_KEY=sk-or-... headroom proxy --backend openrouter

全部选项（含认证、TLS、CCR 存储等）见 代理文档。

六、Agno 智能体框架集成

对 Agno agent 框架的完整集成只需包装模型：

from agno.agent import Agent
from agno.models.anthropic import Claude
from headroom.integrations.agno import HeadroomAgnoModel

model = HeadroomAgnoModel(Claude(id="claude-sonnet-4-20250514"))
agent = Agent(model=model, tools=[your_tools])
response = agent.run("Investigate the issue")

print(f"Tokens saved: {model.total_tokens_saved}")

源码层面，HeadroomAgnoModel 定义在 headroom/integrations/agno/model.py#L68，直接继承 agno.models.base.Model，因此与 Agno Agent 完全兼容，可包装 OpenAIChat、Claude、Gemini 等任意 Agno 模型，在每次 API 调用前自动优化上下文；agno 为可选依赖，未安装时 agno_available() 返回 False。此外 headroom/integrations/init.py 还导出了 HeadroomPreHook/HeadroomPostHook 与 create_headroom_hooks 便捷函数，用于 agent 级指标跟踪。Agno 集成还支持 CrewAI（pip install headroom[crewai]）与 AutoGen（pip install headroom[autogen]）的 HeadroomToolWrapper 工具输出压缩。

hooks、多 provider 与流式细节见 Agno 指南。

七、LangChain 集成

与 LangChain 的完整集成覆盖聊天模型、记忆、检索器、工具包装与流式：

from langchain_openai import ChatOpenAI
from headroom.integrations import HeadroomChatModel

llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))
response = llm.invoke("Hello!")

从 headroom/integrations/init.py#L47-L82 的重导出清单可以看到完整的组件族：

• HeadroomChatModel：任意 LangChain 聊天模型的即插即用包装器
• HeadroomChatMessageHistory：对话历史自动压缩
• HeadroomDocumentCompressor：基于相关性的文档过滤
• HeadroomToolWrapper / wrap_tools_with_headroom：agent 工具输出压缩
• StreamingMetricsTracker / StreamingMetrics：流式过程中的 token 计数
• HeadroomLangSmithCallbackHandler：LangSmith trace 增强

细节与已知限制见 LangChain 指南。

八、TypeScript SDK

面向 Node.js、Next.js 及任意 TypeScript/JavaScript 应用：

npm install headroom-ai

SDK 提供三类适配器：Vercel AI SDK 的 headroomMiddleware() 中间件、OpenAI Node SDK 的 withHeadroom() 客户端包装器、Anthropic TS SDK 的 withHeadroom() 客户端包装器。完整文档（含示例代码）见 TypeScript SDK 指南，源码位于 sdk/typescript 目录。

九、OpenClaw 插件

Headroom 还为 OpenClaw agents 提供上下文压缩插件：

headroom wrap openclaw

将其配置为上下文引擎：

{ "plugins": { "slots": { "contextEngine": "headroom" } } }

不使用 CLI 包装器时也可手动安装：

pip install "headroom-ai[proxy]"
openclaw plugins install --dangerously-force-unsafe-install headroom-ai/openclaw

插件会自动检测正在运行的 Headroom 代理，检测不到就自行启动一个；压缩发生在 OpenClaw 的 assemble() 阶段，对 agent 行为零改动。插件源码位于 plugins/openclaw 目录，可直接查阅完整实现。

十、压缩钩子（进阶）：不修改 Headroom 代码即可定制压缩行为

CompressionHooks 提供三个钩子，分别位于流水线的三个明确阶段（定义于 headroom/hooks.py）：

from headroom import compress, CompressionHooks, CompressContext


class MyHooks(CompressionHooks):
    def pre_compress(self, messages, ctx):
        # Modify messages before compression (dedup, filter, inject)
        return messages

    def compute_biases(self, messages, ctx):
        # Per-message compression aggressiveness
        # >1.0 = keep more, <1.0 = compress more
        return {5: 1.5, 6: 0.5}  # Keep message 5, compress message 6

    def post_compress(self, event):
        # Observe results (logging, analytics, learning)
        print(f"Saved {event.tokens_saved} tokens")


result = compress(messages, model="gpt-4o", hooks=MyHooks())

10.1 三个钩子的契约

对照 headroom/hooks.py#L73-L143：

钩子	时机	返回	典型用途
pre_compress(messages, ctx)	压缩流水线运行前	修改后的消息列表	跨轮去重、记忆注入、预过滤无关消息、按任务阶段重排
compute_biases(messages, ctx)	压缩前	{消息索引: 偏置浮点数}	1.0=默认，>1.0=多保留，<1.0=更激进压缩，缺失索引按 1.0 处理。适合位置感知压缩（中段注意力最弱、给更高偏置）、阶段感知预算、TOIN 学习出的 per-tool 偏置
post_compress(event)	压缩完成后	无（观察型）	失败驱动学习、组织级分析面板、A/B 测试、异常检测

配套数据结构：CompressContext（携带 model、user_query、turn_number、tool_calls、provider）和 CompressEvent（携带 tokens_before/after/saved、compression_ratio、transforms_applied、ccr_hashes 等完整指标）。

compress() 中钩子的实际调用时序可在 headroom/compress.py#L231-L238 确认：先 pre_compress，再 compute_biases，压缩完成后仅当 tokens_saved > 0 才触发 post_compress（headroom/compress.py#L325-L338）。除这三个钩子外，CompressionHooks 还有第四个方法 on_pipeline_event，用于观察规范化流水线的生命周期事件；钩子同样可传入 ASGI 中间件与 LiteLLM 回调的 hooks 构造参数，实现 SDK、compress() 与代理三条链路共用同一套定制逻辑。钩子与流水线的完整集成方式见 架构文档。

十一、FAQ

Q：Headroom 会改变响应格式吗？ 不会。LLM 返回的响应格式不变，Headroom 只修改输入消息。

Q：如果压缩删掉了 LLM 需要的内容怎么办？ Headroom 将原始内容存入 CCR（Compress-Cache-Retrieve），LLM 可以调用 headroom_retrieve 取回完整未压缩内容；压缩摘要会告诉 LLM 有哪些可取回的内容。

Q：支持流式吗？ 支持。压缩发生在请求发出之前，流式响应不受影响。

Q：会增加多少延迟？ 视内容大小与类型约 15–200ms：小 JSON 数组约 15ms，大工具输出 100–200ms。token 节省在 LLM 端省下的生成时间通常远大于压缩开销——对一次 Sonnet 调用做 50% 的 token 削减，能省下数秒生成时间。真实数字见 延迟基准。

十二、小结

Headroom 的集成设计遵循"从两行代码到一条命令"的梯度：库调用方用 compress() + CompressConfig 精确控制压缩范围与激进度，网关方用 LiteLLM 回调或 ASGI 中间件获得透明压缩与 x-headroom-* 指标头，客户端方则通过代理改 base URL 即可接入，框架方（Agno/LangChain/CrewAI/AutoGen/MCP/TS SDK）各有对应包装器。所有路径共享同一条 CacheAligner → ContentRouter 压缩流水线与同一套 Hooks 契约，并内置膨胀回退与异常透传护栏，保证"接入 Headroom"永远不比"不接入"更脆弱。