
























在构建基于大语言模型的对话应用时,上下文窗口限制是一个绕不开的硬约束。随着对话轮次增加,历史消息累积的 Token 数可能轻松突破模型限制,导致请求失败或成本飙升。LangChain 提供了一套完善的聊天记录压缩机制来解决这个问题。本文将深入解析其核心原理、多种实现策略,并提供可直接运行的 Demo 代码。
LangChain 的 Chain 和 Agent 默认是无状态的,每次请求独立处理。为了让对话具备连续性,需要将历史消息作为上下文注入每次请求。但问题随之而来:
Token 爆炸:随着对话增长,历史记录越来越长,Token 消耗呈线性增长。第1轮可能仅需 ~50 tokens,到第100轮可能膨胀至 ~5000 tokens,开始超出大多数模型的上下文窗口。
成本与延迟:更长的上下文意味着更高的 API 调用成本和更长的处理延迟。
上下文稀释:过长的历史消息会使模型注意力分散,反而影响对当前问题的判断质量。
因此,必须引入压缩策略,在保留关键信息和控制 Token 消耗之间取得平衡。
LangChain 的聊天记录压缩本质上是对消息历史的“降维”处理。核心压缩机制主要有以下三种:
原理:只保留最近 K 轮对话,更早的内容直接丢弃。
第1轮 → 第2轮 → 第3轮 → 第4轮 → 第5轮 → 第6轮
↓ 窗口大小=3
第4轮 → 第5轮 → 第6轮
优点:Token 消耗可控,实现简单。
缺点:早期重要信息永久丢失。
原理:当消息数量或 Token 数超过阈值时,调用 LLM 对历史对话生成摘要,用摘要替代原始消息。
完整历史(10轮) → [LLM生成摘要] → 摘要(约200词)+ 最近4轮消息
优点:保留全局上下文精髓,适合长对话。
缺点:摘要生成有额外 LLM 调用成本,细节可能丢失。
原理:结合滑动窗口和摘要压缩——近期对话保留原文,远期对话压缩为摘要。
完整历史 → 超过阈值 → 前段压缩为摘要 + 后段保留原文
优点:兼顾近期精确性和远期连贯性,是生产环境最推荐的方案。
这是 LangChain 官方提供的开箱即用方案,结合了窗口和摘要两种机制。
from langchain.memory import ConversationSummaryBufferMemory
from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain
llm = 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="我前面提到的名字和职业是什么?") # 能从摘要中回忆
LangChain 提供了 trim_messages 工具,可以按 Token 数或消息数量精确裁剪。
from langchain_core.messages import trim_messages
from langchain_openai import ChatOpenAI
model = 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)
对于 Agent 场景,LangChain 提供了 summarizationMiddleware,支持多种触发条件和保留策略。
from langchain import create_agent
from langchain.agents.middleware import summarizationMiddleware
agent = 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%
]
在 LangGraph 中,可以构建带有摘要节点的状态图,在消息数超过阈值时自动执行摘要。
from langgraph.graph import StateGraph, MessagesState
from langchain_core.messages import SystemMessage, RemoveMessage
class State(MessagesState):
summary: str
def 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"
以下是一个包含多种压缩策略的完整示例:
import os
from langchain_openai import ChatOpenAI
from langchain.memory import (
ConversationBufferWindowMemory,
ConversationSummaryBufferMemory,
)
from langchain.chains import ConversationChain
from langchain_core.messages import trim_messages
os.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 自己决定何时压缩。
from deepagents import create_deep_agent
from deepagents.middleware.summarization import create_summarization_tool_middleware
agent = create_deep_agent(
model="openai:gpt-5.4",
middleware=[
create_summarization_tool_middleware(model, backend),
],
)
# Agent 可以主动调用 /compact 工具来压缩自己的上下文
这种模式将压缩决策权从开发者交给 Agent,实现了“该压缩时再压缩”的智能管理,避免了固定阈值策略可能导致的“在不合适时机压缩”的问题。
| 策略 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 滑动窗口 | 简单对话,对早期信息不敏感 | 实现简单,成本可控 | 早期信息永久丢失 |
| 摘要压缩 | 需要保留全局上下文 | 保持对话完整性 | 额外 LLM 调用成本 |
| 混合策略 | 生产环境推荐 | 平衡近期精度与远期全局 | 配置稍复杂 |
| 自治压缩 | 长周期 Agent 任务 | 智能触发,减少人工调参 | 目前是 Deep Agents 特有 |
选择哪种方案取决于你的业务需求:如果早期信息不重要,滑动窗口最经济;如果长对话需要保留全局脉络,混合摘要策略是最佳选择;如果构建复杂 Agent,可以考虑引入自治压缩能力。
此内容由惯性聚合(RSS阅读器)自动聚合整理,仅供阅读参考。 原文来自 — 版权归原作者所有。