# 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 的历史消息
  -> 选择要进入上下文的部分
  -> 调用模型
  -> 保存本轮用户消息、模型回复、工具结果
  -> 下一轮继续读取
1
2
3
4
5
6

缓冲记忆关注的是其中的“选择要进入上下文的部分”。

如果把所有历史都带上,就是全量缓冲。如果只带最近几轮,就是窗口缓冲。如果按 token 上限截断,就是 token 缓冲。如果把旧消息总结成摘要,再拼接最近消息,就是摘要缓冲。

这些策略不是 LangChain 独有,而是所有聊天机器人都绕不开的上下文工程问题。

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
1

窗口缓冲记忆的旧式写法大概是这样:

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})
1
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)
1
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
1

应用启动时创建 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"}},
    )
1
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}
1

这样可以避免不同用户、不同租户、不同会话之间的状态串线。

还要注意 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(),
)
1
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,
        ]
    }
1
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(),
)
1
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 检索内容
  + 当前问题
1
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_model middleware 做裁剪。
  • 用 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"}},
    )
1
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:],
        ]
    }
1
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_model middleware 裁剪或摘要。
  • 想上生产,用数据库 checkpointer、明确的会话隔离、脱敏、删除和可观测性。

参考: