Headroom × LangChain 实战：用 SmartCrusher 压缩 Agent 工具输出，节省 74% Token 且 100% 保留 ERROR

本文以仓库 examples/langchain_demo/ 下的 LangChain 演示套件为核心，讲解 Headroom 如何在 LangChain Agent 的工具调用链路上做上下文压缩：先通过三个可直接运行的脚本（无需 API Key 即可演示）看到 SmartCrusher 对大 JSON 工具输出的 before/after 效果，再结合 smart_crusher 源码、SmartCrusherConfig 配置 与 HeadroomChatModel 集成层 拆解其"保 ERROR、保首尾、保异常点、按相关性打分"的压缩策略，最后给出完整 Agent 前后对比的运行方式与成本测算方法。

为什么工具输出是 Agent 的 Token 大头

examples/langchain_demo/ 是一个面向真实场景的演示：模拟一个客服/运维 Agent，它通过 5 个工具查用户库、搜文档、翻日志、看指标、拉 API 数据——每个工具都返回 50~200 条记录的 JSON。这类"数据库查询返回上百行、日志搜索返回几百条"的输出，正是 Agent 对话中 token 膨胀的主要来源。

演示目录的完整文件布局（见 examples/langchain_demo/README.md）：

文件	作用
mock_tools.py	5 个仿真工具，生成真实感的大体积 JSON 输出
show_compression.py	独立压缩演示，无需 API Key
verify_errors_kept.py	验证 ERROR 条目 100% 保留
run_comparison.py	完整 Agent before/after 对比（需 OPENAI_API_KEY）

演示工具集：mock_tools.py 输出了什么

mock_tools.py 用随机数据模拟 5 类真实 API 响应，并在 TOOL_FUNCTIONS 字典 中统一注册：

工具	模拟场景	条数	顶层 JSON 键
search_users	用户库查询（部门、状态、角色、偏好元数据）	100	results
search_logs	日志检索（DEBUG/INFO/WARN/ERROR，按时间倒序）	200	entries
get_metrics	5 分钟粒度时序指标（CPU、内存、延迟、错误率，5% 概率注入异常点）	100	metrics
search_docs	文档/知识库检索（按 relevance_score 排序）	50	results
fetch_api_data	分页 API（含 pagination 元信息）	75	data

几个细节值得注意：

• 日志生成器 generate_log_entries 用权重列表 ["DEBUG", "INFO", "INFO", "INFO", "WARN", "ERROR"] 让大多数条目是 INFO，只有少数是 ERROR——这正好复现"几百条日志里只有几条是错误"的真实排查场景。
• 指标生成器 generate_metrics_data 以 5% 概率注入 CPU 60-95%、错误率 5-15% 的异常点，用于验证 SmartCrusher 的统计异常检测能力。
• search_docs 的返回会按 relevance_score 降序排序，fetch_api_data 带 pagination.total_pages=10，用来考察"保留首尾条目 + 相关性 Top-N"策略。

快速开始：三个脚本的运行方式

以下命令均在仓库根目录下执行，依赖 tiktoken（以及完整对比所需的 LangChain）：

# 1. 演示压缩效果（无需 API Key）
PYTHONPATH=. python -m examples.langchain_demo.show_compression

# 2. 验证 ERROR 条目 100% 保留
PYTHONPATH=. python -m examples.langchain_demo.verify_errors_kept

# 3. 完整 Agent 对比（需要 OPENAI_API_KEY）
export OPENAI_API_KEY='your-key-here'
PYTHONPATH=. python -m examples.langchain_demo.run_comparison

show_compression：单工具 before/after 压缩演示

show_compression.py 的核心流程在 demonstrate_compression() 中，值得逐段看，因为它展示了 Headroom 的"transform 级"用法（不经过完整 pipeline，直接调用 SmartCrusher）：

第一步：构造 SmartCrusher 并注入上下文。 脚本用 SmartCrusherConfig 配置压缩器，并通过 OpenAIProvider().get_token_counter("gpt-4o") 获取真实 tokenizer（token 计数接口定义见 providers/base.py）：

from headroom.config import SmartCrusherConfig
from headroom.providers import OpenAIProvider
from headroom.transforms import SmartCrusher

