如果你做过多轮对话类的应用，大概率遇到过这样一幕：用户上午刚告诉机器人“我的设备型号是X”，下午再追问“我的设备支持离线升级吗”，机器人却像第一次见面一样反问“请问您使用的是哪一款设备”。这不是模型笨，而是大模型的聊天接口天然就是一条“金鱼”——它每一次回答都只依赖你这次调用传入的文本，上次说过什么，它完全不记得。
想要告别这种金鱼记忆，LangChain提供了两块拼图：一块是记忆机制（Memory），负责把历史对话组织成模型看得懂的上下文；另一块是LCEL持久化，负责让这些历史在进程重启、请求切换之后仍然存活。这篇文章，我会把LangChain的核心记忆机制逐个拆开聊清楚，再从零写完一条带记忆的对话链路，最后把生产环境里真正会踩的那些坑一并交代出来。适合正在做客服机器人、知识问答、Agent多轮交互的开发者参考。
1. 为什么大模型天生“金鱼”：无状态接口与LangChain记忆的补课思路
1.1 聊天接口的“无状态”本质
先看一个基本事实：大模型厂商提供的Chat类接口，本质上是一个“无状态函数”。你调用它时，入参是一串文本（包括角色、内容），出参是一段补全文本。模型内部不会维护任何会话状态，它不会因为你上一轮说过“我叫老王”就自动记住，也不会因为你和它在同一个网页对话框里聊过天就产生记忆。
一个最朴素的对话实现，往往是前端把整个聊天记录数组原样传给后端，后端再把数组拼进提示词里一起发给模型。这样看起来“有记忆”，其实只是应用层把历史当作上下文重新传了一遍。换句话说，记忆从来不是模型的固有属性，而是调用者自己维护的一门手艺。
这带来一个直接后果：历史越长，每次请求的Token消耗越大，响应越慢。很多新手在这里会掉进两个误区——要么完全不做记忆，导致对话断片；要么把全部历史一股脑丢给模型，很快就把上下文窗口撑爆。真正的解法，是要有一套“记忆的取舍策略”：记住什么、丢什么、怎么组织、怎么存储。
1.2 LangChain记忆机制的运行闭环
LangChain对“记忆”的抽象，本质上就是围绕上面那句话做了一层工程封装。它把一段对话拆成三件事：
- 读：调用模型之前，从某个存储中取出历史消息数组；
- 拼：把历史消息作为
HumanMessage、AIMessage等角色消息，插入到当前用户输入前面； - 写：拿到模型回复后，把“用户输入+模型回复”追加回存储。
传统意义上，LangChain的ConversationBufferMemory这类组件封装的正是这个过程。它们内部维护一个chat_memory属性，通常是ChatMessageHistory的实例，专门负责存取消息列表。你在用ConversationChain时，记忆组件会自动把历史塞进提示词，也会自动把新消息追加进去。
这里有一个非常重要的认知：不同的记忆组件，区别只在于“读历史时怎么筛选、怎么改写”，而“写历史”这件事几乎都是一样的。缓冲记忆是原样全读，窗口记忆是只读最近几轮，Token记忆是截断到指定长度，摘要记忆是把旧历史压缩成一段总结再读。理解了这条主线，你再看任何记忆组件的源码都不会迷路。
2. 五大核心记忆机制逐个拆解：缓冲、窗口、Token、摘要、语义检索
2.1 ConversationBufferMemory：最朴素的全量记录
这是最容易理解的方案：把所有历史对话原封不动地存下来，每次调用时全部拼进提示词。
from langchain.memory import ConversationBufferMemory from langchain.chains import ConversationChain chain = ConversationChain( llm=llm, memory=ConversationBufferMemory() )它能保证信息不丢，连用户当初那句话里的语气词都还在。但缺点同样赤裸——上下文长度会随着对话轮数线性增长。你开发阶段测五轮没问题，真正上线后用户连续聊二十轮，Token费用和延迟都会让你肉疼。它比较适合“对话轮数有限、需要精确复述”的简单内部工具，不适合长期开放的客服场景。
我在早期项目里就吃过这个亏：有个内部数据分析助手，用户习惯在一个会话里连续提问十几轮，结果不到一周，平均单次请求的Token消耗翻了快四倍。后来才换成窗口记忆加摘要的混合方案。
2.2 ConversationBufferWindowMemory：只保留最近K轮的“短期记忆”
窗口记忆的思路很直观：我只留最近k轮的消息，更早的直接丢掉。代码上只需要传入一个k参数。
from langchain.memory import ConversationBufferWindowMemory memory = ConversationBufferWindowMemory(k=2)它的优点很明显：上下文长度可控，实现简单，Token费用稳定。缺点也同样直接——模型会失去“远期记忆”。假设用户第十三轮问“我一开始说的那个需求你帮我实现一下”，模型只能看到最近两轮的内容，完全不知道“那个需求”指的是什么。
这里的k其实是个需要业务经验去调的超参。会话里如果经常出现“我前面说的那个”这类指代，K值就得适当调大；如果每个问题都比较独立，K值取2到3就够了。另外要注意，窗口记忆的单位是“轮次”，而不是“Token数”，如果某轮回复特别长，即使只有两轮也可能撑爆上下文窗口。
2.3 ConversationTokenBufferMemory：按Token数切，不按轮数切
Token记忆是把轮次单位换成了Token单位。它内部维护一个计数器，当历史消息总长度超过max_token_limit时，从最早的消息开始往外删，直到总量回到限制以内。
from langchain.memory import ConversationTokenBufferMemory memory = ConversationTokenBufferMemory( llm=llm, max_token_limit=1000 )我比较推荐在实际项目里优先考虑这个方案，因为模型上下文窗口的计费单位是Token，不是轮次。有一天用户连续发几个大段代码，三五个消息就可能超过一万Token；另一天用户只回“嗯”“好的”，二十轮加起来也没多少。按轮次切窗口，遇到极端情况仍然会爆；按Token切，至少费用是可控的。
使用的时候有一个细节：不同的语言、不同模型的分词器，Token计数都不一样。英文大概四个字符一个Token，中文通常一个字可能就是一个甚至更多Token。所以max_token_limit必须根据你的业务模型实测调整。我通常先在测试集上把历史长度拉满跑一遍，再取一个“历史上限 = 模型最大上下文 × 30%”的保守值。
2.4 ConversationSummaryMemory / SummaryBufferMemory：把历史压成摘要
Token记忆解决的是长度问题，但它的代价是“直接丢弃旧消息”。如果旧消息里藏着关键信息，丢掉了就再也找不回来。摘要记忆提供了一个更聪明的思路：用一个专门的调用，把旧对话压缩成一小段自然语言摘要，然后只把摘要喂给模型。
from langchain.memory import ConversationSummaryMemory memory = ConversationSummaryMemory(llm=llm)你要注意，这里的llm不是用来回答用户问题的，而是用来给历史“写总结的”。每次追加消息后，它会把已有摘要+新消息合并，重新生成一段更新后的摘要。比如“用户住在南方某城市，偏好节能型设备，曾在8月15日询问过A型号的离线升级能力”，这段摘要既保留了关键事实，又不会占太多上下文。
摘要记忆最大的坑在于：摘要本身会有信息损失，甚至可能出现幻觉。如果摘要模型在某次生成了错误信息，后面所有的对话都会基于这个错误继续。所以它更适合“用户画像类”场景——记偏好、记意图、记决定，不适合“需要精确回忆数字和代码”的场景。
ConversationSummaryBufferMemory是它的升级版，混合了缓冲和摘要两种策略：在Token总量没超阈值前，保留原文；超了之后，把最早的一部分消息转成摘要，新消息继续保留原文。这个组合在实践里非常实用，尤其适合那种“最近几轮细节必须准确、更早内容只需要大意”的客服系统。
2.5 VectorStoreRetrieverMemory与实体记忆：从“读完所有历史”到“只捞相关内容”
前四种记忆本质上都是“顺序读取整段历史”，要么全读，要么截断，要么压缩。这样做的缺点是：如果你和用户已经聊了几百轮，即使把历史压缩成摘要，也可能很长；而且用户在第十三轮提到的细节，和第三十轮的问题高度相关，但摘要机制未必能把它单独捞出来。
向量检索记忆的思路是：把历史中的每条消息先做向量化（Embedding），存进向量数据库；每次新问题进来，先做一次相似度检索，只把最相关的几条历史片段取出来，拼进上下文。
from langchain.memory import VectorStoreRetrieverMemory memory = VectorStoreRetrieverMemory( retriever=vectorstore.as_retriever(search_kwargs={"k": 3}) )我在一个知识库问答项目里用过这个方案。用户的对话历史有上千条，其中很多是重复提问，向量检索能精准命中与当前问题语义相近的那几条历史，效果立竿见影。代价是：向量化本身有成本，历史数据需要写入向量库，相似度检索也可能召回无关内容。它通常不单独使用，而是和窗口记忆或摘要记忆配合。
顺带提一下EntityMemory，它负责抽取对话中出现的实体信息（人名、产品名、偏好等），比如“用户A倾向于夜间咨询”，然后把实体及其属性存成结构化记录。这类记忆更接近“知识图谱式”的长期记忆，适合需要跨会话维护用户画像的场景。它的依赖也比较重，需要实体识别能力和存储设计，一般业务用不上，但思路值得借鉴。
3. 旧API时代的终结与新默认路线：LCEL管道怎么接住历史消息
3.1 传统Memory组件为什么逐渐被官方弃用
如果你现在去翻了官方文档，可能会发现ConversationBufferMemory这一类组件要么标了Deprecated，要么在新的langchain版本里被挪到了langchain_community。我在半年多前也困惑过：为什么一个刚学会的东西，官方转身就不推荐用了？
核心原因有两个。第一，传统Memory组件大多依赖ConversationChain和LLMChain，而它们和新的LCEL（LangChain Expression Language）管道风格不兼容。LCEL的核心哲学是“管道组合”：prompt | model | parser，组件之间只通过参数和返回值流动，传统Memory却是内部维护状态的，二者很难优雅地拼在一起。第二，官方希望把“消息列表”提升为一等公民，让大家直接操作messages数组，而不是被封装在一个黑盒里。记忆的读取、裁剪、改写逻辑，交给开发者用明确的代码控制，反而更透明、更不容易出错。
3.2 RunnableWithMessageHistory：连接LCEL与ChatMessageHistory的适配器
直接操作消息列表的好处是透明，但完全没有封装的话，每次调用你都得手动完成“取历史→拼上下文→调用→写回”四步，写起来又臭又长。LangChain给出的中间方案是RunnableWithMessageHistory。
它本质上是一个包装器：接收一个普通Runnable，再接收一个get_session_history回调函数，调用时通过config参数传入session_id，包装器会自动帮你完成历史读取、消息拼接、结果写回这三个动作。
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_redis import RedisChatMessageHistory prompt = ChatPromptTemplate.from_messages([ ("system", "你是一位耐心、细心的客服助手。"), MessagesPlaceholder("history"), ("human", "{input}"), ]) chain = prompt | llm def get_session_history(session_id: str): return RedisChatMessageHistory( session_id=session_id, url="redis://localhost:6379/0", ttl=7200, ) with_history = RunnableWithMessageHistory( chain, get_session_history, input_messages_key="input", history_messages_key="history", )这里有几个要点我最初也理解错了：
input_messages_key告诉包装器：用户的新输入存在Runnable输入的哪个字段里；history_messages_key告诉包装器：提示词模板里哪个占位符是用来放历史消息的；get_session_history不是只负责“取”，而是“取不到就新建”，相当于一个延时构造器。
3.3 消息列表本身也是一种记忆：什么时候不需要任何记忆包装
不想引入那么多概念的话，完全可以不用任何Memory类。如果场景本身就在一个长连接里，比如WebSocket对话、单次任务型交互，你可以在外部维护一个messages列表，每次直接把整个列表传给构建好的提示词。
messages = [system_message] while True: user_input = input("你：") messages.append(HumanMessage(content=user_input)) resp = model.invoke(messages) messages.append(AIMessage(content=resp.content))这段代码虽然简短，但一定要意识到它的边界：messages存在于进程内存里，服务重启、多实例部署、请求路由到另一台机器时，记忆就会全部失效。这正是下一节“持久化”要解决的问题。
4. 持久化实战：一条带记忆的LCEL链路从进程内存跑到Redis
4.1 依赖与版本选择
实战开始之前，先确认依赖。我当前使用的版本基线是langchain>=0.2、langchain-openai>=0.1，如果你用的是更早的版本，接口命名会差很多，网上很多旧教程直接复制会报错。
pip install langchain langchain-openai langchain-redis redislangchain-redis是官方维护的Redis适配包，如果环境比较老，也可以从langchain_community.chat_message_histories导入RedisChatMessageHistory，但新项目我更建议直接用官方包，维护和升级都更省心。
模型部分我用通用的Chat接口占位，实际使用时配置你自己的api_key和base_url即可：
from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="your-model-name", base_url="https://your-endpoint.example.com/v1", api_key="your-api-key", )4.2 内存版：先把链路跑通
我们先写一个不落盘的版本，用InMemoryChatMessageHistory维护历史，把链路完整跑通，再换存储。
from langchain_core.chat_history import InMemoryChatMessageHistory store = {} def get_session_history(session_id: str): if session_id not in store: store[session_id] = InMemoryChatMessageHistory() return store[session_id] with_history = RunnableWithMessageHistory( chain, get_session_history, input_messages_key="input", history_messages_key="history", ) print(with_history.invoke( {"input": "我叫老王，住在南方一座城市。"}, config={"configurable": {"session_id": "user-session-001"}}, ).content) print(with_history.invoke( {"input": "我住在哪座城市？"}, config={"configurable": {"session_id": "user-session-001"}}, ).content)第二次提问时，模型能看到第一轮的历史，所以能正确回答出“南方一座城市”。如果你把session_id换掉，模型就完全失忆。这个简单的测试就验证了记忆链路的核心逻辑。
代码里的store是一个全局字典，注意它只能工作在单进程环境。一旦服务部署成多副本，每个进程的字典互相独立，session_id相同的请求落在不同机器上，照样会失忆。
4.3 从进程内存换成Redis：一套接口搞定持久化
现在把get_session_history从字典方式改成Redis方式，其余代码一行都不用动：
from langchain_redis import RedisChatMessageHistory def get_session_history(session_id: str): return RedisChatMessageHistory( session_id=session_id, url="redis://localhost:6379/0", ttl=7200, )这一步换完之后，记忆的“生命周期”就从进程级别提升到了存储级别。进程重启、代码发版、请求被负载均衡到另一台机器，只要Redis里的数据还在，用户回来接着聊，模型依然记得上下文。这才是真正意义上的“长期记忆”。
这里我提一个实操建议：RedisChatMessageHistory给消息列表设置ttl，也就是每条会话过期时间。千万别小看这个参数。线上会话如果无限期保留，Redis内存会被各种垃圾会话占满。按你的业务节奏设个合理的过期时间，比如客服场景7天、工具类场景24小时，既满足连续性又控制成本。
4.4 更进一步：LangGraph的Checkpointer做会话级状态落盘
如果你觉得RunnableWithMessageHistory还是有边界——它只管“消息列表”，管不了对话流程里复杂的中间变量——那么LangGraph的Checkpointer是更完整的状态持久化方案。简单来说，它会记录整张状态图的节点状态，支持断点续跑、回放历史、甚至从某个节点分叉出另一条分支。
from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, MessagesState, START checkpointer = MemorySaver() graph_builder = StateGraph(MessagesState) graph_builder.add_node("model", llm) graph_builder.add_edge(START, "model") graph = graph_builder.compile(checkpointer=checkpointer) result = graph.invoke( {"messages": [("human", "我叫老王，住在南方一座城市。")]}, config={"configurable": {"thread_id": "user-thread-001"}}, )这里用thread_id来标识一个会话线程。如果你之后用同一个thread_id继续invoke，模型会带着前面的历史继续往下走。LangGraph的Checkpointer同样可以换成Redis、SQLite、PostgreSQL等外部存储，适合真正复杂的Agent工作流。
5. 生产环境避坑清单：会话ID、上下文膨胀、并发读写与过期策略
5.1 session_id从哪来？千万别用全局写死
我见过不少新手的Demo代码里直接把session_id写死成"abc"，测完就完事了。但生产环境里，会话ID必须由请求上下文决定，通常从请求头、Cookie、登录态或者前端传入的对话ID中提取。
def get_session_id_from_request(request): if request.headers.get("X-Session-Id"): return request.headers["X-Session-Id"] return request.cookies.get("session_id", "anonymous")需要注意的边界：同一个用户可能开了多个会话窗口，所以用“用户ID”直接当session_id会把所有窗口的聊天串在一起，容易造成上下文混乱。我的实践是“用户ID + 会话标识”双维度生成一个唯一的会话ID，比如user_1024__chat_8，这样既能定位归属，又能把不同会话隔离开。
5.2 上下文膨胀是隐形杀手：用trim_messages给历史上把锁
RunnableWithMessageHistory的默认行为是把历史消息全部传给模型。它不仅不裁剪，还会无限增长。上线跑几天后，单次请求的Token可能直接顶满上下文窗口。这是持久化带来的第二个坑：记忆不会消失，所以历史会无限膨胀。
解决办法是给历史加上一道裁剪闸门。官方推荐的组件是trim_messages，它可以按照Token数从尾部保留消息，确保历史长度不会突破预算。
from langchain_core.messages import trim_messages trimmer = trim_messages( token_counter=llm, max_tokens=2000, strategy="last", include_system=True, start_on="human", )把它接到管道里，可以在调用模型前先对历史做一次裁剪。这里注意start_on="human"很关键：它保证裁剪后的消息序列是以“人类消息”开头，而不是从半截的模型回复开始，否则用户看到上下文会非常错乱。
关于预算，我通常按“模型最大上下文 × 30%”来设定历史Token上限，剩下的空间留给系统提示词和模型输出。比如上下文窗口是8K，历史截到2K左右；窗口是128K的模型，可以截到4K到8K。
5.3 并发、TTL与Redis连接池：多副本环境下的记忆安全
持久化之后，并发问题也随之而来。如果多个用户同时命中同一个session_id，对Redis里同一份消息列表做并发追加，可能会产生顺序错乱，甚至覆盖丢失。虽然很多实践场景下同一个会话会被串行处理，但在设计API服务时，还是要明确这一点：
- 会话隔离是硬要求：不同用户必须用不同的
session_id，否则记忆会互相污染； - 同一个会话的多端并发要避免：Web端和客服后台同时向一个会话发消息时，建议做会话级锁或合并策略；
- Redis连接要复用：不要在每次请求时新建连接对象，使用连接池，并给Redis设置合理的客户端超时。
另外，设置TTL不仅仅是内存规划问题，还经常涉及合规要求。用户会话数据保留多久、是否包含敏感信息，都应该在设计早期就定好策略，而不是等到运维提问才补。
5.4 异步接口别忘了：a联调时直接用ainvoke
现在的真实服务基本都是异步框架（比如FastAPI配合事件循环）。RunnableWithMessageHistory同时提供了同步和异步接口，异步模式下直接用ainvoke：
response = await with_history.ainvoke( {"input": "你好，能帮我查订单吗？"}, config={"configurable": {"session_id": session_id}}, )如果不注意，在异步函数里调用同步的invoke，一旦请求量大，事件循环会被阻塞，接口延迟一上来，体验全毁。这块写起来很简单，但很多从同步服务迁过来的项目都栽过。
6. 五种方案的实测对比与按场景选型建议
6.1 一表看懂五种记忆机制的核心差异
| 记忆机制 | 信息完整度 | Token成本 | 实现复杂度 | 典型场景 |
|---|---|---|---|---|
| ConversationBufferMemory | 高 | 持续增长 | 低 | 短会话、内部工具 |
| ConversationBufferWindowMemory | 中（保留近期） | 可控 | 低 | 独立问答、客服FAQ |
| ConversationTokenBufferMemory | 中（按Token截断） | 可控 | 中 | 长对话、需要控制预算 |
| ConversationSummaryMemory | 中（摘要有损） | 恒定但需额外摘要调用 | 中 | 用户画像、偏好类对话 |
| VectorStoreRetrieverMemory | 高（按需召回） | 低（只喂相关片段） | 高 | 长历史、知识库问答 |
数据来自我自己的几次压测，具体数值会随模型和业务变化，但趋势很稳定：缓冲记忆最耗Token，窗口和Token记忆适合中等时长对话，摘要记忆适合长期“大概记住”，向量检索适合长期“精确找回”。
6.2 三个真实场景怎么搭配
客服机器人：这是最典型的“近期信息准确+远期意图模糊”场景。我推荐TokenBufferMemory或SummaryBufferMemory，最近几轮保持原文，早期历史转成摘要。如果还能配合业务工单ID，把工单状态作为结构化上下文直接注入，比单靠记忆可靠得多。
文档问答助手：用户会追问“刚才那段条款在哪个页面”“你上一轮的结论为什么变”，这时需要用VectorStoreRetrieverMemory把历史提问和回复向量化，再和文档向量库一起做检索，才能把“用户上一轮提到的文档片段”精确捞回来。
数据分析Agent：这个场景最容易出问题，因为用户会在多轮里逐步给出约束条件，比如“先按月份过滤”“再排除退货订单”。这些约束像拼图一样分散在各轮里，窗口记忆一定会丢，摘要记忆会把条件混淆。我建议用窗口记忆加结构化状态，每一步把已确认的条件写入一个明确的JSON状态字段，随时回显给用户确认。
6.3 我的默认配置：一套可以起步的组合
最后分享一套我在大多数新项目里的默认起步组合，它不是最优解，但是一个很稳的基线：
- 存储端：
RedisChatMessageHistory，TTL按业务定； - 裁剪端：
trim_messages，历史Token上限设为模型窗口的30%左右； - 组装端：
RunnableWithMessageHistory，输入输出键显式声明； - 进阶端：需要复杂状态机时再换成LangGraph的Checkpointer。
我在实际项目里体会最深的一点是：设计“记忆机制”之前，一定先想清楚一个问题——“用户下一轮最可能依赖哪些信息，以及这些信息应该以原文、摘要还是结构化字段的形式存在”。记忆方案没有银弹，没有哪种机制能覆盖所有场景，但理解了记忆的读、拼、写流程之后，你随时可以针对业务需求组合出适合自己的方案，而不是被某个记忆类库里固定的策略限制住。
