1. 多轮对话为什么总在第三轮“失忆”
你大概率遇到过这种场景:第一轮告诉模型“我在做一个 Python 爬虫项目,目标站点需要登录”,第二轮追问“那登录态怎么保持”,模型答得头头是道;到第三轮你问“刚才说的那个项目,用哪种存储方案合适”,它却开始泛泛而谈 Session 和 Cookie 的区别,仿佛前两轮从没发生过。这不是模型变笨了,而是 LLM 本身无状态——每次请求对它来说都是全新的开始,你不把历史消息重新塞进去,它就真的什么都不记得。
历史消息存储要解决的就是这件事:把用户输入、模型输出以及中间交互信息,按角色和顺序结构化保存,在下一轮请求时重新组装成上下文喂给模型。它直接决定三件事——对话连不连贯、token 成本高不高、程序重启后记忆还在不在。很多人第一次做多轮对话,习惯用一个 Python 列表把消息 append 进去,跑单次脚本没问题,一旦上服务、开多用户、要重启,立刻暴露问题:内存里的列表跟着进程一起没了,两个用户的对话还会互相串。
LangChain 的记忆组件(Memory)就是把这套“存什么、存多久、怎么取”抽象成可替换的骨架。它不神秘,本质是帮你管理一个消息列表,并在合适时机把它转成 prompt 的一部分。这篇聚焦工程落地,给你一套可复用的会话持久化骨架:内存态怎么跑通、持久化层怎么接、存储边界和清理策略怎么定。适合已经能调通 LLM API、准备把 demo 变成能长期运行的应用的开发者。下面所有配置都可以直接复制,改掉连接串就能用。
2. TaoToken 前置:把模型调用这层先固定下来
记忆组件负责“存历史”,但历史最终要拼成 prompt 发给模型,所以你得先有一个稳定的模型调用入口。我习惯把模型接入和记忆存储分开看:前者是通道,后者是状态。通道这层用 TaoToken 的 OpenAI 兼容接口就行,不用改 LangChain 的调用习惯。
先在控制台创建一个 API Key,地址是 https://taotoken.net/api-keys ,创建后复制保存,它只显示一次。模型对话的调试入口在 https://taotoken.net/chat ,你可以先在里面手动发几轮消息,直观感受一下“不带历史”和“带历史”的差别,再回到代码里对照。
接入时把 base_url 指向 https://taotoken.net/api ,注意这里不加任何查询参数。环境变量这样配:
export TAOTOKEN_API_KEY="sk-你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"如果你后面要做长期编码或 Agent 类应用,会话轮次多、上下文长,可以了解下 Coding Plan( https://taotoken.net/coding-plan ),它更适合高频、长上下文的场景。接入文档在 https://taotoken.net/doc ,遇到参数对不上时以文档为准。这一步的目标只有一个:让ChatOpenAI能正常返回结果,后面记忆组件才有东西可存。
3. 可复制配置:从内存记忆到持久化骨架
3.1 先跑通内存态,确认记忆真的生效
最小可用版本用ConversationBufferMemory,它保留全量历史。关键点是return_messages=True,这样取出来的是 Message 对象列表,而不是拼好的字符串,方便后续做结构化处理。
import os from langchain_openai import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain.chains import ConversationChain llm = ChatOpenAI( model="gpt-4o-mini", api_key=os.environ["TAOTOKEN_API_KEY"], base_url=os.environ["TAOTOKEN_BASE_URL"], temperature=0.3, ) memory = ConversationBufferMemory(return_messages=True) chain = ConversationChain(llm=llm, memory=memory, verbose=False) print(chain.predict(input="我在做一个 Python 爬虫项目,目标站点需要登录")) print(chain.predict(input="登录态一般怎么保持?")) print(chain.predict(input="刚才说的那个项目,用哪种存储方案合适?"))第三轮如果它能接上“爬虫项目”和“登录态”,说明记忆链路通了。ConversationChain会自动把 memory 里的历史拼进 prompt,你不用手动拼。
3.2 换窗口记忆,控制 token 成本
全量历史在长对话里会迅速膨胀,token 账单和延迟都会涨。ConversationBufferWindowMemory只保留最近 k 轮,超出自动丢弃:
from langchain.memory import ConversationBufferWindowMemory memory = ConversationBufferWindowMemory(k=4, return_messages=True)k=4 表示保留最近 4 轮交互。实测下来,日常问答 k 取 3 到 6 比较平衡;如果是需要远期信息的任务,窗口记忆会丢关键上下文,这时候要么加大 k,要么换摘要式。
3.3 摘要式记忆:用 LLM 压缩历史
ConversationSummaryMemory让模型把历史压成一段摘要,适合长文档讨论这类内容冗长的场景:
from langchain.memory import ConversationSummaryMemory memory = ConversationSummaryMemory( llm=llm, return_messages=True, max_token_limit=800, )注意它每次更新摘要都会额外调一次 LLM,成本和延迟都会上升,别在低延迟场景无脑用。
3.4 持久化骨架:把历史落到 Redis
内存记忆一重启就没了,多用户还会串。持久化要解决的是“跨会话、跨进程、跨设备”。下面这套骨架把消息存进 Redis,用 session_id 隔离不同用户:
import json import redis from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langchain_openai import ChatOpenAI r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True) ROLE_MAP = {"human": HumanMessage, "ai": AIMessage, "system": SystemMessage} def save_message(session_id: str, role: str, content: str, ttl: int = 86400): key = f"chat:history:{session_id}" payload = json.dumps({"role": role, "content": content}, ensure_ascii=False) r.rpush(key, payload) r.expire(key, ttl) # 每次写入刷新过期时间 def load_history(session_id: str, limit: int = 20): key = f"chat:history:{session_id}" raw = r.lrange(key, -limit, -1) messages = [] for item in raw: obj = json.loads(item) cls = ROLE_MAP.get(obj["role"], HumanMessage) messages.append(cls(content=obj["content"])) return messages def chat(session_id: str, user_input: str): history = load_history(session_id) history.append(HumanMessage(content=user_input)) resp = llm.invoke(history) save_message(session_id, "human", user_input) save_message(session_id, "ai", resp.content) return resp.content这套骨架的边界很清晰:Redis 负责存,load_history负责取最近 limit 条,ttl负责清理。rpush追加、lrange取尾部,天然按时间顺序。expire每次写入刷新,意味着活跃会话不会过期,沉默会话到期自动释放,这就是最朴素的清理策略。
3.5 存储边界与清理策略怎么定
存储边界要回答两个问题:存多少、存多久。存多少用limit控制,取最近 N 条拼进 prompt,避免上下文无限增长;存多久用ttl控制,按业务定,客服会话 1 天、个性化服务可以 7 天甚至更久。如果历史很长又不想丢信息,可以定期把早期消息用摘要式压缩成一条 SystemMessage 存回去,既省空间又保留关键信息。多用户隔离靠 session_id,千万别用全局 key,否则 A 用户的对话会出现在 B 用户的上下文里。
4. 验证请求:确认多轮上下文真的保持住了
配置写完必须验证,否则你只是“以为”它记住了。分两步:先验证单会话多轮,再验证跨进程持久化。
单会话验证直接跑 3.1 的三轮对话,观察第三轮是否引用前文。更严谨的做法是把每轮实际发给模型的 messages 打出来:
history = load_history("user_001") for m in history: print(type(m).__name__, "->", m.content[:60])你应该看到 HumanMessage 和 AIMessage 交替出现,顺序正确。如果全是 HumanMessage,说明 role 映射写错了。
跨进程验证:跑一次chat("user_001", "我叫小明"),然后关掉 Python 进程,重新开一个,执行load_history("user_001"),如果还能取到“我叫小明”,持久化就通了。再换一个 session_id 取,应该为空,说明隔离生效。
成功结果长这样:同一 session_id 下,第三轮能准确回答“你之前提到的爬虫项目”;重启进程后历史仍在;不同 session_id 互不可见;沉默超过 ttl 后 key 自动消失。这四点都过,骨架就算跑通了。
5. 本篇常见错排查
报错一:ValidationError: base_url相关。多半是 base_url 写成了带路径或带查询参数的形式。正确写法是https://taotoken.net/api,不要加/v1之外的尾巴,也不要拼 UTM 参数。
报错二:第三轮模型答非所问,历史像没生效。先确认return_messages=True,再确认你用的是ConversationChain或手动把 history 传进invoke。如果你自己拼 prompt 却忘了把 history 加进去,记忆组件存了也白存。
报错三:Redis 里 key 越积越多。检查expire是否每次写入都调用。只在第一次写入设过期、后续rpush不刷新,会导致活跃会话提前过期;反过来完全不设 expire,沉默会话永远不释放。两种都要避免。
报错四:多用户对话串了。九成是 session_id 用了固定值或全局变量。每个请求都要从登录态或前端传入独立 session_id,服务端不要缓存成全局。
报错五:摘要式记忆越用越慢。ConversationSummaryMemory每次更新都调 LLM,轮次多了延迟明显。长对话建议“窗口 + 定期摘要”混合:近期用窗口保原文,早期压缩成摘要。
报错六:token 超限。load_history的 limit 太大,或窗口 k 太大。先算一下单条消息平均 token,再反推 limit。一般留出模型输出空间,输入别超过上下文窗口的 70%。
6. 把骨架接进你的应用
这套骨架的价值在于可替换:Redis 换成 PostgreSQL 或 MySQL,只要改save_message和load_history两个函数,上层chat逻辑不动。想加向量检索做长期记忆,就在load_history里先按语义召回相关历史,再拼进上下文。想控制成本,就把窗口记忆和摘要记忆组合起来用。
下一步建议你先把 3.4 的代码跑通,用两个不同 session_id 各发几轮,确认隔离和持久化都对。然后去 https://taotoken.net/api-keys 确认你的 Key 额度正常,接入细节对照 https://taotoken.net/doc 。如果要做长期编码或 Agent,会话轮次多、上下文长,可以看 https://taotoken.net/coding-plan 。调试模型行为时,https://taotoken.net/chat 能帮你快速对比“带历史”和“不带历史”的输出差异,比在代码里反复改快得多。