惯性聚合 高效追踪和阅读你感兴趣的博客、新闻、科技资讯
阅读原文 在惯性聚合中打开

推荐订阅源

V
V2EX
爱范儿
爱范儿
Martin Fowler
Martin Fowler
T
The Blog of Author Tim Ferriss
B
Blog RSS Feed
博客园 - 聂微东
G
GRAHAM CLULEY
Engineering at Meta
Engineering at Meta
Threat Intelligence Blog | Flashpoint
Threat Intelligence Blog | Flashpoint
WordPress大学
WordPress大学
Scott Helme
Scott Helme
AI
AI
S
Security Affairs
T
Threat Research - Cisco Blogs
M
MIT News - Artificial intelligence
T
Troy Hunt's Blog
H
Hackread – Cybersecurity News, Data Breaches, AI and More
人人都是产品经理
人人都是产品经理
AWS News Blog
AWS News Blog
T
Threatpost
Cyberwarzone
Cyberwarzone
www.infosecurity-magazine.com
www.infosecurity-magazine.com
U
Unit 42
V
Vulnerabilities – Threatpost
J
Java Code Geeks
博客园 - Franky
月光博客
月光博客
Blog — PlanetScale
Blog — PlanetScale
NISL@THU
NISL@THU
D
Docker
小众软件
小众软件
N
News and Events Feed by Topic
Microsoft Security Blog
Microsoft Security Blog
Y
Y Combinator Blog
A
Arctic Wolf
D
DataBreaches.Net
云风的 BLOG
云风的 BLOG
Forbes - Security
Forbes - Security
量子位
PCI Perspectives
PCI Perspectives
美团技术团队
K
KPMG report finds enterprise disconnect between AI and its ROI | CIO
I
InfoQ
Security Archives - TechRepublic
Security Archives - TechRepublic
有赞技术团队
有赞技术团队
腾讯CDC
P
Proofpoint News Feed
S
Security @ Cisco Blogs
G
Google Developers Blog
C
Cisco Blogs

博客园 - 若-飞

企业AI Agent落地的核心逻辑与路径 基于langchain,Function Call的成功率怎么解决? LangChain Checkpoint(检查点)是什么?—— Agent 的"存档机制" RAG 设计:Embedding 如何切分 AI 客服系统设计:RAG 知识库设计 Go 百万连接服务器设计:从网卡到业务的全链路解析 深度解析 sync.Pool:从设计哲学到生产实践 Goroutine 泄漏:原因、检测与防范 Go Channel 关闭与超时机制完全指南 Go Map 无限增长问题解决方案 揭开 sklearn 文本分类的核心原理:从词袋到逻辑回归 大模型“胡说八道”怎么办?一张图读懂检测、评估与修复全方案 企业级AI知识库权限隔离设计:让AI“懂规矩”比“懂知识”更重要 RAG系统设计全解析:从架构到多模态的核心知识图谱 RAG召回率提升全攻略:7大核心方法让检索更精准 RAG召回率提升秘籍:Metadata过滤的底层原理与实践 构建更好的RAG系统:深入理解混合搜索 一文搞懂 RAG 中 Retriever 和 Reranker 的区别 一文搞懂 RAG 的召回率(Recall)是什么? LangChain / LangGraph、MCP、Harness Engineer 与 Claude Code 的对应关系 Agent Harness 技术笔记:从 Trajectory 到 Function Calling Loop BLEU 是什么?——从原理到工程实践 一文讲清:Approve / Permit / Permit2 的本质区别 分库分表后跨分页查询的完整方案 ai如何处理私有数据 ai幻觉是啥,以及如何解决 别再让大模型“凭空瞎猜”了!带你认识AI最强外挂:ChromaDB 用 useQuery 管请求:TanStack React Query 入门小结 HD钱包--BIP44 TRON 四种 API 面怎么选:从节点协议到 JSON-RPC 再到 TronGrid 以太坊节点存储与共识机制全解析 BSC节点发现协议全解析:UDP发现、Bootnode引导与Gossip交易广播 以太坊节点发现背后的分布式哈希表(DHT)与 Kademlia 原理解析 Solidity中的bytes与string:深入理解这两种特殊的动态数组 智能合约自毁:当资产还在,合约死了 —— 深度解析 selfdestruct 导致的资产锁定风险 TDengine CLI (taos) 使用指南 —— Docker 本地开发实战 在 macOS 上用 DBeaver 连接 TDengine:踩坑总结与最终配置指南 Solidity Storage Slot 深度解析 Geth Snapshot Export/Import 深度解析: 不是备份工具,而是数据分析利器 基于BSC 公链的数据备份与 Snapshot 机制深度解析 Docker 共享内存完全指南:从原理到实践,避免常见的理解误区 Docker容器"僵尸状态"问题排查与自动重启方案 SSE协议深度解析:被低估的HTTP服务器推送标准 TDengine vs MySQL:时序数据处理的时代之选 Proxmox 启用 QEMU Guest Agent 实战指南 解决 Blockscout "batch too large" 错误的完整指南 一文讲清楚什么是基准测试(Benchmark) Rust中的宏(Macro):编译时的代码生成魔法 Docker优雅关闭的艺术:为什么stop_grace_period能防止数据丢失 为什么 Go 没有依赖注入和 Bean 机制?语言设计哲学对比
LangChain 聊天记录压缩:原理、机制与实战
若-飞 · 2026-07-10 · via 博客园 - 若-飞

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 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="我前面提到的名字和职业是什么?")  # 能从摘要中回忆

3.2 方式二:trim_messages(消息裁剪工具)

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)

3.3 方式三:Summarization Middleware(Deep Agents 方案)

对于 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%
]

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

在 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"

四、完整 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_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 自己决定何时压缩

核心触发时机

  1. 任务边界处:用户表示开始新任务,旧上下文不再相关
  2. 提取大量上下文后:已完成研究任务,获得结论
  3. 消费大量新上下文前:即将生成长文本或读取大文件
  4. 复杂多步流程前:即将开始大型重构、多文件编辑
  5. 新决策覆盖旧上下文:新需求使旧信息失效
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,可以考虑引入自治压缩能力。