最近连着做了几个基于Langchain的对话应用,聊天机器人、文档问答、客服助手都有涉及。做得多了就发现,绝大多数需求绕不开同一个坎,也就是会话记忆。没有记忆的对话,就是一段段彼此孤立的提问,前面说过什么模型一点都不知道,体验非常割裂。这篇博文是我学习Langchain会话记忆的第五篇记录,重点拆解临时记忆与长期记忆两个层面:临时记忆解决单次会话内的上下文连贯,长期记忆解决跨会话、跨时段的身份与偏好识别。我会从原理、代码实践、参数细节到踩坑经验完整过一遍,适合刚接触Langchain的开发者,也适合想在生产环境里把记忆模块做扎实的同学。
1. 会话记忆:为什么对话系统必须自带“记事本”
1.1 没有记忆的对话体验有多崩溃
先聊一个很直观的场景。你拨通客服电话,把问题说了一遍,对方让你等一会再描述一遍,好不容易说完了,转接给下一个人,你又得从上到下重新讲一遍。这种体验放在AI对话里,就是最典型的“无记忆”状态。
很多刚开始做Langchain的人会有个误解:大模型本身不是已经“记得”我上一句说了什么吗?其实不是。模型接口本质上是一个无状态函数,每一次请求接收一组消息,返回一组回答,请求结束之后这组消息就没了。所谓“上一轮说过什么”,完全靠应用层自己去缓存、拼装,再在下一轮请求里重新发给模型。Langchain里的Memory模块,做就是这件事:历史消息的存储、格式化和注入。
我遇到过真实的项目事故。某个内部知识问答机器人,上线第一版只接了文档检索,完全没有会话记忆。用户问“它的使用限制是什么”,得到回答后,再追问“那它的替代品呢?”,模型根本不知道“它”指的是什么,回答得牛头不对马嘴。后来把记忆模块补上,体验立刻就不一样了。这也印证了一件事:记忆不是优化项,而是对话系统的刚需。
1.2 临时记忆与长期记忆到底怎么分工
在Langchain的生态里,会话记忆通常被分成临时记忆和长期记忆两个层面。很多人会把它们混着用,一提到记忆就想着往向量数据库里塞,结果越做越重。其实先想清楚分工,后面做起来会顺很多。
临时记忆可以理解成“工作记忆”,它的生命周期紧跟着当前会话。用户进入聊天页面,开始一轮对话,这期间的上下文由临时记忆负责;会话结束或者服务重启,临时记忆一般就清空了。它解决的核心问题是“这一轮对话里,前文都说了什么”,典型实现是ConversationBufferMemory、ConversationBufferWindowMemory、ChatMessageHistory这类组件。
长期记忆解决的是另一类问题:用户第二天又来了,AI还记不记得这个用户上次聊过什么偏好、确认过什么事实。这种记忆需要持久化存储,至少要做到进程间共享,甚至跨天、跨周级别地保留。它的生命周期不跟随某一次会话,而是跟随用户、跟随业务实体。典型实现是外部存储(比如Redis、数据库)加向量检索。
我习惯用一个生活化的比喻:临时记忆是便利贴,贴在桌面上,用完就扔;长期记忆是笔记本里的档案页,按用户归档,下次见面还能翻出来。两者并不互斥,真正靠谱的实战方案几乎都是组合使用。
1.3 本篇适合谁、能收获什么
如果你正在做聊天机器人、文档问答、客服助手这类需要多轮上下文的项目,这篇博文应该能帮你把记忆模块从“能跑”提升到“好用”。我会先讲临时记忆最常用的几种组件和参数细节,再讲长期记忆的两种落地方案,最后整理一些实战中很容易踩的坑。读完你至少能回答这几个问题:临时记忆和长期记忆分别用什么组件实现;Buffer、Window、Summary之间怎么选;向量检索式的长期记忆怎么和对话流程配合。
2. 临时记忆:让单轮对话连成一段连贯交流
2.1 最省事的BufferMemory到底怎么工作
临时记忆里最容易上手的是ConversationBufferMemory。它做的事情非常朴素:把历史对话全部存下来,下一轮对话的时候,把所有历史拼成一段提示词,一起发给模型。
我用一个简化示例说明。为了不绑定某个具体模型服务商的SDK,下面代码里的get_llm()是我项目里的一个封装,返回已经配置好的对话模型客户端,你换成自己项目里实际调通的那个即可。
from langchain.chains import ConversationChain from langchain.memory import ConversationBufferMemory llm = get_llm() # 替换为你项目里已配置好的模型客户端 memory = ConversationBufferMemory() conversation = ConversationChain(llm=llm, memory=memory) print(conversation.run("我叫小明,住在杭州。")) print(conversation.run("我是哪里人?"))第一轮运行结束之后,memory里实际保存的内容类似这样:Human: 我叫小明,住在杭州。AI: ...。第二轮运行前,ConversationChain会把这段历史取出来,和当前输入“我是哪里人?”一起组装成提示词,再交给模型。模型看到前文说过“住在杭州”,自然就能答出来。
这里有几个细节要留意。ConversationChain内部默认的提示词模板里,历史变量名固定是history,所以ConversationBufferMemory的memory_key默认也是history。如果你自己写prompt模板,变量名必须和memory_key对齐,这一点在第四节会重点讲。
BufferMemory的好处是逻辑简单、信息无损耗,但它有个致命缺陷:历史越长,提示词越长,模型花费和响应时间也跟着涨。实测下来,当对话超过三十轮,提示词里大部分内容可能都是早期无关紧要的闲聊,既占上下文窗口,又干扰模型聚焦当前问题。这也是为什么实际项目很少直接裸用无限Buffer。
2.2 BufferWindowMemory:限制上下文长度的实用方案
如果你既想要临时的上下文连贯,又不想让提示词无限膨胀,ConversationBufferWindowMemory是最直接的方案。它的做法是只保留最近k轮对话,更早的历史直接丢弃。
from langchain.memory import ConversationBufferWindowMemory memory = ConversationBufferWindowMemory(k=6) conversation = ConversationChain(llm=get_llm(), memory=memory)这里的k非常容易理解错。它不是保留最近6条消息,而是保留最近6轮“用户输入-模型输出”的来回。你可以理解成6个回合,对应12条消息(6条用户消息、6条AI消息)。至于具体触发多少次对话后会开始截断,取决于你调用run的轮数。比如做了10轮对话,k=6,memory里就只剩下最后6轮,前面4轮的内容不会再出现在提示词中。
我在实际项目中,k的值一般从10到20之间取。如果任务本身对前文依赖很强,比如用户会反复修正某个要求,k可以调大一些;如果只是普通问答,k小一点反而效果更稳定,因为提示词更短、噪音更少、响应更快。一个常见的坑是:用户在第一轮说过“我叫小明”,聊到第十五轮你问“我叫什么”,窗口已经把它挤掉了,模型当然不知道。这种关键信息丢失问题,不是调大k就能根本解决的,正确的做法是把这类事实放到长期记忆里,而不是指望短期窗口一直兜住。
2.3 ChatMessageHistory:更底层、更灵活的消息容器
ConversationBufferMemory和WindowMemory都是高层封装,它们底层其实都在操作一个叫ChatMessageHistory的消息容器。如果你需要更细粒度地控制消息,而不是跟字符串模板打交道,可以直接上手操作这个类。
from langchain.memory import ChatMessageHistory history = ChatMessageHistory() history.add_user_message("你好,请帮我查一下订单状态。") history.add_ai_message("好的,请提供你的订单号。") history.add_user_message("订单号是ABC123。") history.add_ai_message("已查到,您的订单预计明天送达。")每个add_user_message和add_ai_message调用,都会往history里追加一条带角色的消息。这个结构的好处是,它天然符合大模型接口的message格式,很多场景下你可以直接把这个列表传给模型的predict_messages接口,而不需要拼字符串。
我自己在自定义chain的时候,比较喜欢用ChatMessageHistory作为“原始消息来源”,再结合ConversationBufferMemory的chat_memory参数来装配。举个例子:做一个客服机器人,前几轮是固定的话术确认,这部分可以手动写入history,之后再进入真正的对话循环。这样的话术段就不会被误当成需要模型生成的普通对话。
另外要注意,直接修改history时如果中途出错,容易造成数据不一致。我一般会把对history的追加操作集中在一个方法里,同时在写入前做校验,避免把空消息或者重复消息塞进去。
2.4 临时记忆的适用边界
临时记忆虽然好用,但它只负责“当前会话”,这是一个容易忽略的边界。如果你把它当成全局记忆来用,会遇到两个典型问题。
第一,进程内存不落盘。服务一重启、容器一重建,临时记忆里所有内容全部清空,用户下次再来就是“陌生人”。在开发环境里这个问题不明显,因为本地进程一直活着;但一旦上了生产的容器调度,重启是很频繁的事。
第二,上下文窗口终究有限。哪怕用窗口记忆把长度控制在20轮,它依然只解决“最近聊了什么”,解决不了“用户一直以来的偏好是什么”。比如一个老用户说“还是老规矩”,模型需要知道他上次默认的偏好是“货到付款”,而这条信息可能藏在3天前的对话里。这已经不是临时记忆能承担的任务,该上长期记忆了。
3. 长期记忆:让AI跨会话仍然“记得你是谁”
3.1 思路从“带全历史”变成“按需检索”
长期记忆在工程上有一个关键的思路转变:不要试图把所有历史从头到尾都塞进提示词,而是把对话中产生的“有效信息”抽取出来,持久化保存,下次需要时再按需检索并注入。
为什么不能简单地把整段历史存起来、下回继续拼进去?首先是成本问题,长期积累的对话可能有几百上千轮,一次请求根本放不下;其次是效率问题,无关的旧信息会稀释模型对当前问题的注意力,答案质量反而变差。所以长期记忆的核心是“沉淀信息”和“按需召回”,而不是“无脑堆历史”。
3.2 方案一:对话记录持久化,下次接着聊
最简单的长期记忆,就是把聊天记录持久化到外部存储。Langchain提供了RedisChatMessageHistory,使用起来非常直接。
from langchain.memory import RedisChatMessageHistory from langchain.memory import ConversationBufferMemory history = RedisChatMessageHistory( session_id="user_001", url="redis://localhost:6379" ) memory = ConversationBufferMemory(chat_memory=history)这里面的关键设计点是session_id。在真实业务中,它不应该是一个简单的字符串,而是要和你的用户体系绑定,比如user_001、order_888、tenant_xx_user_yy。这样同一个用户在不同时间、不同设备上发起的新会话,都能用同一个session_id找到历史记录。
我最初做这个方案时,把session_id设成了会话ID而不是用户ID,导致用户每次刷新页面都开启新会话,历史全部丢失。后来改成用户ID作为维度,才真正做到了跨会话记忆。
如果不方便引入Redis,也可以用本地方案。Langchain生态里有接口可以对接本地存储,核心是自定义一个历史存储类,实现保存和加载两个动作。比如用SQLite存消息:
import sqlite3 import json from langchain.memory import ChatMessageHistory class SQLiteChatHistory(ChatMessageHistory): def __init__(self, session_id, db_path="memory.db"): self.session_id = session_id self.db_path = db_path self.conn = sqlite3.connect(db_path) self._create_table() self._load() def _create_table(self): self.conn.execute(""" CREATE TABLE IF NOT EXISTS chat_history ( session_id TEXT, role TEXT, content TEXT ) """) self.conn.commit() def _load(self): cur = self.conn.execute( "SELECT role, content FROM chat_history WHERE session_id=? ORDER BY rowid", (self.session_id,) ) self.messages = [] for row in cur.fetchall(): if row[0] == "human": self.add_user_message(row[1]) else: self.add_ai_message(row[1]) def save_message(self, role, content): self.conn.execute( "INSERT INTO chat_history (session_id, role, content) VALUES (?, ?, ?)", (self.session_id, role, content) ) self.conn.commit()这段代码我简化过,重点表达两个思路:一是消息永久落盘,二是用session_id作为索引。注意我这里save_message需要在你每次向模型发送消息之后手动调用,如果你用的是ConversationChain,最省事的做法是重写或者封装save_context,让它同时更新内存和SQLite。
持久化方案适合“接着上一次聊”的场景,但它也有明显的局限:当历史记录极度膨胀时,你还是得考虑怎么截断、怎么摘要,否则每次加载全部历史,问题依然存在。所以生产级长期记忆,最常用的是下面的向量化检索方案。
3.3 方案二:向量化长期记忆,按语义召回
向量化记忆的思路,是把对话中产生的“事实片段”转成向量,存进向量数据库,每次对话开始前用当前输入去检索语义相关的记忆,把命中的内容注入提示词。
为什么用向量检索而不是SQL模糊查询?因为记忆的匹配是语义层面的。“我住在杭州”和“你住在哪个城市”没有共同的关键词,但语义高度相关。向量检索能通过语义相似度找到这种关联,这是关键字匹配做不到的。
我实现过一个小型的“用户画像记忆”机器人,大致流程是这样的:每一轮对话结束后,判断用户有没有说出值得长期保存的事实(比如姓名、城市、偏好、身份),有的话抽取成一句自然语言文本,向量化后存入向量库;下一轮对话开始时,用用户当前输入去向量库检索,把命中的记忆拼进提示词。
伪代码结构如下:
from langchain.vectorstores import FAISS from langchain.memory import ConversationBufferMemory class LongTermMemoryBot: def __init__(self, embedding_model): self.embedding_model = embedding_model # 如果已经有历史向量库,加载;否则新建 self.vectorstore = None # 短期记忆负责当前会话的上下文 self.short_memory = ConversationBufferMemory(return_messages=True) def _recall(self, query, k=3): if self.vectorstore is None: return [] docs = self.vectorstore.similarity_search(query, k=k) return [d.page_content for d in docs] def _remember(self, fact): if self.vectorstore is None: self.vectorstore = FAISS.from_texts([fact], self.embedding_model) else: self.vectorstore.add_texts([fact]) def run(self, query): memories = self._recall(query) memory_block = "\n".join(memories) if memory_block: prompt = f"已知的用户信息:\n{memory_block}\n\n当前问题:{query}" else: prompt = query # 这里把prompt发给模型,并拿到回答 answer = get_llm().invoke(prompt) self.short_memory.save_context({"input": query}, {"output": answer}) return answer这里面有几个工程细节很关键。第一,每次对话结束后不但要保存短期记忆,还要做一个“事实抽取”动作。可以基于规则(比如检测“我叫”“我住在”“我喜欢”等句式),也可以用大模型来抽取。我用的是模型抽取,每轮结束后让模型判断“用户透露了哪些值得长期记住的信息”,抽取出来再写入向量库,这样比规则通用得多。第二,向量库的写入需要做去重合并。用户如果改了城市,不能把新旧两条都存进去,否则检索时会产生冲突,模型反而不知道以哪条为准。我通常让模型抽取时带上“事实ID”,相同ID的信息采用覆盖式写入。第三,similarity_search的k值一般取3到5,这是召回精度和提示词长度的平衡点,取多了提示词里全是重复信息,取少了又可能漏掉关键事实。
3.4 摘要记忆:压缩历史的折中方案
在临时记忆和长期记忆之间,其实还有一类折中方案:ConversationSummaryMemory。它的思路是用模型实时把历史对话总结成一段摘要,然后只保存摘要,不再保存完整历史。
from langchain.memory import ConversationSummaryMemory memory = ConversationSummaryMemory(llm=get_llm()) conversation = ConversationChain(llm=get_llm(), memory=memory)每一轮对话之后,SummaryMemory都会让模型更新一次摘要。随着对话进行,摘要会越来越长,但总体来说它压缩了大量原始对话,成本远低于全量Buffer。适合那种对话特别长、预算可以承担额外模型调用、关注整体语义多于逐字细节的场景。
它的缺点也很明显:细节会丢。用户在某轮说过一个具体订单号,摘要可能只会写着“用户咨询订单问题”,细节信息直接消失。所以我的建议是,摘要记忆适合做“整体背景”,不适合做“关键事实存储”。关键事实还是放到向量化长期记忆里更可靠。
有一个经验分享:实际项目中,摘要记忆可以当成长期记忆的“入口”。每轮对话结束,如果发现对话已经很长,先把摘要写出来,再把摘要里的关键事实向量化存入向量库。这样既控制了成本,又让长期记忆有稳定的信息输入来源。
4. 常见问题、踩坑记录与搭配实践
4.1 版本差异导致API飘忽不定
Langchain的版本更新非常快,不同版本之间Memory类的位置、构造函数、方法名都可能不一样。我自己踩过的典型例子:某个项目的调用了ConversationBufferMemory.load_memory_variables拿历史,换了一个版本之后,这个方法的返回结构变了,提示词拼装直接报错。
遇到这种情况,处理原则很简单:不要背API,去看你当前安装版本对应的官方文档和源码。写记忆相关代码时,重点关注几个位置:memory_key的默认值、load_memory_variables的返回值结构、save_context接收的键名。这些地方只要有一个不对齐,对话链就会要么报错、要么静默丢失上下文。
我建议在项目初始化时,先写一个20行左右的“记忆自检脚本”:模拟两轮对话,打印memory里实际存了什么。这样升级依赖后,自检脚本一跑就知道有没有问题。
4.2 memory_key与history注入位置错误
自定义Chain或者自定义PromptTemplate时,最容易踩的坑是变量名对不上。默认情况下,ConversationChain里的提示词模板有{history}占位符,ConversationBufferMemory的memory_key也是history。但如果你自己写模板,把历史变量叫成了{chat_history},就必须显式指定memory_key:
from langchain.prompts import PromptTemplate from langchain.memory import ConversationBufferMemory memory = ConversationBufferMemory(memory_key="chat_history") prompt = PromptTemplate( input_variables=["input", "chat_history"], template="以下是历史对话:\n{chat_history}\n\nHuman: {input}\nAI:" ) conversation = ConversationChain( llm=get_llm(), prompt=prompt, memory=memory )这里的规律是:模板里写什么占位符,memory_key就得叫什么;同时input_variables里也要补上这个变量名。我见过很多次报错“Prompt got unexpected keys: history”,本质就是memory_key和模板占位符不一致。
还有一处容易忽略:当你手动调用memory.save_context({"input": query}, {"output": answer})时,传入的键名必须和对话链的输入输出变量名一致。如果ConversationChain里的输入变量叫input,你这么传就没问题;如果你在模板里把输入变量改成了question,这里也得改成{"question": query}。
4.3 token失控与性能下降
临时记忆最大的隐性成本是token。每个请求都会把历史重新发给模型,模型按token计费,历史越长花费越高。尤其是中文场景,一段中文对话转换成token之后占用量往往比英文更大,这个现象很容易被忽略,很多人在英文示例里调好的k值,换到中文就变得又贵又慢。
我处理这个问题的经验是三层控制。第一层用窗口记忆,限制最近N轮;第二层用摘要记忆,把更早的内容压缩成背景;第三层做关键事实抽取,把真正需要长期保存的信息向量化。这三层配合下来,提示词长度基本能被控制在一个稳定范围,不会随对话轮数无限增长。
还有一个小建议:如果你发现单轮响应延迟明显上升,第一步不是加显卡、加并发,而是先打印一下当前提示词的长度。八成情况是历史积累太多了,裁掉一半无关历史,速度立刻恢复。
4.4 临时与长期记忆的最优组合方案
经过几次项目迭代,我现在比较固定的组合方案是这样的:短期记忆用窗口记忆保持最近10轮对话,负责保证当前会话流畅;长期记忆用向量库保存用户的关键事实,负责跨会话识别用户;如果对话特别长再叠加摘要记忆,摘要结果也向量化入库。
整个流程跑起来是这样:
- 用户发起新对话,系统先用当前输入去向量库做一次相似度检索,召回3到5条相关记忆;
- 把召回到的记忆作为“已知信息”注入提示词,再和短期窗口历史一起发给模型;
- 模型返回回答后,先更新窗口记忆;
- 再用一个抽取模型判断这轮是否产生了值得长期保存的事实,有则写入向量库。
这个方案的优点是把两类记忆的职责分得很清楚:窗口记忆管“刚才说了什么”,向量记忆管“这个用户是谁”。就算服务重启,用户下次再来说一句“还是老规矩”,只要召回记忆里存了“上次默认货到付款”,模型就能理解。
记得给长期记忆加“遗忘机制”。用户改了收货城市,旧记录如果还留在向量库里,检索时可能同时命中新旧两条,模型会困惑。我常用的做法是给每条事实带一个字段时间戳或者覆盖标记,写入新事实时,按照事实ID把旧向量删掉,保证同一类事实只有最新一条。没有遗忘机制的记忆系统,时间越久越容易自相矛盾。
4.5 中文场景下的embedding与检索细节
如果你做中文对话系统,向量化长期记忆时embedding模型的选择很关键。不同模型的检索质量差距非常大,我建议不要只看宣传指标,而是用一个包含常见口语表达的测试集实际跑一轮检索,看召回结果是不是你想要的。特别注意一些简称、口头语,比如用户说“老地方见”,好的embedding模型应该能召回“我们常去的那家咖啡店”相关记忆,差的模型可能直接召回一堆无关内容。
另外,FAISS这类向量库在写入前最好对文本做一定清洗。不要直接把整段对话塞进去,一条记忆最好是一句信息密度高的短句,比如“用户住在杭州”“用户喜欢靠窗座位”,而不是“用户说:你好,我住在杭州,离地铁站很近,旁边有个大商场”。短句的向量更聚焦,检索命中率更高。保存前做一次合并,把同一个用户的相似信息聚到一起,也能减少向量库体积、提高检索准确度。
最后聊一个调试小技巧。我在开发阶段,习惯在每次对话前把“本次注入的长期记忆”和“短期历史长度”打印出来:
print("--- recalled long-term memory ---") for mem in memories: print(mem) print("--- short-term memory variables ---") print(short_memory.load_memory_variables({}))这样能看到模型实际看到了什么。大部分记忆丢失问题,其实肉眼一看打印结果就能定位:要么长期没召回,要么短期窗口被挤爆。这个小习惯帮我省掉了大量盲猜时间。会话记忆体系其实没有太多玄学,核心就是把存储边界划清楚,把注入规则做严谨,剩下的问题自然就少了。希望这篇记录能帮你在Langchain项目里少踩几个我踩过的坑。