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 方法）处理对话消息，核心步骤如下：-
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 # 未达阈值，不压缩# 触发压缩... -
安全截断点选择为避免拆分 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 -
对话总结生成将截断点之前的消息（待总结部分）传入 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() -
替换历史消息用总结结果替换旧消息，并保留最近的消息，通过
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
三、总结
- 多轮对话历史压缩：核心依赖
SummarizationMiddleware（summarization.py），通过总结旧对话、保留新消息实现上下文压缩，适配长对话场景。 - 文档相关上下文压缩：通过
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 方法）处理对话消息，核心步骤如下：-
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 # 未达阈值，不压缩# 触发压缩... -
安全截断点选择为避免拆分 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 -
对话总结生成将截断点之前的消息（待总结部分）传入 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() -
替换历史消息用总结结果替换旧消息，并保留最近的消息，通过
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
三、总结
- 多轮对话历史压缩：核心依赖
SummarizationMiddleware（summarization.py），通过总结旧对话、保留新消息实现上下文压缩，适配长对话场景。 - 文档相关上下文压缩：通过
ContextualCompressionRetriever（contextual_compression.py）和LLMChainExtractor（chain_extract.py）处理检索文档，间接减少对话上下文长度。
这些方案共同解决了多轮对话中的 Token 限制问题，平衡了上下文完整性与系统性能。