smart_config = SmartCrusherConfig(
    enabled=True,
    min_tokens_to_crush=200,   # 只有超过 200 token 才压缩
    max_items_after_crush=20,  # 压缩后最多保留 20 条
)
provider = OpenAIProvider()
tokenizer = provider.get_token_counter("gpt-4o")
crusher = SmartCrusher(config=smart_config)

第二步：把工具输出放进标准 Agent 对话消息序列。 脚本构造了 system → user（带问题上下文）→ assistant(tool_calls) → tool 四条消息，模拟真实 Agent 中"用户提问后工具返回大 JSON"的形态：

messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": context},          # 用户问题作为相关性上下文
    {"role": "assistant", "content": None,
     "tool_calls": [{"id": "call_1",
                     "function": {"name": tool_name, "arguments": "..."}}]},
    {"role": "tool", "content": raw_output, "tool_call_id": "call_1"},
]
result = crusher.apply(messages, tokenizer=tokenizer)

关键点：SmartCrusher 的 apply() 接收的是整条消息序列，token 计数由调用方以 tokenizer= 参数传入；压缩后的结果从 result.messages 最后一条 tool 消息取出。

第三步：五个演示场景。 main() 依次运行 5 个场景，每个场景的"用户上下文"都刻意指向数据中存在的特征（活跃工程用户、ERROR 日志、CPU 尖峰、认证文档、pending 订单），让相关性打分有命中目标，最后汇总输出各工具的前后 token 数与总成本。

README 记录的实测结果

README 的 Token Savings 表 给出了该演示的基准运行结果（token 计数基于 cl100k_base）：

工具	压缩前	压缩后	节省
search_users (100 items)	15,453	2,014	87%
search_logs (200 items)	25,679	3,213	87%
get_metrics (100 items)	11,517	8,425	27%
search_docs (50 items)	6,912	2,127	69%
fetch_api_data (75 items)	15,786	3,622	77%
合计	75,347	19,401	74%

注意 get_metrics 只有 27% 的节省率——因为时序指标每条记录字段少、数值密度高，且异常点必须保留；而日志/用户库这类"长文本 + 大量无关行"的结构压缩率最高。按 gpt-4o $2.50/1M 输入 token 计价：单请求约 $0.19 → $0.05，按每天 1000 次请求估算每月可节省约 $4,196。这些数字来自演示脚本自身的输出逻辑（show_compression.py 的成本段），实际节省取决于你的真实工具输出分布。

SmartCrusher 的五大保留策略与配置参数

SmartCrusherConfig 的 docstring 写明了设计目标：保留原始 JSON Schema 不变——输出只包含原数组中的条目，不包装、不生成文本、不加元数据。README 概括的五条压缩策略与其配置一一对应：

1. 100% 保留 ERROR 条目——错误项永不丢弃（源码中 "Error items never dropped"）；
2. 保留首/尾条目——first_fraction: 0.3 / last_fraction: 0.15 控制 Kneedle 自适应 K 值中首尾各占的比例（默认 30%/15%，其余名额给重要性打分），用于分页上下文；
3. 保留统计异常点——数值偏离均值超过 variance_threshold（默认 2.0 个标准差）的条目保留，捕获 CPU 尖峰、内存飙升等；
4. 相关性打分——通过 RelevanceScorerConfig（relevance 字段）让匹配用户查询的条目优先入坑；
5. 变化点保留——preserve_change_points: True，在数据发生显著跳变的位置留样。

完整参数表（默认值均来自 config.py）：

参数	默认值	说明
enabled	True	工具输出压缩的默认实现
min_items_to_analyze	5	小于该条数的小数组不做统计分析
min_tokens_to_crush	200	仅当输出超过该 token 数才压缩
variance_threshold	2.0	变化点检测的标准差倍数（调低如 1.5 更保守）
uniqueness_threshold	0.1	低于该值视为近常量字段
similarity_threshold	0.8	相似字符串聚类阈值
max_items_after_crush	15	压缩后目标最大条数（演示中调为 20）
preserve_change_points	True	保留数据跳变点
factor_out_constants	False	不抽取常量，保持原 schema
include_summaries	False	不生成摘要文本
dedup_identical_items	True	多种保留机制选中同一条目时只保留一份
lossless_min_savings_ratio	0.15	无损表格化路径相对有损路径的最小字节节省比
lossless_only	False	严格无损模式：宁可放弃压缩也不产生 CCR 标记

docstring 同时列出了已知边界（GOTCHAS）：统计分析每条约 5-10ms 开销；变化点检测用固定窗口（5 条），可能漏掉缓慢渐变；Top-N 策略假设"分数越高越相关"。官方建议：关键数据可调高 max_items_after_crush，variance_threshold 调低。

实现入口是 headroom/transforms/smart_crusher.py 中的 SmartCrusher 类（第 242 行），它作为 Transform 接入 Headroom 的 TransformPipeline。

verify_errors_kept：ERROR 保留的自动化验证

verify_errors_kept.py 是一个可直接执行的不变量检查，流程为：

1. 用 generate_log_entries("test-service", count=200) 生成 200 条日志，统计其中 level == "ERROR" 的原始条数；
2. 以相同配置（min_tokens_to_crush=200, max_items_after_crush=20）构造 SmartCrusher，把日志作为 tool 消息压缩，用户上下文设为 "Find ERROR entries in the logs"；
3. 解析压缩后 JSON（若 SmartCrusher 附加了标记文本，脚本会用正则 (\{.*\}) 兜底提取 JSON 主体，见 第 59-73 行），比对 ERROR 条数。

判定逻辑在 第 84-91 行：压缩后 ERROR 条数 ≥ 原始则输出 SUCCESS: All ERROR entries were preserved，部分保留输出 PARTIAL，一条不留输出 FAILURE。README 记录的测试运行结果是 27/27 全部保留，同时 200 条日志被压到约 20 条——即"数量砍到 10%，关键信息 100% 存活"。这也是该演示最想传达的工程原则：压缩的验收标准不是压缩率，而是关键数据的不丢失。

run_comparison：完整 Agent 的 before/after 对比

run_comparison.py 把对比拉到完整 Agent 层面：同一个支持 Agent、同一组工具、同一批用户问题，分别以 baseline（裸模型）和 headroom（包装后模型）各跑一遍。

三个对比场景

SCENARIOS 定义了三个贴近真实工单的问题：

1. User Account Investigation——某用户无法登录，查账户状态 + 日志认证错误 + 相关文档；
2. Service Performance Investigation——payment-service 变慢，查指标异常 + 近期错误日志 + 性能排障文档；
3. Multi-User Issue——工程部门多名用户报错，搜工程用户 + 查 user-service 日志 + 查文档。

两侧的运行方式

Baseline 侧（run_agent_baseline）是标准 LangChain 工具循环：ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)，最多 5 轮迭代，每轮累加输入 token（本地计数）、执行模型请求的工具调用、把结果作为 ToolMessage 追加回对话。

Headroom 侧（run_agent_headroom）唯一区别是用 HeadroomChatModel 包装同一模型，压缩在 invoke() 内部发生，工具循环代码完全不变：

from headroom import HeadroomConfig
from headroom.integrations import HeadroomChatModel

base_model = ChatOpenAI(model="gpt-4o-mini", api_key=api_key, temperature=0)

config = HeadroomConfig(
    smart_crusher_threshold=500,   # 工具输出 > 500 tokens 才压缩
    smart_crusher_max_items=20,    # 最多保留 20 条
    cache_alignment=True,          # 稳定 system prompt 提升缓存命中
    rolling_window=True,
)
headroom_model = HeadroomChatModel(
    wrapped_model=base_model,
    headroom_config=config,
).bind_tools(tools)

跑完后脚本用 headroom_model.get_total_tokens_saved() 取 Headroom 自报的节省量，从累计输入 token 中扣除，得到"实际发送给上游的 token 数"。

两点版本说明需要留意（以当前仓库源码为准）：

