langchain中的上下文压缩方案

chain_extract.pycontextual_compression.py在 LangChain 中，多轮对话的上下文压缩主要用于解决长对话场景下的 Token 超限问题，通过对历史对话内容进行总结、提取关键信息或截断，在保留核心上下文的同时减少 Token 消耗。以下从核心方案、具体实现代码及逻辑展开分析：

一、核心方案：对话历史总结（Summarization Middleware）

针对多轮对话的上下文压缩，LangChain 中最直接的实现是 SummarizationMiddleware（对话总结中间件）。该方案通过监控对话历史的 Token 数量，当接近预设阈值时，自动总结较早的对话内容，保留最近的消息，并确保 AI 与工具调用的消息对不被拆分，维持上下文连贯性。

具体代码位置

文件路径：langchain/libs/langchain_v1/langchain/agents/middleware/summarization.py

核心逻辑解析

SummarizationMiddleware 实现了 AgentMiddleware 接口，在 Agent 调用模型前（before_model 方法）处理对话消息，核心步骤如下：

 

1. Token 监控与触发条件

    

   计算当前对话历史的总 Token 数，若超过 max_tokens_before_summary 阈值，则触发压缩逻辑。
    

    

   def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:messages = state["messages"]self._ensure_message_ids(messages)  # 确保消息有唯一ID，用于后续替换total_tokens = self.token_counter(messages)if self.max_tokens_before_summary is not None and total_tokens < self.max_tokens_before_summary:return None  # 未达阈值，不压缩# 触发压缩...

    
    
2. 安全截断点选择

    

   为避免拆分 AI 消息与对应的工具调用消息（如 AIMessage 和 ToolMessage 成对出现），通过 _find_safe_cutoff 找到合适的截断索引，确保保留的消息中不包含拆分的消息对。
    

    

   def _find_safe_cutoff(self, messages: list[AnyMessage]) -> int:if len(messages) <= self.messages_to_keep:return 0  # 消息数量不足，无需截断target_cutoff = len(messages) - self.messages_to_keep  # 目标截断点（保留最近N条）# 从目标截断点向前搜索安全位置，避免拆分AI-工具消息对for i in range(target_cutoff, -1, -1):if self._is_safe_cutoff_point(messages, i):return ireturn 0

    
    
3. 对话总结生成

    

   将截断点之前的消息（待总结部分）传入 LLM 生成总结，并替换原历史消息，保留截断点之后的最近消息。
    

    

   def _create_summary(self, messages_to_summarize: list[AnyMessage]) -> str:if not messages_to_summarize:return "No previous conversation history."trimmed_messages = self._trim_messages_for_summary(messages_to_summarize)  # 确保总结输入不超限response = self.model.invoke(self.summary_prompt.format(messages=trimmed_messages))  # 生成总结return cast("str", response.content).strip()

    
    
4. 替换历史消息

    

   用总结结果替换旧消息，并保留最近的消息，通过 RemoveMessage 指令清除原历史，再插入新的总结和保留消息。
    

    

   return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES),  # 清除原历史*self._build_new_messages(summary),  # 插入总结*preserved_messages,  # 插入保留的最近消息]
   }

    
    

二、辅助方案：文档压缩（间接支持对话上下文）

在多轮对话中，若涉及外部文档检索（如 RAG 场景），对话上下文可能包含检索到的长文档，此时会用到文档压缩方案，间接减少上下文 Token 消耗。

1. 上下文压缩检索器（ContextualCompressionRetriever）

• 功能：对检索到的文档先压缩再返回，减少传入对话的文档长度。
• 代码位置：langchain/libs/langchain/langchain_classic/retrievers/contextual_compression.py
• 核心逻辑：
   
  包装基础检索器（base_retriever）和文档压缩器（base_compressor），检索后自动调用压缩器处理文档。
   

  class ContextualCompressionRetriever(BaseRetriever):base_compressor: BaseDocumentCompressor  # 文档压缩器base_retriever: RetrieverLike  # 基础检索器def _get_relevant_documents(self, query: str, **kwargs) -> list[Document]:docs = self.base_retriever.invoke(query,** kwargs)  # 检索文档if docs:compressed_docs = self.base_compressor.compress_documents(docs, query)  # 压缩文档return list(compressed_docs)return []

   
   

2. LLM 链提取器（LLMChainExtractor）

• 功能：通过 LLM 提取文档中与当前查询相关的内容，剔除无关信息。
• 代码位置：langchain/libs/langchain/langchain_classic/retrievers/document_compressors/chain_extract.py
• 核心逻辑：
   
  使用 LLM 链（llm_chain）根据查询从文档中提取相关片段，仅保留有用内容。
   

  class LLMChainExtractor(BaseDocumentCompressor):llm_chain: Runnable  # 用于提取的LLM链def compress_documents(self, documents: Sequence[Document], query: str, **kwargs) -> Sequence[Document]:compressed_docs = []for doc in documents:input_ = self.get_input(query, doc)  # 构造查询+文档输入output = self.llm_chain.invoke(input_,** kwargs)  # 提取相关内容if output:  # 仅保留非空结果compressed_docs.append(Document(page_content=output, metadata=doc.metadata))return compressed_docs

   
   

三、总结

1. 多轮对话历史压缩：核心依赖 SummarizationMiddleware（summarization.py），通过总结旧对话、保留新消息实现上下文压缩，适配长对话场景。
2. 文档相关上下文压缩：通过 ContextualCompressionRetriever（contextual_compression.py）和 LLMChainExtractor（chain_extract.py）处理检索文档，间接减少对话上下文长度。

 

