# LangChain 缓冲记忆实践:从 ConversationBufferMemory 到新版短期记忆
缓冲记忆解决的是一个很朴素的问题:用户和模型已经聊过几轮了,下一轮调用模型时,要不要把前面的内容带上?如果要带,带多少?按轮数带,按 token 带,还是把旧内容压缩成摘要?
早期 LangChain 里,这类能力通常由 ConversationBufferMemory、ConversationBufferWindowMemory、ConversationTokenBufferMemory 这些组件完成。它们很适合理解“聊天上下文怎么进入 Prompt”,但如果今天从零写生产应用,就不应该再把这些类当成主线设计。
当前 LangChain 的推荐路径已经变成:
- Agent 场景:使用
create_agent、checkpointer和thread_id管理短期记忆。 - 长对话裁剪:使用 middleware 在模型调用前裁剪、删除或摘要消息。
- 生产持久化:使用数据库 backed checkpointer,例如 Postgres。
- 旧项目维护:
ConversationBufferMemory一类 API 归到langchain-classic,适合兼容,不适合作为新架构起点。
所以这篇不只讲“缓冲记忆怎么用”,更重要的是讲清楚:旧版缓冲记忆的设计思想如何迁移到当前 LangChain 的实现方式里。
# 缓冲记忆到底在缓冲什么
LLM 每次调用本身是无状态的。所谓“记忆”,本质上是应用在模型调用前后做了两件事:
用户输入
-> 读取当前 thread 的历史消息
-> 选择要进入上下文的部分
-> 调用模型
-> 保存本轮用户消息、模型回复、工具结果
-> 下一轮继续读取
2
3
4
5
6
缓冲记忆关注的是其中的“选择要进入上下文的部分”。
如果把所有历史都带上,就是全量缓冲。如果只带最近几轮,就是窗口缓冲。如果按 token 上限截断,就是 token 缓冲。如果把旧消息总结成摘要,再拼接最近消息,就是摘要缓冲。
这些策略不是 LangChain 独有,而是所有聊天机器人都绕不开的上下文工程问题。

