第一次把 LangChain 的 Memory 接进对话链,多数人会经历一个挺困惑的阶段:前两轮效果很好,第三轮开始答非所问,第五轮直接把 A 用户的问题答给了 B 用户。我见过一个小组为了这个现象连着排查了两天,最后定位到的既不是模型能力问题,也不是提示词写坏了,而是memory_key和 prompt 里的占位符名字对不上——prompt 里写的是{chat_history},Memory 默认吐出来的键名是history,占位符压根没被填充,模型每轮拿到的都是一段空字符串。这个坑很典型,因为它说明 Memory 模块的性质经常被误解:它不是"给模型装记忆",而是在每一轮请求前,决定往 prompt 里塞什么、塞多少。
LangChain 的 Memory 模块在整个框架里属于小而关键的一层,代码量不大,但几乎所有做多轮对话的项目都会用到它。它解决的问题很明确:大模型的 Chat 接口本身是无状态的,每次调用都得把历史重新拼一遍,而上下文窗口有上限、按 token 计费,所以必须有人来做"历史如何压缩进上下文"这件事。这套东西适合谁看?如果你正在写一个带记忆的问答助手、客服机器人、或者任何需要跨轮次保持一致性的应用,并且已经卡在"历史越堆越长、成本越来越高、模型还开始跑偏"的阶段,那这篇的实践细节应该能省你不少时间。下面我按抽象层、源码实现、可跑示例、踩坑记录、版本迁移这几个角度,把自己趟过的路完整走一遍。
1. 多轮对话为什么会"失忆":Memory 到底在管什么
1.1 无状态接口与有状态业务之间的落差
理解 Memory 之前,先把模型的调用模型想清楚。不管是哪家的 Chat 接口,一次请求的结构永远是"一串消息进、一条消息出",服务端不会替你保存上一条请求的上下文。你觉得它在跟你聊天,其实是你的代码每次都把前面所有对话重新发了一遍。打个比方,这就像每次打电话都要先把之前所有通话录音从头播放一遍,对方才能接上话。
这个设计带来的直接后果是:历史长度直接等于每次请求的输入长度。聊到第 50 轮时,你每问一句"今天天气怎么样",都要连带把前面 49 轮的内容重新付费传一次。业务上需要的是"有状态的连续会话",接口提供的却是"无状态的单次推理",这中间的落差就是 Memory 模块存在的意义。
2.1 里会讲到,LangChain 把这块抽象得很薄,但薄不代表简单,因为真正的难点不在"存",而在"取多少"。
1.2 Memory 不是数据库,它的核心职责是分配上下文预算
新手最常见的误解是把 Memory 当成持久化存储层,觉得接上它对话就"记住"了。实际上 LangChain 内置的几个 Memory 实现,数据大多存在进程内存的一个列表里,进程一退就没了。它真正干的活,是在每一轮请求前回答三个问题:这次要带哪些历史、这些历史以什么格式塞进 prompt、以及超出预算的部分怎么处理。
所以更准确的说法是:Memory 是一个上下文预算管理器。存储只是它的手段,裁剪才是它的核心。你会发现 LangChain 里所有有意思的 Memory 变体,差异全部集中在"裁剪策略"上——有的按条数裁,有的按 token 数裁,有的把旧内容喂给模型压缩成一段摘要,有的干脆不做线性拼接,改成按语义相似度检索几条相关的。理解了这个视角,后面看源码就不会觉得这些类只是名字不同。
1.3 四种预算策略,对应四类业务取舍
把内置实现的策略归归类,其实就四种思路。第一种是全量保留,历史一条不落地拼进去,信息零损失,代价是 token 随轮次线性增长,迟早撞上窗口上限。第二种是窗口截断,只保留最近 k 轮,实现简单、成本可控,缺点是早期的关键信息会被硬切掉,比如用户在第一轮说的姓名。第三种是摘要压缩,把超出部分交给模型写成一段摘要,用一次额外的模型调用换长期的 token 节省。第四种是语义检索,把所有历史向量化存起来,每轮按当前问题的相似度召回最相关的几条。
这四种没有优劣之分,只有场景匹配度。客服工单这种需要精确回溯细节的场景,摘要压缩容易丢关键数字;而长周期陪伴类对话,全量保留的成本又扛不住。我自己做项目时的默认起点是窗口截断加摘要兜底,也就是ConversationSummaryBufferMemory那套混合策略,后面第 3 章会细说它内部的临界点是怎么设计的。
2. BaseMemory 拆解:四个方法撑起的一层薄抽象
2.1 memory_variables 与 load_memory_variables 的声明与供给分离
LangChain 的 Memory 抽象层非常克制,BaseMemory这个基类只定义了四个必须实现的东西。前两个是一对:memory_variables是一个属性,返回这个 Memory 会向 prompt 注入哪些变量名;load_memory_variables是实际取数的方法,接收当前轮的输入,返回一个字典。
from abc import ABC, abstractmethod from typing import Any, Dict, List class BaseMemory(ABC): @property @abstractmethod def memory_variables(self) -> List[str]: """声明会向 prompt 注入哪些变量名""" @abstractmethod def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]: """根据当前输入,返回要注入 prompt 的变量字典""" @abstractmethod def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None: """一轮结束后,把输入输出写回记忆""" @abstractmethod def clear(self) -> None: """清空记忆"""为什么要把"声明变量名"和"实际取数"拆成两件事?因为 Chain 在真正执行之前,需要先知道这个 Memory 会占用哪些 prompt 变量,才能校验模板里的占位符是否都被覆盖了。这就是开头那个memory_key对不上的 bug 的根源——Chain 拿memory_variables去比对模板变量,发现对不上就直接报错或者静默留空。把声明和供给分开,让校验可以发生在执行之前,这个设计本身没问题,问题在于默认值history太容易被忽略。
2.2 save_context 一次调用写入两条消息
save_context的签名值得单独说。它接收两个字典:inputs是这一轮用户的输入,outputs是模型返回的结果。两个都是字典而不是字符串,因为一个 Chain 可能有多个输入输出变量,Memory 需要知道从哪个键里取值。
在BaseChatMemory这个中间层里,取值规则被固化了下来。如果不显式指定input_key和output_key,它会走一套推断逻辑:输入侧从inputs里找第一个不是 Memory 自身变量的键,输出侧要求outputs必须只有一个键,否则直接抛ValueError。这段逻辑看着琐碎,但它解释了为什么在多链组合的场景里,你经常需要手动指定这两个参数——推断逻辑在单链场景下够用,一旦链的输出包含多个字段,它就没法猜了。
from langchain.memory import ConversationBufferMemory memory = ConversationBufferMemory( memory_key="history", # 注入 prompt 的变量名 input_key="question", # 从输入的哪个键取用户话 output_key="answer", # 从输出的哪个键取模型话 return_messages=True, # 返回消息对象还是拼好的字符串 )return_messages这个参数也值得留意,它决定了load_memory_variables返回的是BaseMessage对象列表还是一段拼接好的纯文本。用 Chat 模型配 ChatPromptTemplate 时设成True更自然,因为消息角色能被模型正确识别;用纯文本补全模型时设成False,让它输出Human: xxx\nAI: xxx这种格式。
2.3 chat_memory 才是真正存数据的地方
BaseChatMemory里有一个字段叫chat_memory,类型是BaseChatMessageHistory。这是整个 Memory 体系里真正装东西的容器,一个只负责增删查的简单消息列表。区分 Memory 和 MessageHistory 这两层很重要:Memory 是"策略层",负责决定怎么压缩、怎么格式化;MessageHistory 是"存储层",只负责存消息。
这个分层带来了一个很实用的扩展点。想换存储介质,不用改 Memory 的实现,只要换一个BaseChatMessageHistory的子类塞进去就行。官方生态里有基于各种外部存储的实现,但我觉得更稳妥的做法是自己写一个轻量的,代码不到三十行,可控性还高:
import json, os from langchain_core.chat_history import BaseChatMessageHistory from langchain_core.messages import BaseMessage, HumanMessage, AIMessage class JsonFileHistory(BaseChatMessageHistory): path: str _messages: list = [] def __init__(self, path: str): super().__init__() self.path = path self._load() @property def messages(self): return self._messages def add_message(self, message: BaseMessage) -> None: self._messages.append(message) self._flush() def clear(self): self._messages = [] self._flush() def _load(self): if not os.path.exists(self.path): return with open(self.path, "r", encoding="utf-8") as f: for item in json.load(f): cls = HumanMessage if item["type"] == "human" else AIMessage self._messages.append(cls(content=item["content"])) def _flush(self): with open(self.path, "w", encoding="utf-8") as f: json.dump( [{"type": "human" if isinstance(m, HumanMessage) else "ai", "content": m.content} for m in self._messages], f, ensure_ascii=False, indent=2, )这里有个小坑得提前说:BaseChatMessageHistory是 Pydantic 模型,直接在__init__里给未声明的属性赋值会触发校验错误。上面这段把path和_messages声明成了类级字段,就是为了绕开这个问题。我第一次写的时候用了普通的实例属性,报错信息指向字段校验,排查了半天才反应过来是 Pydantic 在管字段。
2.4 clear 方法与生命周期管理
四个方法里最容易被忽略的是clear。它在基类里只是个抽象方法,很多自定义实现就随手写个pass应付过去。但在真实服务里,清空记忆是高频操作:用户点"新建对话"要清、会话超时要清、敏感信息需要定期清理也要清。更要紧的是,如果 Memory 实例是全局共享的(下一章会讲这个坑有多致命),clear没实现好就意味着旧会话的数据永远赖着不走。
我的习惯是给每个用户会话一个独立的 Memory 实例,clear里除了清消息列表,还要把摘要缓存、token 计数器这些派生状态一并重置。这些东西一旦和消息列表不一致,下一轮load_memory_variables就会返回一段对不上的内容,而且这种不一致很难从日志里看出来。
3. 五种内置 Memory 的源码走读:每一种都在拿什么换什么
3.1 ConversationBufferMemory:零损失的代价是线性膨胀
这是最直白的一个实现,load_memory_variables基本上就是把chat_memory.messages原样返回或者拼成字符串。没有裁剪、没有压缩,历史完整保留。
class ConversationBufferMemory(BaseChatMemory): human_prefix: str = "Human" ai_prefix: str = "AI" memory_key: str = "history" @property def buffer(self): if self.return_messages: return self.chat_memory.messages return get_buffer_string( self.chat_memory.messages, human_prefix=self.human_prefix, ai_prefix=self.ai_prefix, ) @property def memory_variables(self): return [self.memory_key] def load_memory_variables(self, inputs): return {self.memory_key: self.buffer}源码就这么点,一眼看完。它的价值在于简单可预测,调试的时候你能明确知道模型看到了什么。问题也直接:token 消耗随轮次线性增长。我做过一个粗算,一个中文对话轮次平均 80 到 120 个 token,来回算一次完整对话大概 200 token,50 轮之后单次请求的输入就要多出 1 万 token,再往后成本曲线陡得没法看。所以这个实现适合的场景其实很窄——短会话、调试期、或者你明确知道对话不会超过十几轮。
3.2 ConversationBufferWindowMemory:k 值的真实含义
窗口版加了一个参数k,只保留最近 k 轮。很多人对k的理解有偏差,以为k=5是保留 5 条消息,实际上是 5 个往返,也就是 10 条消息(5 条用户消息 + 5 条模型消息)。这个差异在排查"为什么历史比我预期的长"时经常出现。
class ConversationBufferWindowMemory(BaseChatMemory): human_prefix: str = "Human" ai_prefix: str = "AI" memory_key: str = "history" k: int = 5 @property def buffer(self): if self.return_messages: return self.chat_memory.messages[-self.k * 2:] return get_buffer_string( self.chat_memory.messages[-self.k * 2:], human_prefix=self.human_prefix, ai_prefix=self.ai_prefix, )注意[-self.k * 2:]这个切片。这里藏着一个边界问题:如果消息列表的长度是奇数(比如某次模型调用失败,只写入了用户消息),切片之后最早的那条消息可能是 AI 消息,对话就从"模型说话"开始,某些对角色顺序敏感的模型会表现异常。LangChain 自己没有处理这个边界,实际项目里我一般会在load_memory_variables外面包一层,检查首条消息的角色,如果是 AI 就往后挪一位。
窗口版最让人纠结的是k怎么定。给个参考:客服类场景按业务流程走,k取 3 到 5 够用,因为核心信息集中在最近几轮;任务型对话(比如填表单、走流程)可以取 8 到 10,因为需要回溯前面填过的字段;至于那种用户第一轮自报姓名、后面全靠代词的场景,窗口截断基本无解,得靠摘要或者把它单独抽成结构化字段存起来。
3.3 ConversationSummaryMemory:用一次额外调用换长期节省
摘要版的思路完全不同。它不保留原文,而是维护一段不断累积的摘要。每轮结束后,把新的对话内容连同已有摘要一起送给模型,让它输出更新后的摘要。
class SummarizerMixin: def predict_new_summary(self, messages, existing_summary: str) -> str: new_lines = get_buffer_string( messages, human_prefix=self.human_prefix, ai_prefix=self.ai_prefix, ) chain = self.prompt | self.llm return chain.invoke({ "summary": existing_summary, "new_lines": new_lines, }).content这里的关键取舍在于:每轮多一次模型调用。对话轮次多了,调用次数是线性叠加的,而且这次调用的输入还包含累积的摘要,成本并不低。所以它的节省体现在"长期"上——聊到第 100 轮时,buffer 版的上下文可能已经上万 token,摘要版还稳定在几百 token。
代价除了调用开销,还有信息有损。摘要模板我见过不少版本,最常见的提示是"逐步总结以上对话,在最新总结中返回新提炼的摘要,并加上之前摘要的信息"。这个措辞给了模型很大的自由裁量权,它可能把订单号、金额、日期这类精确信息压缩掉。我踩过一次很实的坑:用户第一轮报了工单号,二十轮后需要核对,摘要里只剩下"用户提到了一个工单",号码没了。后来我在摘要模板里硬加了一条规则——数字、编号、日期、专有名词必须原样保留,情况才好转。
3.4 ConversationSummaryBufferMemory:prune 循环里的临界点设计
这是混合策略,也是我实际项目里用得最多的一个。它同时维护一份原始消息列表和一段移动摘要,当原始部分超出max_token_limit时,从头部开始弹出消息、逐条或逐批压缩进摘要。
class ConversationSummaryBufferMemory(BaseChatMemory, SummarizerMixin): max_token_limit: int = 120 moving_summary_buffer: str = "" memory_key: str = "history" def prune(self): while self.buffer_size > self.max_token_limit: pruned_memory = self.buffer.pop(0) self.moving_summary_buffer = self.predict_new_summary( pruned_memory, self.moving_summary_buffer, ) def load_memory_variables(self, inputs): buffer = self.buffer if self.moving_summary_buffer: buffer = [SystemMessage(content=self.moving_summary_buffer)] + buffer return {self.memory_key: buffer}max_token_limit的默认值 120 小得离谱,基本上聊两轮就会触发压缩,所以生产环境一定要手动调。这个参数怎么定,我一般按模型窗口的 15% 到 25% 来估:假设窗口是 8K,留给历史的部分大概 1500 token,取 1200 到 1800 之间都行,剩下的留给系统提示、检索到的文档和模型输出。
prune的循环实现有个容易被忽略的性能问题:它是逐条弹出并逐次调用模型生成摘要的。如果一次积压了 20 条消息需要压缩,那就是 20 次连续调用,接口延迟会非常明显。我在自己的实现里把它改成了批量压缩——一次弹出若干条、拼起来作为new_lines送进去,调用次数从 20 降到 2 到 3 次。改动不大,但用户能明显感觉到"卡顿消失了"。
3.5 VectorStoreRetrieverMemory:召回式记忆的适用边界
这个实现跳出了"线性拼接"的思路,把所有历史写成文档存进向量库,每轮根据当前输入做相似度检索,只把最相关的几条注入上下文。
from langchain.memory import VectorStoreRetrieverMemory memory = VectorStoreRetrieverMemory( retriever=vectorstore.as_retriever(search_kwargs={"k": 3}), memory_key="history", )它解决的是"长期记忆 + 稀疏相关"的问题,比如一个知识型助手需要记住几周前用户提过的偏好。但它的短板也很明显:一是依赖嵌入质量和相似度阈值,检索不准就会漏掉关键上下文;二是它不保证时序,召回的三条可能是第 50 轮和第 3 轮混在一起,缺少对话的连续感;三是每次写入都要做一次向量化,成本和延迟都上去了。
我个人的判断是,这类召回式记忆更适合作为辅助,而不是主力。比较稳的架构是:用窗口或摘要 Memory 保证最近对话的连贯性,再单独挂一个向量检索作为"长期事实库",两者在 prompt 里分不同变量注入——比如{recent_chat}和{recalled_facts},让模型能区分"刚说的"和"以前说过的"。混在一个变量里,模型很容易把几周前的事当成当前上下文来处理。
把五个实现放一起对照,取舍就很清楚了:
| 实现 | 裁剪依据 | 额外模型调用 | token 增长 | 适合场景 |
|---|---|---|---|---|
| ConversationBufferMemory | 无 | 无 | 线性 | 短会话、调试、演示 |
| ConversationBufferWindowMemory | 最近 k 轮 | 无 | 恒定 | 客服、任务型流程 |
| ConversationSummaryMemory | 模型摘要 | 每轮 1 次 | 近似恒定 | 长周期闲聊、陪伴类 |
| ConversationSummaryBufferMemory | 原始+摘要混合 | 触发时调用 | 有上界 | 通用生产场景 |
| VectorStoreRetrieverMemory | 语义相似度 | 检索+嵌入 | 与 k 相关 | 长期事实召回、知识助手 |
4. 三个可直接运行的示例:从最小骨架到多用户会话隔离
4.1 最小骨架:先跑通再谈优化
为了让示例能直接跑,模型初始化这一段我用了个工厂函数占位。各家 SDK 的初始化和鉴权差异不小,这部分不涉及 Memory 的逻辑,你自己替换成手头的实现就行,只要返回的对象满足BaseChatModel接口即可。
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.output_parsers import StrOutputParser from langchain.memory import ConversationBufferMemory from langchain.chains import LLMChain def get_chat_model(): """返回任意一个实现了 BaseChatModel 的模型对象,按你的环境自行实现""" ... llm = get_chat_model() memory = ConversationBufferMemory( memory_key="chat_history", return_messages=True, output_key="text", ) prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个简洁、直接的技术助手。"), MessagesPlaceholder(variable_name="chat_history"), ("human", "{input}"), ]) chain = LLMChain(llm=llm, prompt=prompt, memory=memory, output_key="text") print(chain.invoke({"input": "我在做一个带记忆的问答机器人"})["text"]) print(chain.invoke({"input": "你觉得上一句提到的方案,最大风险是什么?"})["text"])第二句能问出"上一句"指的是什么,说明记忆生效了。这段代码里有三个地方值得对照着看:memory_key="chat_history"必须和MessagesPlaceholder的variable_name完全一致,包括大小写和下划线;return_messages=True让 Memory 返回消息对象,直接匹配MessagesPlaceholder的期望类型;output_key="text"要和链的输出键对齐,LLMChain默认的输出键就是text。
跑通之后建议做一件事:把chain.verbose打开或者干脆打印memory.load_memory_variables({}),亲眼看一看每轮实际注入的是什么。我见过太多人一直靠猜,结果猜错了默认值,白白耗掉一天。
4.2 摘要记忆接入:参数调优比代码更重要
换成摘要版,代码改动很小,但参数必须重新算过:
from langchain.memory import ConversationSummaryBufferMemory summary_memory = ConversationSummaryBufferMemory( llm=llm, max_token_limit=1500, memory_key="chat_history", return_messages=True, output_key="text", ) chain = LLMChain(llm=llm, prompt=prompt, memory=summary_memory, output_key="text")这里有个容易忽略的点:ConversationSummaryBufferMemory需要单独传一个llm,这个模型专门用来做摘要,和你主对话用的可以是同一个,也可以分开。我倾向于分开——摘要任务对能力要求低,用一个便宜、快、稳定的模型就够了,主对话用能力强的。这样能明显压低平均延迟,尤其是对话密集的时候。
还有个细节:压缩发生在save_context之后,也就是上一轮结束时就完成了,不会拖慢当前轮的用户感知延迟。但如果一次积压大量消息(比如用户离线几天后回来),压缩会集中在那一轮触发,表现为突然的卡顿。我的做法是给prune加一个软上限,单次最多压缩 N 条,剩下的留到下一轮再处理,宁可摘要稍滞后也不让某一轮卡死。
4.3 多用户隔离:session_id 才是记忆的路由键
前面两个示例都是单会话的,一旦上服务就必然面对多用户。这一步做错,就是开头说的"把 A 的问题答给 B"。正确的做法是让每次请求携带一个会话标识,用它去映射到独立的消息历史。
from langchain_core.chat_history import InMemoryChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory store: dict[str, InMemoryChatMessageHistory] = {} def get_session_history(session_id: str) -> InMemoryChatMessageHistory: if session_id not in store: store[session_id] = InMemoryChatMessageHistory() return store[session_id] prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个简洁、直接的技术助手。"), MessagesPlaceholder(variable_name="history"), ("human", "{input}"), ]) chain = prompt | llm | StrOutputParser() chain_with_history = RunnableWithMessageHistory( chain, get_session_history, input_messages_key="input", history_messages_key="history", ) resp = chain_with_history.invoke( {"input": "记住我的项目代号是 Atlas"}, config={"configurable": {"session_id": "user-001"}}, ) print(resp) resp = chain_with_history.invoke( {"input": "我的项目代号是什么?"}, config={"configurable": {"session_id": "user-001"}}, ) print(resp)用不同的session_id发同样的第二问,你会得到"不知道"的回答,这就验证了隔离生效。这段代码的结构和第 4.1 节的差别值得注意:新版接口里,消息历史不再挂在 Chain 上,而是作为 Runnable 的一层包装,通过config里的configurable字段传递路由键。这个变化看起来是接口调整,实际影响的是整个应用的架构——记忆从"链的内部状态"变成了"外部注入的配置",天然适配无状态服务。
store这个字典在生产环境当然不能这么写。多进程部署时它是进程内私有的,同一个用户换个 worker 就丢历史;长时间运行还会无限增长。所以要么换成外部存储,要么至少加个过期清理。把store换成第 2.3 节那个JsonFileHistory,改一行就行:
def get_session_history(session_id: str): return JsonFileHistory(path=f"./sessions/{session_id}.json")要接数据库的话,实现一个BaseChatMessageHistory的子类即可,add_message、messages、clear三个成员补齐就能用,上层逻辑一行都不用动。这也是那层薄抽象带来的实际好处。
5. 实操踩坑记录:会话串号、token 雪崩与序列化
5.1 全局 Memory 对象是最大的雷
有个写法我见过不止一次:在模块顶层创建一个 Memory 实例当全局变量,所有请求共用一个。单用户测试时完全正常,一上线就出现用户之间互相看到对方对话的情况,而且因为并发写入,消息顺序还会错乱。
根因不难理解:Memory 实例持有一个chat_memory列表,多个请求往同一个列表里写,读写自然纠缠。更麻烦的是这类 bug 的表现很不稳定——用户量小的时候可能几天不出现,出问题时又很难复现,因为取决于请求的时间交错。
排查这类问题的思路是:先看 Memory 实例的创建位置,如果它出现在模块级、类级或者依赖注入的全局单例里,基本就能锁定。修法很直接,按session_id懒创建并缓存。但缓存要有淘汰策略,我一般用带过期时间的字典,比如 30 分钟无访问就清理,否则内存会一直涨。这里有个次生坑:只在缓存里删掉实例是不够的,如果实例对应的外部存储还在,下次重建时会重新加载历史,用户会看到"我明明清了记录,怎么又回来了"。所以clear必须同时清缓存和存储。
5.2 token 雪崩:buffer 不设上限的后果
Buffer 版最隐蔽的问题是超限后的表现。上下文一旦超过模型窗口,不同服务的处理方式不一样:有的直接报错,有的静默截断,有的截掉开头。静默截断最坑,因为系统提示往往在开头,被截掉之后模型的角色设定就丢了,回答风格会突然变得很不一样,而你从日志里看不出来任何异常。
我的处理习惯是加一层主动防线:在load_memory_variables之后、送进模型之前,做一次显式的裁剪和日志埋点。
from langchain_core.messages import trim_messages trimmer = trim_messages( max_tokens=2000, strategy="last", token_counter=llm, include_system=True, allow_partial=False, start_on="human", ) trimmed = trimmer.invoke(history_messages) print(f"history tokens before={len(history_messages)}, after={len(trimmed)}")strategy="last"表示保留最近的;include_system=True保证系统提示不被裁掉;start_on="human"保证裁剪后的第一条是用户消息,不会出现以 AI 消息开头的对话;allow_partial=False保证不会把一条消息切成两半。这几个参数我基本是每次都带全,少一个就可能在某个边缘场景出问题。日志埋点也别省,before和after两个数字能帮你判断到底是历史太长还是别的原因。
5.3 摘要记忆把关键信息"摘要没了"
前面提过这个坑,这里展开说处理办法。摘要的丢失是渐进式的,不会立刻暴露,等你发现的时候往往已经过了几十轮。我的做法是把摘要拆成两类:一段自由文本的对话摘要,加一组结构化的关键事实。
结构化的部分单独存,不进摘要模板,每轮以固定格式注入:
[会话关键事实] - 项目代号:Atlas - 用户角色:后端工程师 - 已确认约束:响应时间 < 2s这些字段的更新不走模型,而是用规则抽取——比如用一个固定的抽取 prompt 单独跑一次,要求输出 JSON,然后和已有字段做合并,冲突时以新值为准。多一次调用,但换来的是关键信息不会丢。对于工单号、订单号、金额、日期这类绝对不能错的字段,这个代价完全值得。
另外摘要模板本身也要改。默认模板给的自由度太大,我一般会写成"保留所有数字、编号、日期和专有名词的原始写法,不要转述"这种硬约束。措辞上要具体,"保留重要信息"这种说法基本没用,模型对"重要"的判断和你不一致。
5.4 序列化与持久化的边界
BaseMemory继承自Serializable,支持导出成 JSON。这在调试时很好用,能直接把某个时间点的记忆状态存下来复现问题。但用 pickle 做持久化要慎重:Memory 对象里可能持有模型客户端、向量库连接这类不可序列化的东西,pickle 会直接抛错,或者更糟——序列化成功了,反序列化回来是个半残的对象,调用时才报错。
我的建议是只序列化数据,不序列化对象。具体来说就是存chat_memory.messages的消息列表和moving_summary_buffer这类纯文本状态,恢复的时候重新构造 Memory 实例再把数据灌进去。这样对象和状态解耦,换版本、换实现都不受影响。JsonFileHistory那段代码就是按这个思路写的——文件里只有type和content两个字段,任何版本的 LangChain 都能读。
还有个和持久化相关的小细节:消息的时间戳和元数据。默认的消息对象不带时间信息,做审计或者按时间范围回溯时会很麻烦。我一般会在add_message里给消息补一个additional_kwargs={"ts": ...},存下来成本很低,用的时候很救命。
6. 旧教程跑不通之后:迁移到新版消息历史接口的完整路径
6.1 ConversationChain 弃用的现实影响
如果你照着一年前的教程写代码,大概率会在导入时看到一条弃用告警,或者干脆在某些版本里找不到ConversationChain。这不是小改动,它反映的是框架层面的方向调整:从"链内置状态"转向"Runnable + 外部历史注入"。改了之后有两个实打实的好处——一是链条可以完全无状态,横向扩容时不用做会话粘滞;二是历史可以按session_id动态路由,多租户场景实现起来自然得多。
迁移的核心在于把三样东西对应上:原来写在 Memory 上的memory_key,现在对应RunnableWithMessageHistory的history_messages_key;原来 Chain 的输入键,对应input_messages_key;原来靠全局变量传的会话标识,现在走config["configurable"]["session_id"]。第三点最容易被忽略,因为它不在函数参数里,忘了传就是一个ValueError,或者更麻烦——传了一个固定的 ID,所有用户的历史混在一起。
6.2 迁移对照表与逐项校验
把常见的对应关系整理成表,迁移时挨个对一遍能省不少时间:
| 旧写法 | 新写法 | 备注 |
|---|---|---|
ConversationChain(llm, memory=...) | prompt | llm | parser+RunnableWithMessageHistory | 链条本身不再持有记忆 |
ConversationBufferMemory() | InMemoryChatMessageHistory | 存储层替代,无压缩逻辑 |
memory.memory_key | history_messages_key | 变量名必须和模板占位符一致 |
chain.predict(input=...) | .invoke({"input": ...}, config=...) | 返回值需要自取.content或用 parser |
| 全局 memory 变量 | get_session_history(session_id) | 路由键通过 config 传入 |
memory.clear() | history.clear() | 清理改在存储层做 |
我特别建议在迁移后做一次"回放测试":把旧系统里的一段真实对话按顺序重放,对比两版输出的差异。因为新接口默认不压缩,如果旧代码依赖了窗口或摘要的裁剪行为,迁移后上下文会变长,token 消耗上升,极端情况下还会超窗口。这个变化不会报错,只会在账单和延迟上体现,回放测试是最容易发现的方式。
6.3 迁移之后怎么补回压缩能力
新版接口干脆利落,但它只给了你存储,没给裁剪。想要窗口或摘要能力,得自己加回来。窗口最简单,在get_session_history里做文章,返回一个带裁剪的历史对象;或者在送入模型前用trim_messages处理一遍,这是最灵活的方式,因为裁剪策略可以按场景动态选。
from langchain_core.runnables import RunnableLambda def trim_history(payload): messages = payload["history"] + [payload["input"]] payload["history"] = trimmer.invoke(messages)[:-1] return payload chain = RunnableLambda(trim_history) | prompt | llm | StrOutputParser() chain_with_history = RunnableWithMessageHistory( chain, get_session_history, input_messages_key="input", history_messages_key="history", )把裁剪做成链条的一环,在这个环节上还能顺手加日志、加异常兜底、按租户切换不同策略。我个人觉得这比塞在 Memory 的load_memory_variables里更清晰,因为裁剪的输入输出都是消息列表,纯函数逻辑,好测也好读。
要把摘要能力补回来,做法是写一个独立的压缩步骤:在save_context之后检查消息总数或 token 数,超阈值就取头部的一段生成摘要,把摘要存进一个单独的字段,原消息删掉。下次组装消息时,把摘要作为一条SystemMessage放在最前面。这套逻辑三十行左右能写完,比依赖一个内部实现会变的类更让人踏实——这也是我在新项目里越来越少直接用内置 Memory 类的原因,抽象层提供了很好的思路参考,但真正上生产的往往是照着思路自己写的那一小段。
最后分享一个我觉得挺有用的小习惯:把每次请求的实际上下文完整落一份到本地,包括系统提示、摘要、历史消息和当前输入,文件按session_id和轮次命名。出问题时不用靠猜,直接翻文件就能看到模型到底看到了什么。这个习惯帮我定位过至少三四个"模型抽风"的问题,每次追下去,问题都不在模型身上,而在上下文拼装的那几行代码里。