这些方案共同解决了多轮对话中的 Token 限制问题，平衡了上下文完整性与系统性能。在 LangChain 中，多轮对话的上下文压缩主要用于解决长对话场景下的 Token 超限问题，通过对历史对话内容进行总结、提取关键信息或截断，在保留核心上下文的同时减少 Token 消耗。以下从核心方案、具体实现代码及逻辑展开分析：

一、核心方案：对话历史总结（Summarization Middleware）

针对多轮对话的上下文压缩，LangChain 中最直接的实现是 SummarizationMiddleware（对话总结中间件）。该方案通过监控对话历史的 Token 数量，当接近预设阈值时，自动总结较早的对话内容，保留最近的消息，并确保 AI 与工具调用的消息对不被拆分，维持上下文连贯性。

具体代码位置

文件路径：langchain/libs/langchain_v1/langchain/agents/middleware/summarization.py

核心逻辑解析

SummarizationMiddleware 实现了 AgentMiddleware 接口，在 Agent 调用模型前（before_model 方法）处理对话消息，核心步骤如下：

 

1. Token 监控与触发条件

    

   计算当前对话历史的总 Token 数，若超过 max_tokens_before_summary 阈值，则触发压缩逻辑。
    

   def before_model(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:messages = state["messages"]self._ensure_message_ids(messages)  # 确保消息有唯一ID，用于后续替换total_tokens = self.token_counter(messages)if self.max_tokens_before_summary is not None and total_tokens < self.max_tokens_before_summary:return None  # 未达阈值，不压缩# 触发压缩...

    
    
2. 安全截断点选择

    

   为避免拆分 AI 消息与对应的工具调用消息（如 AIMessage 和 ToolMessage 成对出现），通过 _find_safe_cutoff 找到合适的截断索引，确保保留的消息中不包含拆分的消息对。
    

    

   def _find_safe_cutoff(self, messages: list[AnyMessage]) -> int:if len(messages) <= self.messages_to_keep:return 0  # 消息数量不足，无需截断target_cutoff = len(messages) - self.messages_to_keep  # 目标截断点（保留最近N条）# 从目标截断点向前搜索安全位置，避免拆分AI-工具消息对for i in range(target_cutoff, -1, -1):if self._is_safe_cutoff_point(messages, i):return ireturn 0

    
    
3. 对话总结生成

    

   将截断点之前的消息（待总结部分）传入 LLM 生成总结，并替换原历史消息，保留截断点之后的最近消息。
    

   def _create_summary(self, messages_to_summarize: list[AnyMessage]) -> str:if not messages_to_summarize:return "No previous conversation history."trimmed_messages = self._trim_messages_for_summary(messages_to_summarize)  # 确保总结输入不超限response = self.model.invoke(self.summary_prompt.format(messages=trimmed_messages))  # 生成总结return cast("str", response.content).strip()

    
    
4. 替换历史消息

    

   用总结结果替换旧消息，并保留最近的消息，通过 RemoveMessage 指令清除原历史，再插入新的总结和保留消息。
    

   return {"messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES),  # 清除原历史*self._build_new_messages(summary),  # 插入总结*preserved_messages,  # 插入保留的最近消息]
   }

    
    

二、辅助方案：文档压缩（间接支持对话上下文）

在多轮对话中，若涉及外部文档检索（如 RAG 场景），对话上下文可能包含检索到的长文档，此时会用到文档压缩方案，间接减少上下文 Token 消耗。

1. 上下文压缩检索器（ContextualCompressionRetriever）

• 功能：对检索到的文档先压缩再返回，减少传入对话的文档长度。
• 代码位置：langchain/libs/langchain/langchain_classic/retrievers/contextual_compression.py
• 核心逻辑：
   
  包装基础检索器（base_retriever）和文档压缩器（base_compressor），检索后自动调用压缩器处理文档。
   

  class ContextualCompressionRetriever(BaseRetriever):base_compressor: BaseDocumentCompressor  # 文档压缩器base_retriever: RetrieverLike  # 基础检索器def _get_relevant_documents(self, query: str, **kwargs) -> list[Document]:docs = self.base_retriever.invoke(query,** kwargs)  # 检索文档if docs:compressed_docs = self.base_compressor.compress_documents(docs, query)  # 压缩文档return list(compressed_docs)return []

   
   

2. LLM 链提取器（LLMChainExtractor）

• 功能：通过 LLM 提取文档中与当前查询相关的内容，剔除无关信息。
• 代码位置：langchain/libs/langchain/langchain_classic/retrievers/document_compressors/chain_extract.py
• 核心逻辑：
   
  使用 LLM 链（llm_chain）根据查询从文档中提取相关片段，仅保留有用内容。
   

   

  class LLMChainExtractor(BaseDocumentCompressor):llm_chain: Runnable  # 用于提取的LLM链def compress_documents(self, documents: Sequence[Document], query: str, **kwargs) -> Sequence[Document]:compressed_docs = []for doc in documents:input_ = self.get_input(query, doc)  # 构造查询+文档输入output = self.llm_chain.invoke(input_,** kwargs)  # 提取相关内容if output:  # 仅保留非空结果compressed_docs.append(Document(page_content=output, metadata=doc.metadata))return compressed_docs

   
   

三、总结

1. 多轮对话历史压缩：核心依赖 SummarizationMiddleware（summarization.py），通过总结旧对话、保留新消息实现上下文压缩，适配长对话场景。
2. 文档相关上下文压缩：通过 ContextualCompressionRetriever（contextual_compression.py）和 LLMChainExtractor（chain_extract.py）处理检索文档，间接减少对话上下文长度。

 

这些方案共同解决了多轮对话中的 Token 限制问题，平衡了上下文完整性与系统性能。