从流程上看,四类缓冲记忆的共同点都是:用户提问前,把记忆模块里的聊天历史填充到模型上下文;模型生成回答后,再把新的交互写回记忆模块。差异在于历史保留策略:全量缓冲不处理历史,窗口缓冲按轮数截断,token 缓冲按预算淘汰旧消息,字符串缓冲则把历史降级成普通文本。
# 旧版缓冲记忆的四种形态
早期 LangChain 常见的缓冲记忆可以分成四类。
第一类是 ConversationBufferMemory。
它会把对话历史完整保存下来,并在下一次调用时全部注入 Prompt。优点是简单,缺点也很明显:对话越长,token 成本越高,响应越慢,旧信息越容易干扰模型。
第二类是 ConversationBufferWindowMemory。
它只保留最近 k 轮对话。这里的 k 通常指对话轮数,一轮包含用户消息和模型消息,因此最后进入上下文的大致是 2 * k 条消息。它适合客服、助手、短任务编排这类“近期上下文最重要”的场景。
第三类是 ConversationTokenBufferMemory。
它不按轮数截断,而是按 token 预算截断。相比固定窗口,它更贴近真实成本,因为有些消息很短,有些消息可能是一大段日志、代码或文档片段。
第四类是 ConversationStringBufferMemory。
它把历史保存成字符串,而不是结构化 message。这个方式现在不太推荐作为新项目选择,因为 chat model 更适合接收带角色的消息列表。字符串历史会丢失 human、ai、tool 这些角色边界,工具调用和多模态消息也更难表达。
用今天的眼光看,这四类组件真正留下来的价值不是类名本身,而是四种策略:
| 旧版组件 | 策略 | 当前更推荐的实现 |
|---|---|---|
ConversationBufferMemory | 全量历史 | checkpointer 保存 thread state |
ConversationBufferWindowMemory | 最近 N 轮 | before_model middleware 裁剪 messages |
ConversationTokenBufferMemory | token 预算 | trim_messages 或摘要 middleware |
ConversationStringBufferMemory | 字符串历史 | 尽量保留结构化 messages |
# 旧代码还能怎么写
如果维护老项目,可以继续使用 classic 包里的缓冲记忆组件。关键点是:不要再从 langchain.memory 里假设一切都可用,v1 之后很多旧能力都迁到了 langchain-classic。
安装:
pip install langchain-classic
窗口缓冲记忆的旧式写法大概是这样:
from operator import itemgetter
from langchain_classic.memory import ConversationBufferWindowMemory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_openai import ChatOpenAI
memory = ConversationBufferWindowMemory(
input_key="query",
memory_key="history",
return_messages=True,
k=2,
)
prompt = ChatPromptTemplate.from_messages(
[
("system", "你是一个严谨的 Python 后端助手。"),
MessagesPlaceholder("history"),
("human", "{query}"),
]
)
llm = ChatOpenAI(model="gpt-4.1-mini")
chain = (
RunnablePassthrough.assign(
history=RunnableLambda(memory.load_memory_variables) | itemgetter("history")
)
| prompt
| llm
| StrOutputParser()
)
query = "我叫 Alice,正在做 Flask 项目。"
chain_input = {"query": query}
answer = chain.invoke(chain_input)
memory.save_context(chain_input, {"output": answer})
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
这段代码能跑,也能帮助理解 Memory 的运行方式:调用前 load_memory_variables 读取历史,调用后 save_context 保存本轮输入输出。
但生产里它有几个问题:
- 记忆对象通常在进程内,服务重启后容易丢。
- 多用户、多会话隔离要自己处理。
- 工具调用、中间状态、流式事件和 Agent 状态不容易统一保存。
- 长对话治理能力偏弱,通常要自己再包一层。
因此新项目更建议把“记忆”放进 Agent state 和 checkpointer 里。
# 新版实现:Agent 短期记忆
当前 LangChain 官方短期记忆文档推荐在创建 agent 时传入 checkpointer。短期记忆会作为 agent state 的一部分保存,其中最核心的字段是 messages。每次调用时通过 thread_id 区分会话。
本地开发可以使用 InMemorySaver:
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
def get_user_profile() -> str:
"""读取当前用户资料。"""
return "暂时没有用户资料。"
agent = create_agent(
model="openai:gpt-5.5",
tools=[get_user_profile],
system_prompt="你是一个严谨的 Python 后端助手。",
checkpointer=InMemorySaver(),
)
config = {"configurable": {"thread_id": "user-1:chat-1001"}}
first = agent.invoke(
{"messages": [{"role": "user", "content": "我叫 Alice,正在做 Flask 项目。"}]},
config,
)
second = agent.invoke(
{"messages": [{"role": "user", "content": "我刚才说我在做什么?"}]},
config,
)
print(second["messages"][-1].content)
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
这里没有显式调用 save_context。Agent 每一步执行时会把 state 交给 checkpointer 保存;下一次相同 thread_id 调用时,再从 checkpoint 恢复。
这就是新版和旧版 Memory 最大的差异:
- 旧版 Memory 更像 chain 外挂的历史变量注入器。
- 新版 checkpointer 是 Agent 执行图的状态持久化能力。
# 生产实现:使用数据库 Checkpointer
InMemorySaver 只适合开发和测试。生产环境必须考虑服务重启、横向扩容、多实例并发和会话恢复,因此要使用数据库 backed checkpointer。
以 Postgres 为例:
pip install langgraph-checkpoint-postgres
应用启动时创建 checkpointer:
from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver
DB_URI = "postgresql://postgres:postgres@localhost:5432/postgres?sslmode=disable"
def get_user_profile() -> str:
"""读取当前用户资料。"""
return "暂时没有用户资料。"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup()
agent = create_agent(
model="openai:gpt-5.5",
tools=[get_user_profile],
system_prompt="你是一个严谨的 Python 后端助手。",
checkpointer=checkpointer,
)
result = agent.invoke(
{"messages": [{"role": "user", "content": "记住,我的项目用 Flask。"}]},
{"configurable": {"thread_id": "user-1:chat-1001"}},
)
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
生产里不要把 thread_id 写成简单的 "1"。更稳妥的方式是把业务身份编码进去:
{tenant_id}:{user_id}:{conversation_id}
这样可以避免不同用户、不同租户、不同会话之间的状态串线。
还要注意 checkpointer.setup() 的执行时机。它会创建所需表结构,适合在部署初始化或应用启动阶段执行。严肃生产环境里,表结构变更最好纳入数据库迁移流程,而不是每个请求里临时执行。
# 新版窗口缓冲:只保留最近几轮
旧版 ConversationBufferWindowMemory(k=2) 的含义是:只把最近两轮对话交给模型。新版里可以用 before_model middleware 做同样的事。
from typing import Any
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime
@before_model
def keep_recent_messages(
state: AgentState,
runtime: Runtime,
) -> dict[str, Any] | None:
messages = state["messages"]
if len(messages) <= 6:
return None
first_message = messages[0]
recent_messages = messages[-6:]
return {
"messages": [
RemoveMessage(id=REMOVE_ALL_MESSAGES),
first_message,
*recent_messages,
]
}
agent = create_agent(
model="openai:gpt-5.5",
tools=[],
system_prompt="你是一个简洁、可靠的技术助手。",
middleware=[keep_recent_messages],
checkpointer=InMemorySaver(),
)
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
这段代码的效果接近窗口记忆:保留最开始的关键消息,再保留最近若干条消息,中间旧消息从本次模型上下文中移除。
需要注意一个细节:裁剪消息时不能破坏消息合法性。很多模型要求消息序列满足这些约束:
- 工具调用消息后面必须跟对应的工具结果。
- 有些 provider 对第一条用户消息、系统消息位置有要求。
- 不能只保留一条孤立的 tool message。
所以生产里不要简单粗暴地 messages[-N:]。如果你的 Agent 会调用工具,裁剪策略必须按完整的 user -> assistant -> tool -> assistant 片段处理。
# 新版 Token 缓冲:按预算裁剪或摘要
窗口缓冲的缺点是“不看内容长度”。两轮短对话可能只有几十 token,两轮代码分析可能几千 token。生产里更常见的是按 token 预算处理。
LangChain 短期记忆文档给出的方向是:长对话接近上下文窗口时,可以在模型调用前 trim messages、delete messages,或者 summarize messages。
如果只是做硬裁剪,可以把裁剪逻辑放进 before_model:
from typing import Any
from langchain.agents import AgentState
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage, trim_messages
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime
@before_model
def trim_by_token_budget(
state: AgentState,
runtime: Runtime,
) -> dict[str, Any] | None:
messages = state["messages"]
trimmed = trim_messages(
messages,
strategy="last",
token_counter=len,
max_tokens=12,
include_system=True,
allow_partial=False,
)
if len(trimmed) == len(messages):
return None
return {
"messages": [
RemoveMessage(id=REMOVE_ALL_MESSAGES),
*trimmed,
]
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
这里的 token_counter=len 只是为了展示接口形态,生产环境应该使用真实模型 tokenizer 或模型对象做 token 估算,否则预算不准。
如果不想丢掉太多早期信息,更稳妥的方式是摘要。LangChain v1 迁移文档里已经把 summarization 作为 pre-model middleware 的典型场景,并提供了 SummarizationMiddleware:
from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver
agent = create_agent(
model="openai:gpt-5.5",
tools=[],
system_prompt="你是一个严谨的 Python 后端助手。",
middleware=[
SummarizationMiddleware(
model="openai:gpt-5.5",
trigger=("tokens", 4000),
)
],
checkpointer=InMemorySaver(),
)
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
摘要策略比单纯裁剪更适合长任务,例如需求讨论、代码审查、排障分析。它可以把早期上下文压成更短的状态,再保留最近消息作为细节补充。
# 缓冲记忆的生产边界
缓冲记忆不是越多越好。真正上线时,至少要把下面几个边界想清楚。
第一,缓冲记忆只解决会话内连续性,不等于长期记忆。
用户在当前会话里说“我正在做 Flask 项目”,这可以进入 thread state。用户长期偏好“回答时优先给生产实践”,更适合写入长期 profile store。两者生命周期不同,不要混在一个 messages 列表里。
第二,历史消息不一定都可信。
用户可能纠正自己,模型也可能答错。如果把错误内容长期保留,每一轮都带给模型,模型会被旧错误牵着走。生产里应该有覆盖、删除、冲突解决机制。
第三,缓冲内容需要脱敏。
日志、token、手机号、身份证、数据库连接串、内部系统地址,都可能出现在聊天历史里。如果这些内容被永久 checkpoint,后续治理成本会很高。敏感内容最好在入库前做脱敏或分类存储。
第四,缓冲策略要可观测。
排查“模型为什么这样回答”时,需要知道本轮模型到底看到了哪些历史消息、哪些消息被裁剪、哪些内容被摘要。否则线上问题会很难复现。
第五,清空记忆必须是业务能力。
用户退出登录、删除会话、撤销授权、管理员清理数据时,都可能要求删除历史。不要只实现追加,不实现删除。
# 迁移建议
如果现在还有旧版 ConversationBufferMemory 代码,可以按下面顺序迁移。
第一步,把全局 memory 对象改成按 thread_id 隔离。
很多老代码会在模块级别创建一个 memory = ConversationBufferMemory()。这在单用户 demo 里没问题,但在 Web 服务里会导致所有用户共享历史。先把会话隔离补上。
第二步,把 chain 外挂 memory 改成 message state。
如果是 Agent,优先迁到 create_agent + checkpointer。如果是纯 LCEL 链,可以继续使用 RunnableWithMessageHistory 或自己管理 BaseChatMessageHistory,但要明确它只是消息历史,不是完整 Agent 状态。
第三步,把“保存全部历史”改成“保存状态,按需裁剪上下文”。
存储层可以保存完整 state,模型调用前通过 middleware 决定本次带哪些消息。不要把“数据库保存多少”和“Prompt 里放多少”混为一谈。
第四步,为长对话加摘要或 token 预算。
窗口策略适合短任务,长任务更需要摘要。尤其是代码分析、需求评审、知识库问答,一条消息就可能很长,固定轮数并不可靠。
第五步,接入可观测性。
至少记录 thread_id、模型输入 messages 数量、裁剪前后 token 估算、摘要触发次数、checkpoint 写入状态。这样线上成本、延迟和回答质量才能被持续优化。
# 问题
缓冲记忆的问题,是简单但粗糙。
全量缓冲会让历史无限增长;窗口缓冲会裁掉早期关键约束;token 缓冲能控制预算,但可能破坏工具调用消息结构;字符串缓冲会丢失 message role,让 chat model 难以区分用户、模型、工具和系统规则。
生产里更麻烦的是:缓冲记忆很容易被当成“模型记住了”。实际上模型没有记忆,只是应用把历史重新放进上下文。只要 thread_id 错了、历史没写入、裁剪过头、摘要漏了,模型就会“忘记”。
缓冲记忆还会放大隐私风险。用户说过的密钥、手机号、内部地址、错误配置,如果长期保存在 checkpoint 或消息表里,又反复进入模型上下文,会造成治理压力。
# 拓展
缓冲记忆可以拓展成多层上下文策略。
第一层是最近消息,保留当前对话的连续性。
第二层是会话摘要,把旧消息压缩成任务背景。
第三层是长期 Store,保存用户偏好、项目事实、组织规则。
第四层是 RAG 或向量检索,召回外部知识和相似历史事件。
一个更完整的上下文组合是:
system prompt
+ 用户长期偏好
+ 当前项目事实
+ 较早对话摘要
+ 最近 N 条消息
+ RAG 检索内容
+ 当前问题
2
3
4
5
6
7
这样做比单纯 messages[-N:] 稳得多。
# 实际生产是否使用
会使用,但不会只用它。
缓冲记忆是所有聊天系统的基础能力,生产里一定需要最近消息、窗口裁剪、token 预算。但只靠缓冲记忆不够,尤其是长任务、工具调用、RAG、跨会话偏好场景。
实际生产通常会这样分工:
- 短会话:最近消息 + token 裁剪即可。
- 长会话:最近消息 + 摘要。
- Agent:checkpointer 保存 thread state。
- 跨会话:Store 保存长期偏好和事实。
- 高敏场景:默认少存或脱敏存储。
所以结论是:缓冲策略会用,旧版缓冲组件不会直接作为生产中心。
# 现在是否抛弃
旧版 ConversationBufferMemory、ConversationBufferWindowMemory、ConversationTokenBufferMemory 没有完全消失,但已经属于旧 Memory 体系,更多用于兼容和理解原理。
当前 LangChain 新项目更推荐:
- 用 checkpointer 保存短期 thread state。
- 用
before_modelmiddleware 做裁剪。 - 用
trim_messages做 token 预算控制。 - 用
SummarizationMiddleware处理长对话压缩。
所以不是“缓冲记忆被抛弃”,而是“旧版缓冲类不再是推荐主线”。
# 最新生产如何实现
最新版生产实现应以 checkpointer 为短期记忆主线。
from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup()
agent = create_agent(
model="gpt-5.5",
tools=[],
checkpointer=checkpointer,
)
agent.invoke(
{"messages": [{"role": "user", "content": "我叫 Alice。"}]},
{"configurable": {"thread_id": "tenant-1:user-42:chat-1001"}},
)
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
窗口裁剪放到 middleware:
from typing import Any
from langchain.agents import AgentState
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime
@before_model
def keep_recent_messages(
state: AgentState,
runtime: Runtime,
) -> dict[str, Any] | None:
messages = state["messages"]
if len(messages) <= 12:
return None
return {
"messages": [
RemoveMessage(id=REMOVE_ALL_MESSAGES),
messages[0],
*messages[-12:],
]
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
生产里还要补上:真实 token counter、工具消息完整性校验、敏感信息过滤、trace metadata、删除和过期策略。
# 总结
缓冲记忆的核心思想没有过时:模型无状态,所以应用必须在调用前提供必要历史,在调用后保存新的上下文。
过时的是把 ConversationBufferMemory 这类旧组件当成生产架构中心。当前 LangChain 更推荐用 Agent state、checkpointer、middleware 来管理短期记忆,用 store 管理长期记忆。
可以这样记:
- 想理解原理,看
ConversationBufferMemory、ConversationBufferWindowMemory、ConversationTokenBufferMemory。 - 想写新 Agent,用
create_agent + checkpointer + thread_id。 - 想控制上下文长度,用
before_modelmiddleware 裁剪或摘要。 - 想上生产,用数据库 checkpointer、明确的会话隔离、脱敏、删除和可观测性。
参考: