Headroom LangChain 集成实战：聊天模型、记忆、检索器与 Agent 的六种上下文压缩接入方式

Headroom 通过 headroom.integrations.langchain 子包将上下文压缩能力嵌入 LangChain 生态，覆盖聊天模型、消息历史、文档检索器、工具调用、流式指标与 LangSmith 可观测性六个层面。本篇以仓库文档 wiki/langchain.md 为主线，逐一对应到 headroom/integrations/langchain/ 下的真实源码实现，讲清楚每个组件的公开 API、默认参数、底层调用链与实测验证方式，读完你可以把 Headroom 无缝接入现有的 LangChain/LangGraph 应用并量化 token 节省。

为什么选择"包装模型"而不是回调

在展开配置之前，先理解一个关键的架构决策。LangChain 的 callback 机制在设计上无法修改消息内容，因此无法用回调实现压缩。Headroom 的解法是直接继承 BaseChatModel 包装模型本身，在每次 API 调用前拦截并变换消息。这一决策写在了 chat_model.py 的模块 docstring 中：

Key insight: LangChain callbacks CANNOT modify messages (by design).
Therefore, we wrap the chat model itself to intercept and transform messages.

从源码结构看，整个子包按职责拆分为七个模块：

模块文件	核心组件	职责
chat_model.py	HeadroomChatModel、HeadroomCallbackHandler、HeadroomRunnable、optimize_messages()	聊天模型包装与消息级优化
memory.py	HeadroomChatMessageHistory	消息历史的阈值触发压缩
retriever.py	HeadroomDocumentCompressor	检索文档的相关性过滤
agents.py	HeadroomToolWrapper、wrap_tools_with_headroom	工具输出压缩与指标
streaming.py	StreamingMetricsTracker、StreamingMetricsCallback	流式输出 token 统计
langsmith.py	HeadroomLangSmithCallbackHandler	压缩指标写入 LangSmith 追踪
langgraph.py	compress_tool_messages、create_compress_tool_messages_node	LangGraph 图级 ToolMessage 压缩

所有组件对 LangChain 均为可选依赖：源码在每个模块内用 try/except ImportError 守卫导入，未安装时提供降级 stub 并在构造时抛出带安装指引的 ImportError。

安装

pip install "headroom-ai[langchain]"

该命令安装 Headroom 并附带 LangChain 依赖。对应 pyproject.toml 中的 extra 定义（约 L212-L221），langchain extra 实际拉取：

• langchain-core>=1.3.3,<4.0
• langchain-openai>=1.1.14,<2.0

如果你要用到 LangGraph 压缩节点（后文 Example 1b 的场景），仓库还单独提供了 langgraph extra，在 langchain extra 之上额外引入 langgraph>=1.0,<2.0：

pip install "headroom-ai[langgraph]"

快速开始：一行代码包装任意聊天模型

from langchain_openai import ChatOpenAI
from headroom.integrations import HeadroomChatModel

# Wrap your model - that's it!
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))

# Use exactly like before
response = llm.invoke("Hello!")

包装后 Headroom 自动完成四件事：

• 检测 provider（OpenAI、Anthropic、Google）
• 压缩对话历史中的工具输出
• 面向 provider 的缓存做前缀优化
• 跟踪 token 节省

底层调用链：一次 invoke 发生了什么

HeadroomChatModel 定义在 chat_model.py，是一个 Pydantic 化的 BaseChatModel 子类，公开字段为：

wrapped_model: Any        # 被包装的 LangChain BaseChatModel
headroom_config: Any      # HeadroomConfig 实例，默认 HeadroomConfig()
mode: HeadroomMode        # AUDIT / OPTIMIZE / SIMULATE，默认 OPTIMIZE
auto_detect_provider: bool  # 默认 True

核心方法是 _generate（chat_model.py#L404-L433），调用链为：

1. _optimize_messages() 把 LangChain 消息（SystemMessage/HumanMessage/AIMessage/ToolMessage）转换为 OpenAI 格式 dict；
2. 惰性构建 TransformPipeline——pipeline property 在首次访问时通过 get_headroom_provider(self.wrapped_model) 自动探测 provider，再取该模型上下文上限 get_context_limit(model)；
3. 调用 pipeline.apply(messages=..., model=..., model_limit=...) 执行变换，得到 tokens_before/tokens_after/transforms_applied，封装为 OptimizationMetrics 记入 _metrics_history（只保留最近 100 条）；
4. 把压缩后的消息转回 LangChain 格式，转发给被包装模型的 _generate。

这里有一个容易踩坑的细节：bind_tools() 返回的是 RunnableBinding，其绑定参数（tools、tool_choice、response_format）只在公共 Runnable 接口生效，直接对 binding 调用私有 _generate 会静默丢失工具定义。Headroom 用 _unwrap_binding()（chat_model.py#L379-L402）显式解开多层 binding、按"外层 binding 优先"的语义合并 kwargs 后再转发；bind_tools 本身也被覆写为返回新的 HeadroomChatModel（chat_model.py#L546-L554）。

# 工具调用无需额外配置：
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Search the web."""
    return {"results": [...]}  # 大 JSON 响应

llm_with_tools = llm.bind_tools([search])
response = llm_with_tools.invoke("Search for Python tutorials")
# 工具输出在后续轮次自动被压缩

异步支持

_agenerate 与 _astream 均被实现，ainvoke/astream 开箱即用：

# 异步调用
response = await llm.ainvoke("Hello!")

# 异步流式
async for chunk in llm.astream("Tell me a story"):
    print(chunk.content, end="", flush=True)

从源码看，异步路径把 CPU 密集的优化步骤放进线程执行器（loop.run_in_executor(None, self._optimize_messages, messages)），避免阻塞事件循环；并且当被包装模型 streaming=True 时，_agenerate 会为该次调用创建 streaming=False 的副本，防止并发 ainvoke 共享可变状态（chat_model.py#L461-L511 的注释中标注了这是针对 GitHub issue #1285 评审反馈的修复）。

查询节省情况

# 使用一段时间后
print(llm.get_savings_summary())
# {'total_requests': 50, 'total_tokens_saved': 12500, 'average_savings_percent': 45.2,
#  'total_tokens_before': ..., 'total_tokens_after': ...}

get_savings_summary() 的实现见 chat_model.py#L556-L572，字段与返回值一一对应；另外源码还暴露了 total_tokens_saved property 和 metrics_history（最近 100 条 OptimizationMetrics）。注意：仓库当前源码中没有 get_metrics() 方法，获取指标请以 get_savings_summary() 或 total_tokens_saved 为准。

集成模式一：聊天模型包装的完整配置

HeadroomChatModel 支持 OpenAI、Anthropic 及自定义配置：

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from headroom.integrations import HeadroomChatModel

# OpenAI
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))

# Anthropic（自动探测）
llm = HeadroomChatModel(ChatAnthropic(model="claude-3-5-sonnet-20241022"))

# 自定义配置
from headroom import HeadroomConfig, HeadroomMode

config = HeadroomConfig(default_mode=HeadroomMode.OPTIMIZE)
llm = HeadroomChatModel(
    ChatOpenAI(model="gpt-4o"),
    config=config,
)

HeadroomChatModel(
    wrapped_model,        # 任意 LangChain BaseChatModel
    headroom_config=HeadroomConfig(),  # Headroom 配置
    auto_detect_provider=True,  # 从被包装模型自动探测 provider
)

provider 自动探测逻辑在 providers.py 的 get_headroom_provider()：根据被包装模型的类路径匹配（如 ChatAnthropic → AnthropicProvider），确保 token 计数使用与目标 provider 一致的 tokenizer。

集成模式二：记忆层自动压缩（HeadroomChatMessageHistory）

长对话很容易膨胀到 50K+ tokens。HeadroomChatMessageHistory 包装任意 BaseChatMessageHistory，在历史超过 token 阈值时自动压缩旧轮次：

from langchain.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import ChatMessageHistory
from headroom.integrations import HeadroomChatMessageHistory

# 包装任意 history
base_history = ChatMessageHistory()
compressed_history = HeadroomChatMessageHistory(
    base_history,
    compress_threshold_tokens=4000,  # 超过 4K tokens 时触发压缩
    keep_recent_turns=5,  # 始终保留最近 5 轮
)

# 与任意 memory 类组合
memory = ConversationBufferMemory(chat_memory=compressed_history)

# 你的 chain 零改动！
chain = ConversationChain(llm=llm, memory=memory)

为什么重要：长对话可以膨胀到 50K+ tokens。HeadroomChatMessageHistory 自动压缩旧轮次、保留近期上下文。

# 查看压缩统计
print(compressed_history.get_compression_stats())
# {'compression_count': 12, 'total_tokens_saved': 28000}

源码级实现：阈值检查与 live-zone 压缩

实现位于 memory.py：

• 构造签名（memory.py#L98-L124）：base_history、compress_threshold_tokens=4000、keep_recent_turns=5、model="gpt-4o"（用于 token 计数）、provider=None（缺省用 OpenAIProvider）；
• 压缩是读时触发的：messages property 每次访问时用 provider tokenizer 统计 token，未超阈值直接返回原始消息，超阈值才走 _apply_compression()；
• _apply_compression()（memory.py#L206-L235）把消息转成 OpenAI 格式后交给 TransformPipeline.apply()。源码注释明确说明：在 PR-B1 的 live-zone 重构之后，"丢弃消息"不再是策略，只做 per-block 内容压缩，压缩结果仍可能超过阈值——调用方应把阈值当作参考值而非硬上限。这与旧版 RollingWindow 语义不同，属于源码注释中说明的行为约定；
• 之所以能适配任何存储后端（内存、Redis、PostgreSQL、LangGraph checkpointer），是因为包装发生在存储层接口（add_message/messages）之上，对底层实现无感知。

集成模式三：检索器文档过滤（HeadroomDocumentCompressor）

向量检索经常返回大量"边缘相关"文档。HeadroomDocumentCompressor 是一个 BaseDocumentCompressor，配合 ContextualCompressionRetriever 实现"高召回检索 + 高精确压缩"两段式：

from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.vectorstores import FAISS
from headroom.integrations import HeadroomDocumentCompressor

# 建立向量库检索器（多召回以保证 recall）
vectorstore = FAISS.from_documents(documents, embeddings)
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 50})

# 用 Headroom 压缩包装（保留最优以保证 precision）
compressor = HeadroomDocumentCompressor(
    max_documents=10,  # 最多保留 10 篇
    min_relevance=0.3,  # 最低相关分
    prefer_diverse=True,  # MMR 式多样性选择
)

retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever,
)

# 检索 50 篇，返回最优 10 篇
docs = retriever.invoke("What is Python?")

相关性评分的源码实现

retriever.py 中该类有三个可配置字段：max_documents=10、min_relevance=0.0、prefer_diverse=False。评分与选择逻辑：

• BM25 风格打分（_score_document，retriever.py#L242-L289）：采用简化 BM25（无 IDF，因为拿不到语料库统计），参数固定为 k1=1.5、b=0.75、平均文档长度假设为 100；分数归一化到 0–1，若查询短语在文档中精确出现再额外 +0.3 加成（上限 1.0）；
• MMR 多样性选择（_select_diverse，retriever.py#L304-L348）：prefer_diverse=True 时，按 mmr_score = 0.5 * relevance - 0.5 * max_similarity 贪心选文档，相似度用词项 Jaccard 系数（_document_similarity，retriever.py#L350-L371）；
• 短路优化：文档数不超过 max_documents 时不做裁剪，但仍会记录各文档分数到 _last_metrics；
• 指标：last_metrics property 返回 CompressionMetrics（文档前后数量、移除数、相关分列表），get_compression_stats() 提供 dict 视图。

一个工程细节：BaseDocumentCompressor 的导入路径随 LangChain 版本变化（langchain_core.documents.compressor / compressors / 伞包路径），retriever.py#L53-L86 按顺序尝试三个位置并带 fallback stub——解析到真实基类是 ContextualCompressionRetriever 校验 base_compressor 类型的前提。

集成模式四：Agent 工具输出压缩（wrap_tools_with_headroom）

Agent 场景中每次工具调用都可能往上下文注入 10–50K tokens。wrap_tools_with_headroom 把工具输出在回到上下文之前压缩：

from langchain.agents import create_openai_tools_agent, AgentExecutor
from langchain_core.tools import tool
from headroom.integrations import wrap_tools_with_headroom


@tool
def search_database(query: str) -> str:
    """Search the database."""
    # 返回 1000 条结果的 JSON
    return json.dumps({"results": [...], "total": 1000})


@tool
def fetch_logs(service: str) -> str:
    """Fetch service logs."""
    # 返回 500 条日志
    return json.dumps({"logs": [...]})


# 用压缩包装工具
tools = [search_database, fetch_logs]
wrapped_tools = wrap_tools_with_headroom(
    tools,
    min_chars_to_compress=1000,  # 只压缩大输出
)

# 用包装后的工具创建 agent
agent = create_openai_tools_agent(llm, wrapped_tools, prompt)
executor = AgentExecutor(agent=agent, tools=wrapped_tools)

# 工具输出被自动压缩
result = executor.invoke({"input": "Find users who logged in yesterday"})

按工具维度查看指标：

from headroom.integrations import get_tool_metrics

metrics = get_tool_metrics()
print(metrics.get_summary())
# {
#   'total_invocations': 25,
#   'total_compressions': 18,
#   'total_chars_saved': 450000,
#   'by_tool': {
#     'search_database': {'invocations': 15, 'chars_saved': 320000},
#     'fetch_logs': {'invocations': 10, 'chars_saved': 130000},
#   }
# }

源码级实现要点

实现位于 agents.py：

• HeadroomToolWrapper（agents.py#L137-L195）：__call__ 先 tool.invoke() 拿到输出，长度小于 min_chars_to_compress（默认 1000 字符）时原样返回并记 was_compressed=False；否则交给 MCP 模块的 compress_tool_result(content=output, tool_name=self.name) 压缩——注意这里复用了 MCP 集成的压缩器（agents.py#L42 的导入），压缩失败时静默回退原文；
• schema 保真：as_langchain_tool()（agents.py#L336-L356）用 StructuredTool.from_function 重建工具时，显式继承原工具的 args_schema（_usable_args_schema 会校验 schema 是 dict 或 pydantic 模型，否则回退到推断），保证模型看到的仍是原始类型化参数，而不是 (*args, **kwargs)；
• 指标收集器：ToolMetricsCollector（agents.py#L70-L119）默认使用模块级全局实例（只保留最近 1000 条记录），get_tool_metrics()/reset_tool_metrics() 操作该全局实例；get_summary() 输出即上文示例的字段结构，还包含 average_compression_ratio；
• 从源码结构看，当前 wrap_tools_with_headroom 的签名是 (tools, min_chars_to_compress=1000, metrics_collector=None)（agents.py#L359-L363），文档配置章节提到的 smart_crusher_config 参数在现源码中对应的是可注入 metrics_collector；SmartCrusher 压缩实际发生在内部 compress_tool_result 层。

集成模式五：流式指标跟踪

from headroom.integrations import StreamingMetricsTracker

tracker = StreamingMetricsTracker(model="gpt-4o")

for chunk in llm.stream("Write a poem about coding"):
    tracker.add_chunk(chunk)
    print(chunk.content, end="", flush=True)

metrics = tracker.finish()
print(f"\nOutput tokens: {metrics.output_tokens}")
print(f"Duration: {metrics.duration_ms:.0f}ms")

上下文管理器风格：

from headroom.integrations import StreamingMetricsCallback

with StreamingMetricsCallback(model="gpt-4o") as tracker:
    for chunk in llm.stream(messages):
        tracker.add_chunk(chunk)
        print(chunk.content, end="")

print(f"Metrics: {tracker.metrics}")

streaming.py 中的实现要点：

• add_chunk 兼容多种 chunk 形态：AIMessageChunk、ChatGenerationChunk、含 content 键的 dict、纯字符串（_extract_content，streaming.py#L142-L173）；
• output_tokens 是对累积内容用 provider tokenizer 实时计数（streaming.py#L201-L207），finish() 后 StreamingMetrics 还包含 chunk_count、content_length、起止时间与 duration_ms；
• 除文档中的两个类外，该模块还提供两个便捷函数：track_streaming_response(stream, model) 与 track_async_streaming_response(stream, model)，一次性消费整个流并返回 (累积内容, 指标) 元组（streaming.py#L277-L341）。

集成模式六：LangSmith 追踪增强

把 Headroom 指标写进 LangSmith trace，让每次调用的压缩收益在追踪平台可见：

from headroom.integrations import HeadroomLangSmithCallbackHandler

# 创建 callback handler
langsmith_handler = HeadroomLangSmithCallbackHandler()

# 与你的 LLM 一起使用
llm = HeadroomChatModel(
    ChatOpenAI(model="gpt-4o"),
    callbacks=[langsmith_handler],
)

# 调用后，LangSmith trace 中会出现这些指标：
# - headroom.tokens_before
# - headroom.tokens_after
# - headroom.tokens_saved
# - headroom.compression_ratio

源码中的 metadata 键名与上报条件

langsmith.py 中 handler 实际上写入 run metadata 的键为（_attach_metrics_to_run, langsmith.py#L228-L256）：

• headroom.tokens_before / headroom.tokens_after / headroom.tokens_saved
• headroom.savings_percent（压缩率，保留两位小数）
• headroom.transforms_applied（应用的变换列表）
• headroom.optimization_timestamp

handler 自动通过 LangSmith Client 的 update_run(..., extra={"metadata": ...}) 上报（失败仅记 debug 日志，不阻断业务）。上报前提是 LANGCHAIN_API_KEY 已设置且 LANGCHAIN_TRACING_V2=true；模块还提供 is_langsmith_available() 与 is_langsmith_tracing_enabled() 两个检查函数（langsmith.py#L309-L324）。get_summary() 可跨 run 汇总 total_runs、total_tokens_saved、average_savings_percent。

实战示例一：LangGraph ReAct Agent

ReAct 是最常见的 agent 架构，优化方式如下：

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from headroom.integrations import HeadroomChatModel, wrap_tools_with_headroom


# 定义返回大输出的工具
@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    # 模拟大型搜索结果
    return json.dumps(
        {
            "results": [
                {"title": f"Result {i}", "snippet": "..." * 100, "url": f"https://..."}
                for i in range(100)
            ],
            "total": 1000,
        }
    )


@tool
def query_database(sql: str) -> str:
    """Execute SQL query."""
    return json.dumps(
        {
            "rows": [{"id": i, "data": "..." * 50} for i in range(500)],
            "total": 500,
        }
    )


# 用 Headroom 包装模型
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))

# 用压缩包装工具
tools = wrap_tools_with_headroom([search_web, query_database])

# 创建 ReAct agent
agent = create_react_agent(llm, tools)

# 运行 —— 工具输出在迭代之间被自动压缩
result = agent.invoke(
    {"messages": [("user", "Find all users who signed up last week and their activity")]}
)

# 查看节省
print(f"Tokens saved: {llm.total_tokens_saved}")

注：原文档此处写作 llm.get_metrics()['tokens_saved']；对照 chat_model.py 源码，当前实现暴露的是 total_tokens_saved property 与 get_savings_summary()，上例已按源码修正。

效果对比：无 Headroom 时，每次工具调用向上下文追加 10–50K tokens；接入后工具输出被压缩到 1–2K tokens 量级，agent 迭代更快、更省。

实战示例二：LangGraph 自定义图中的压缩节点

如果不用 create_react_agent，而是自建 LangGraph StateGraph，可以在 tools 与 agent 节点之间插入压缩节点，压缩图状态里全部 ToolMessage 内容再交给 LLM：

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from headroom.integrations.langchain import create_compress_tool_messages_node


# 定义你的 agent 与 tools 节点
def agent_node(state: MessagesState):
    llm = ChatOpenAI(model="gpt-4o")
    response = llm.invoke(state["messages"])
    return {"messages": [response]}


def tools_node(state: MessagesState):
    # 你的工具执行逻辑
    ...


# 构建带压缩步骤的图
graph = StateGraph(MessagesState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tools_node)
graph.add_node(
    "compress",
    create_compress_tool_messages_node(
        min_tokens_to_compress=100,  # 只压缩约 >100 tokens 的输出
    ),
)

# 接线：tools -> compress -> agent（而非 tools -> agent 直连）
graph.add_edge(START, "agent")
graph.add_edge("tools", "compress")
graph.add_edge("compress", "agent")
# ... 按需添加 agent 到 tools/END 的条件边

app = graph.compile()
result = app.invoke({"messages": [HumanMessage(content="Find sales data")]})

也可以直接把 compress_tool_messages 当独立函数使用：

from headroom.integrations.langchain import compress_tool_messages

# 压缩任意 LangChain 消息列表中的 ToolMessage
result = compress_tool_messages(messages, min_tokens_to_compress=100)
compressed_messages = result.messages
print(f"Saved {result.total_tokens_saved} tokens across {result.messages_compressed} messages")

源码中的错误保护与配置

langgraph.py 模块注释说明它面向 LangGraph 的 ToolMessage 溢出问题，配置集中在 CompressToolMessagesConfig（langgraph.py#L91-L104）：

• min_tokens_to_compress=100：ToolMessage 估算 token（len(text)//4 启发式）低于该值不压缩；
• preserve_errors=True：内容命中错误指示符时跳过压缩；
• error_indicators 默认为 ('"error"', '"ERROR"', "Error:", "Traceback")——错误栈和错误信息绝不能被压缩器改动，这是保证 agent 排障能力的护栏；
• 压缩实际由 SmartCrusher/SmartCrusherConfig 执行，并对配置中排除的工具名（is_tool_excluded）跳过。

CompressToolMessagesResult 暴露 messages、total_tokens_saved、messages_compressed 属性，与上例用法一致（langgraph.py#L107-L120）；create_compress_tool_messages_node 工厂定义在 langgraph.py#L366。

实战示例三：RAG 管道 + 文档过滤

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.retrievers import ContextualCompressionRetriever
from headroom.integrations import HeadroomChatModel, HeadroomDocumentCompressor

# 建立向量库
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(documents, embeddings)

# 高召回检索器（取更多候选）
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 50})

# Headroom 压缩器负责精度
compressor = HeadroomDocumentCompressor(
    max_documents=5,  # 只保留 top 5
    min_relevance=0.4,  # 相关度必须 ≥ 40%
    prefer_diverse=True,  # 避免冗余文档
)

# 组合成压缩检索器
retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever,
)

# 包装 LLM
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))

# 构建 QA 链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=retriever,
    return_source_documents=True,
)

# 查询：检索 50 篇，使用最好的 5 篇
result = qa_chain.invoke({"query": "How do I configure authentication?"})
print(f"Answer: {result['result']}")
print(f"Sources: {len(result['source_documents'])} docs")

影响估算（原文档给出的量化口径）：

• 不过滤：50 篇 × 约 500 tokens = 25K 上下文 tokens；
• 接入 Headroom：5 篇 × 约 500 tokens = 2.5K 上下文 tokens（约 90% 缩减）。

实战示例四：带记忆的会话 Agent

from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain.chains import ConversationChain
from headroom.integrations import HeadroomChatModel, HeadroomChatMessageHistory

# 包装 LLM
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))

# 包装记忆，自动压缩
base_history = ChatMessageHistory()
compressed_history = HeadroomChatMessageHistory(
    base_history,
    compress_threshold_tokens=8000,  # 超过 8K 触发
    keep_recent_turns=10,  # 始终保留最近 10 轮
)

memory = ConversationBufferMemory(
    chat_memory=compressed_history,
    return_messages=True,
)

# 创建会话链
chain = ConversationChain(llm=llm, memory=memory)

# 长对话 —— 记忆自动压缩
for i in range(100):
    response = chain.invoke({"input": f"Tell me about topic {i}"})
    print(f"Turn {i}: {len(response['response'])} chars")

# 查看记忆统计
print(compressed_history.get_compression_stats())
# {'compression_count': 8, 'total_tokens_saved': 45000}

影响：无压缩时 100 轮对话 = 100K+ tokens；接入 HeadroomChatMessageHistory 后上下文维持在阈值（8K）附近，同时保留近期完整对话。

实战示例五：多工具研究 Agent

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
from headroom.integrations import (
    HeadroomChatModel,
    wrap_tools_with_headroom,
    get_tool_metrics,
    reset_tool_metrics,
)


@tool
def search_arxiv(query: str) -> str:
    """Search arXiv for papers."""
    return json.dumps(
        {"papers": [{"title": f"Paper {i}", "abstract": "..." * 200} for i in range(50)]}
    )


@tool
def search_github(query: str) -> str:
    """Search GitHub repositories."""
    return json.dumps(
        {
            "repos": [
                {"name": f"repo-{i}", "description": "..." * 100, "stars": i * 100}
                for i in range(100)
            ]
        }
    )


@tool
def fetch_documentation(url: str) -> str:
    """Fetch documentation from URL."""
    return "..." * 5000  # 大型文档内容


# 全部包装
llm = HeadroomChatModel(ChatOpenAI(model="gpt-4o"))
tools = wrap_tools_with_headroom([search_arxiv, search_github, fetch_documentation])

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "You are a research assistant. Use tools to gather information."),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ]
)

agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 重置本次会话的指标
reset_tool_metrics()

# 运行复杂研究任务
result = executor.invoke(
    {
        "input": "Research the latest advances in LLM context compression and find relevant GitHub projects"
    }
)

# 查看按工具维度的指标
metrics = get_tool_metrics().get_summary()
print(f"Total chars saved: {metrics['total_chars_saved']:,}")
print(f"Per-tool breakdown: {metrics['by_tool']}")

配置参数速查

HeadroomChatModel

HeadroomChatModel(
    wrapped_model,        # 任意 LangChain BaseChatModel
    headroom_config=HeadroomConfig(),  # Headroom 配置
    auto_detect_provider=True,  # 从被包装模型自动探测
)

HeadroomChatMessageHistory

HeadroomChatMessageHistory(
    base_history,              # 任意 BaseChatMessageHistory
    compress_threshold_tokens=4000,  # 触发压缩的 token 阈值
    keep_recent_turns=5,       # 最少保留轮次
    model="gpt-4o",           # token 计数所用模型
)

（源码还支持 provider 参数注入自定义 Headroom provider，见 memory.py#L98-L117。）

HeadroomDocumentCompressor

HeadroomDocumentCompressor(
    max_documents=10,   # 最多返回文档数
    min_relevance=0.0,  # 最低相关分（0-1）
    prefer_diverse=False,  # 使用 MMR 求多样性
)

wrap_tools_with_headroom

wrap_tools_with_headroom(
    tools,                     # LangChain 工具列表
    min_chars_to_compress=1000,  # 最小输出长度（字符）
    metrics_collector=None,    # 共享指标收集器（缺省用全局实例）
)

导入参考

from headroom.integrations import (
    # Chat Model
    HeadroomChatModel,
    # Memory
    HeadroomChatMessageHistory,
    # Retrievers
    HeadroomDocumentCompressor,
    # Agents
    HeadroomToolWrapper,
    wrap_tools_with_headroom,
    get_tool_metrics,
    reset_tool_metrics,
    # Streaming
    StreamingMetricsTracker,
    StreamingMetricsCallback,
    track_streaming_response,
    # LangSmith
    HeadroomLangSmithCallbackHandler,
    # Provider Detection
    detect_provider,
    get_headroom_provider,
)

# 也可以直接从子包导入
from headroom.integrations.langchain import HeadroomChatModel
from headroom.integrations.langchain.memory import HeadroomChatMessageHistory

上述名称均在 headroom/integrations/init.py 的 __all__ 中确认导出；该文件同时说明 headroom.integrations 顶层还复用了 MCP/Agno/CrewAI/AutoGen 子包的同类组件（如 compress_tool_result），LangChain 的工具压缩与其共享同一压缩器实现。

故障排查

LangChain 未被检测到

from headroom.integrations import langchain_available

if not langchain_available():
    print("Install with: pip install headroom-ai[langchain]")

Provider 探测失败

# 强制指定特定 provider
from headroom.providers import AnthropicProvider

llm = HeadroomChatModel(
    ChatAnthropic(model="claude-3-5-sonnet-20241022"),
    auto_detect_provider=False,
)
llm._provider = AnthropicProvider()

从源码结构看，auto_detect_provider=False 时 pipeline 的惰性初始化会默认回落到 OpenAIProvider()（chat_model.py#L226-L235），所以手动指定 llm._provider 的做法应理解为先于 pipeline 首次构建完成的注入；更稳妥的验证方式是检查 get_headroom_provider 对被包装模型类路径的识别结果（headroom.providers 中 AnthropicProvider 的导出见 providers/init.py）。

记忆不触发压缩

确认消息 token 数确实超过阈值；调试阶段可以先调低阈值、缩短保留轮数：

history = HeadroomChatMessageHistory(
    base_history,
    compress_threshold_tokens=1000,  # 调低阈值
    keep_recent_turns=2,  # 减少保留轮数
)

注意：token 统计走 provider.get_token_counter(model) 逐消息计数（memory.py#L190-L204），消息过少或过短时不会达到默认 4000 阈值。

性能与实践建议

1. Agent 优先用工具包装——带工具的 agent 是压缩收益最大的场景；
2. 设置合理阈值——小对话不要压缩，min_chars_to_compress=1000 与 compress_threshold_tokens=4000 的默认值即为此设计；
3. RAG 开启多样性——prefer_diverse=True 的 MMR 选择能减少冗余文档、提升回答质量；
4. 用 LangSmith 持续监控——通过 callback handler 在追踪平台里逐 run 观察 headroom.tokens_saved；
5. 批量化相似请求——provider 前缀缓存在稳定前缀下命中率更高，HeadroomChatModel 的 provider 缓存优化正服务于这一目标。

延伸阅读

• 完整官方文档：wiki/langchain.md
• 可运行的对比演示：examples/langchain_demo/（含 run_comparison.py、show_compression.py、verify_errors_kept.py 等脚本，可对比压缩前后差异并验证错误信息保护）
• 压缩变换管线：headroom/transforms/
• 项目总览：README.md