• 从 headroom/config.py 的 HeadroomConfig 结构看，当前版本采用嵌套配置：smart_crusher: SmartCrusherConfig 字段承载压缩参数（即 min_tokens_to_crush / max_items_after_crush），示例脚本中扁平的 smart_crusher_threshold=500, smart_crusher_max_items=20 写法属于演示脚本自身的历史 API；在当前源码上运行前建议以 HeadroomConfig(smart_crusher=SmartCrusherConfig(...)) 的写法为准，具体以 headroom/integrations/langchain/chat_model.py 对 HeadroomConfig 的消费方式为准。
• 对比脚本调用的 get_total_tokens_saved() 方法在当前 chat_model.py 中对应的是 total_tokens_saved 属性与 get_savings_summary() 方法，运行新版源码时注意按属性访问。

输出与成本核算

print_comparison() 为每个场景输出对照表：输入/输出 token、工具调用次数、消息数、耗时，以及按 gpt-4o-mini 价格（$0.15/1M 输入、$0.60/1M 输出）估算的美元成本与节省百分比。无 API Key 时脚本自动降级为 SIMULATION 模式：只统计 3 个工具输出的 token 体量并给出约 5 倍压缩的估算（max 20 items 上限下的粗略值）。

集成层原理：为什么 Headroom 选择包装 ChatModel

headroom/integrations/langchain/chat_model.py 的模块 docstring 解释了架构动机：LangChain 的 callback 机制按设计不能修改消息，所以 Headroom 不挂 callback 改消息，而是直接包装 BaseChatModel 本身——HeadroomChatModel（第 118 行）继承自 BaseChatModel，对外表现为一个普通 LangChain 模型。

关键实现点：

• 懒加载 pipeline 与 provider 自动探测：pipeline property 在首次调用时根据 wrapped_model 的类路径探测上游厂商（ChatOpenAI → OpenAIProvider 等），再构建 TransformPipeline(config, provider)，保证 token 计数与目标 provider 一致；
• 工具调用兼容：bind_tools() 被重写（第 512 行），返回包装后的 HeadroomChatModel，因此对比脚本里 .bind_tools(tools) 后依然能继续被 Headroom 拦截；
• 可观测性：每次优化产生一条 OptimizationMetrics（tokens_before/after、savings_percent、transforms_applied），累积在 metrics_history 与 total_tokens_saved，get_savings_summary() 提供汇总（第 522 行）；
• 多入口：除 HeadroomChatModel 外，integrations/langchain/init.py 还导出 Agent/Retriever/Streaming/LangGraph 等封装，并有独立的 optimize_messages() 函数（第 916 行）供手工调用；

依赖方面，pyproject.toml 提供 langchain extra（langchain-core>=1.3.3、langchain-openai>=1.1.14，见 pyproject.toml 第 212-214 行），演示脚本则额外要求 tiktoken。

测试与回归验证

LangChain 集成的回归测试集中在 tests/test_integrations/langchain/ 目录，包含 test_chat_model.py、test_evals.py、test_agents.py、test_streaming.py、test_langgraph.py 等。README 提到的 12 项 eval 覆盖：ERROR 保留（100%）、异常检测、相关性匹配、压缩效率、schema 保留与边缘用例。

需要说明：README 中给出的 eval 命令 pytest tests/test_integrations/test_langchain_evals.py -v 对应的文件在当前仓库中不存在（实际为 tests/test_integrations/langchain/test_evals.py），运行前请以目录内实际文件名为准：

PYTHONPATH=. pytest tests/test_integrations/langchain/ -v

适用前提与小结

• 环境：仓库根目录运行，PYTHONPATH=.；show_compression / verify_errors_kept 仅需 tiktoken；run_comparison 需要 langchain-core、langchain-openai 与 OPENAI_API_KEY（真实模式会产生 API 费用）；
• 数据边界：演示 token 数字基于 mock 数据与 cl100k_base 计数，README 中的 74% 总节省率与成本折算是该仿真场景下的结果，不代表所有工作负载；get_metrics 类高密度时序数据节省率明显低于长文本类输出；
• 核心结论：这套演示展示了一条完整的落地路径——用 SmartCrusher 的统计保真压缩（错误/首尾/异常点/相关性/变化点五重保留）处理工具输出，用 HeadroomChatModel 以"包装 ChatModel"方式零侵入接入 LangChain Agent，最终在 token 大幅下降的同时维持关键信息 100% 可达。