LangChain 聊天记录压缩：原理、机制与实战 - 若

LangChain 聊天记录压缩：原理、机制与实战

在构建基于大语言模型的对话应用时，上下文窗口限制是一个绕不开的硬约束。随着对话轮次增加，历史消息累积的 Token 数可能轻松突破模型限制，导致请求失败或成本飙升。LangChain 提供了一套完善的聊天记录压缩机制来解决这个问题。本文将深入解析其核心原理、多种实现策略，并提供可直接运行的 Demo 代码。

---

一、为什么要压缩聊天记录？

LangChain 的 Chain 和 Agent 默认是无状态的，每次请求独立处理。为了让对话具备连续性，需要将历史消息作为上下文注入每次请求。但问题随之而来：

Token 爆炸：随着对话增长，历史记录越来越长，Token 消耗呈线性增长。第1轮可能仅需 ~50 tokens，到第100轮可能膨胀至 ~5000 tokens，开始超出大多数模型的上下文窗口。

成本与延迟：更长的上下文意味着更高的 API 调用成本和更长的处理延迟。

上下文稀释：过长的历史消息会使模型注意力分散，反而影响对当前问题的判断质量。

因此，必须引入压缩策略，在保留关键信息和控制 Token 消耗之间取得平衡。

---

二、核心压缩机制

LangChain 的聊天记录压缩本质上是对消息历史的“降维”处理。核心压缩机制主要有以下三种：

2.1 滑动窗口（Buffer Window）

原理：只保留最近 K 轮对话，更早的内容直接丢弃。

第1轮 → 第2轮 → 第3轮 → 第4轮 → 第5轮 → 第6轮↓ 窗口大小=3第4轮 → 第5轮 → 第6轮

优点：Token 消耗可控，实现简单。
缺点：早期重要信息永久丢失。

2.2 摘要压缩（Summarization）

原理：当消息数量或 Token 数超过阈值时，调用 LLM 对历史对话生成摘要，用摘要替代原始消息。

完整历史（10轮） → [LLM生成摘要] → 摘要（约200词）+ 最近4轮消息

优点：保留全局上下文精髓，适合长对话。
缺点：摘要生成有额外 LLM 调用成本，细节可能丢失。

2.3 混合策略（Summary Buffer）

原理：结合滑动窗口和摘要压缩——近期对话保留原文，远期对话压缩为摘要。

完整历史 → 超过阈值 → 前段压缩为摘要 + 后段保留原文

优点：兼顾近期精确性和远期连贯性，是生产环境最推荐的方案。

---

三、LangChain 具体实现方式

3.1 方式一：ConversationSummaryBufferMemory（混合策略）

这是 LangChain 官方提供的开箱即用方案，结合了窗口和摘要两种机制。

from langchain.memory import ConversationSummaryBufferMemory
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChainllm = ChatOpenAI(model="gpt-4o", temperature=0)# 当总 Token 超过 2000 时触发摘要压缩
# 保留最近消息的原文，更早的压缩为摘要
memory = ConversationSummaryBufferMemory(llm=llm,max_token_limit=2000,return_messages=True
)conversation = ConversationChain(llm=llm,memory=memory,verbose=True
)# 多轮对话测试
conversation.predict(input="我叫张三，是一名 Python 工程师")
conversation.predict(input="我正在学习 LangChain 的记忆机制")
conversation.predict(input="我前面提到的名字和职业是什么？")  # 能从摘要中回忆

3.2 方式二：trim_messages（消息裁剪工具）

LangChain 提供了 trim_messages 工具，可以按 Token 数或消息数量精确裁剪。

from langchain_core.messages import trim_messages
from langchain_openai import ChatOpenAImodel = ChatOpenAI(model="gpt-4o")# 修剪消息：保留最后 10 条消息，确保以 HumanMessage 开头（符合模型要求）
trimmer = trim_messages(max_tokens=4000,strategy="last",           # 保留最近的include_system=True,       # 始终保留系统消息start_on="human",          # 从 HumanMessage 开始
)# 在调用模型前应用修剪
messages = state.get("messages", [])
trimmed_messages = trimmer.invoke(messages)
response = model.invoke(trimmed_messages)

3.3 方式三：Summarization Middleware（Deep Agents 方案）

对于 Agent 场景，LangChain 提供了 summarizationMiddleware，支持多种触发条件和保留策略。

from langchain import create_agent
from langchain.agents.middleware import summarizationMiddlewareagent = create_agent(model="openai:gpt-5.4",tools=[weather_tool, calculator_tool],middleware=[summarizationMiddleware(model="openai:gpt-5.4-mini",           # 用于生成摘要的模型trigger={                              # 触发条件（AND 逻辑）"tokens": 4000,                    # 超过 4000 token 触发"messages": 10                     # 且消息数超过 10 条},keep={"messages": 20},                 # 保留最近 20 条消息),],
)# Agent 将自动在上下文接近阈值时执行压缩

高级配置：支持多触发条件（OR 逻辑）和分数比例触发。

# 多条件触发：任一满足即触发
trigger=[{"tokens": 3000, "messages": 6},{"fraction": 0.8}   # 或达到上下文窗口的 80%
]

3.4 方式四：LangGraph 自定义摘要节点

在 LangGraph 中，可以构建带有摘要节点的状态图，在消息数超过阈值时自动执行摘要。

from langgraph.graph import StateGraph, MessagesState
from langchain_core.messages import SystemMessage, RemoveMessageclass State(MessagesState):summary: strdef summarize_conversation(state: State):summary = state.get("summary", "")# 构建摘要提示词if summary:prompt = f"现有摘要：{summary}\n请基于最新对话扩展摘要："else:prompt = "请为以下对话生成摘要："messages = state["messages"] + [HumanMessage(content=prompt)]response = model.invoke(messages)# 删除除最近 2 条外的所有消息delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]return {"summary": response.content, "messages": delete_messages}def should_summarize(state: State):# 消息数 > 6 时触发摘要return "summarize" if len(state["messages"]) > 6 else "end"

---

四、完整 Demo 代码

以下是一个包含多种压缩策略的完整示例：

import os
from langchain_openai import ChatOpenAI
from langchain.memory import (ConversationBufferWindowMemory,ConversationSummaryBufferMemory,
)
from langchain.chains import ConversationChain
from langchain_core.messages import trim_messagesos.environ["OPENAI_API_KEY"] = "your-api-key"
llm = ChatOpenAI(model="gpt-4o", temperature=0)# ========== 示例1：滑动窗口 ==========
print("=== 滑动窗口（仅保留最近3轮）===")
window_memory = ConversationBufferWindowMemory(k=3)
chain1 = ConversationChain(llm=llm, memory=window_memory, verbose=True)
chain1.predict(input="我叫张三，我是工程师")
chain1.predict(input="我喜欢 Python")
chain1.predict(input="你记得我的名字和职业吗？")
# 此时第1轮"我叫张三"已被丢弃# ========== 示例2：摘要混合（推荐） ==========
print("\n=== 摘要混合策略 ===")
summary_memory = ConversationSummaryBufferMemory(llm=llm,max_token_limit=500,   # 低阈值便于演示触发return_messages=True,
)
chain2 = ConversationChain(llm=llm, memory=summary_memory, verbose=True)for i in range(15):chain2.predict(input=f"这是第{i+1}轮对话，我的用户ID是{1000+i}")print(f"第{i+1}轮后的内存状态，摘要已生效")# ========== 示例3：trim_messages ==========
print("\n=== trim_messages 精确裁剪 ===")
trimmer = trim_messages(max_tokens=300,strategy="last",include_system=True,start_on="human",
)# 模拟多个历史消息
from langchain_core.messages import HumanMessage, AIMessage
messages = [HumanMessage(content=f"这是第{i}条消息") for i in range(20)
]trimmed = trimmer.invoke(messages)
print(f"原始消息数: {len(messages)}, 裁剪后: {len(trimmed)}")

---

五、进阶：自治上下文压缩

LangChain 最新推出的 Deep Agents 实现了更智能的压缩策略：不仅可以在 Token 阈值触发时自动压缩，还允许 Agent 自己决定何时压缩。

核心触发时机

1. 任务边界处：用户表示开始新任务，旧上下文不再相关
2. 提取大量上下文后：已完成研究任务，获得结论
3. 消费大量新上下文前：即将生成长文本或读取大文件
4. 复杂多步流程前：即将开始大型重构、多文件编辑
5. 新决策覆盖旧上下文：新需求使旧信息失效

from deepagents import create_deep_agent
from deepagents.middleware.summarization import create_summarization_tool_middlewareagent = create_deep_agent(model="openai:gpt-5.4",middleware=[create_summarization_tool_middleware(model, backend),],
)
# Agent 可以主动调用 /compact 工具来压缩自己的上下文

这种模式将压缩决策权从开发者交给 Agent，实现了“该压缩时再压缩”的智能管理，避免了固定阈值策略可能导致的“在不合适时机压缩”的问题。

---

六、总结

策略	适用场景	优点	缺点
滑动窗口	简单对话，对早期信息不敏感	实现简单，成本可控	早期信息永久丢失
摘要压缩	需要保留全局上下文	保持对话完整性	额外 LLM 调用成本
混合策略	生产环境推荐	平衡近期精度与远期全局	配置稍复杂
自治压缩	长周期 Agent 任务	智能触发，减少人工调参	目前是 Deep Agents 特有

选择哪种方案取决于你的业务需求：如果早期信息不重要，滑动窗口最经济；如果长对话需要保留全局脉络，混合摘要策略是最佳选择；如果构建复杂 Agent，可以考虑引入自治压缩能力。