对话记忆
本章配套可运行示例:demo/demomemory.py。demo 从「手动维护 messages 列表」起步,一路跑到多会话隔离、trimmessages 开窗、摘要压缩、JSON 原子化持久化;本机没启动 Ollam…
本章配套可运行示例:demo/demo_memory.py。demo 从「手动维护 messages 列表」起步,一路跑到多会话隔离、
trim_messages开窗、摘要压缩、JSON 原子化持久化;本机没启动 Ollama 时自动降级——店员回复换成内置剧本、摘要换成截断式假摘要,历史管理逻辑全程照样跑。本章你将学会
- 记忆的真相:模型没有记性,
messages列表才是记性——每轮对话都是把整本记录重新递给模型- 手动维护历史的三步循环:入账 → 调用 → 回账,以及把它封装成三行代码的
chat()函数- 多会话管理:
dict[session_id, list],别让小雅和小周的记录本串台trim_messages:按条数或 token 数开一扇滑动的窗,include_system保人设、start_on保合法开头- 长程记忆:手写摘要链,把几十条旧消息压成一张「便签」,渐进式摘要的失真怎么防
- 持久化:
messages_to_dict/messages_from_dict+ 临时文件 +os.replace原子写入- 第三方存储:什么时候该把记忆从 JSON 文件搬进 Redis / Postgres(概念)
- 一张「老 API → 1.x 新做法」对照表——看老教程时看到
ConversationBufferMemory不再迷路
第 08 章结尾留了一个缺口:读者每次提问都是孤立的。「接着刚才说的,那借期呢?」——RAG 链答不上来,因为上一轮聊过什么,它一无所知。本章补上这块拼图。
比喻:拾光书屋有位店员,业务水平一流,但记性只有七秒——你上句话说完,他已经忘了。幸运的是,他有个好习惯:只要开口前有人把「对话记录本」摊在他面前,他就能接着上文聊得头头是道。所谓「模型有记忆」,从来不是模型自己记住了什么,而是你的程序每次都把整本记录重新递过去。本章讲的全是这本记录本的学问:怎么记(列表)、怎么翻(裁剪)、怎么缩(摘要)、怎么存(持久化)。
一条主线看全貌:
用户发言 ──► messages.append(HumanMessage) 入账
│
▼
model.invoke(messages) 整本记录递给模型
│
▼
messages.append(reply) 回账
│
┌───────┴────────┐
▼ ▼
记录太厚了? 进程要重启了?
trim_messages save_messages(JSON)
(窗口裁剪) (原子写入)
│
▼
窗口丢信息?
摘要压缩(旧消息 → 便签)
技术基线:Python + LangChain 1.x(langchain + langchain-ollama),本地 Ollama 跑 qwen3:4b。没装的话回 第 01 章看环境准备。
后面示例共用一份起步代码:
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from langchain_ollama import ChatOllama
model = ChatOllama(model="qwen3:4b", temperature=0, reasoning=False)
system = SystemMessage("你是拾光书屋的店员,语气亲切,回答简洁。")
1. 记忆的真相:模型没记性,列表才是记性
比喻:把大模型想成一位「按页收费」的超强临时工。你递一页纸,他看一页纸——纸上是五分钟前聊的还是五年前聊的,对他毫无区别;纸上没有的内容,他就当没发生过。messages 列表就是你递过去的这沓纸。
作用:从根上理解「对话历史 = 消息列表」。模型每次调用都是无状态的:上一轮你告诉它会员号 2048,这一轮不递,它就不知道。所以「记住」的唯一办法,就是每一轮把从对话开始到现在的所有消息,原封不动地重新递一遍。
签名(invoke 的输入类型,记忆能跑起来的关键):
# BaseChatModel.invoke 接受三类输入:
model.invoke("你好") # 1. 纯字符串(当成一条 human 消息)
model.invoke(HumanMessage("你好")) # 2. 单条消息
model.invoke([system, HumanMessage("你好")]) # 3. 消息列表 ← 对话记忆走的是这扇门
消息列表里的常驻角色(第 02 章的旧相识):
SystemMessage:店员守则——人设、规矩,一般放列表开头HumanMessage:用户发言AIMessage:模型回复
最小示例:
from langchain_core.messages import AnyMessage, HumanMessage
from langchain_ollama import ChatOllama
model = ChatOllama(model="qwen3:4b", temperature=0, reasoning=False)
# 一切记忆,从这一个列表开始
messages: list[AnyMessage] = []
# 第 1 轮
messages.append(HumanMessage("你好,我是小雅,刚办了拾光书屋的年卡,会员号 2048"))
reply = model.invoke(messages) # 把「到目前为止的整本记录」递给模型
messages.append(reply) # invoke 的返回值就是 AIMessage,直接归档
print(reply.content)
# 第 2 轮:不传字符串,传整个列表
messages.append(HumanMessage("你知道我的会员号吗?"))
print(model.invoke(messages).content) # 能答上 2048——不是它记性好,是记录本递全了
注意第 1 轮那个顺手的小细节:model.invoke(messages) 的返回值本身就是 AIMessage,直接 append 回列表即可,不需要再手动包一层。这是 Python 版比其他语言版本省事的地方之一。
顺带一提:model.stream(messages) 的输入规则完全相同——第 04 章学的流式本领对「带记忆的对话」照常生效,把 invoke 换成 stream、把分片拼成完整回复再归档就行。
常见坑:
- 忘了回账:调完
invoke拿到回复,只顾着 print,忘了messages.append(reply)。下一轮模型立刻「失忆」——这是新手对话代码排第一的 bug。 - 把历史拼成一大坨字符串:
"用户:我叫小雅\nAI:你好\n用户:……"这种手工拼接丢掉了消息结构,system / human / ai 的角色信息全糊在一起,模型对格式的敏感度远超你的想象。消息就老老实实用消息对象装。 - 在循环里重建列表:
messages是列表引用,循环体里手滑写了messages = [],前面攒的历史瞬间清零。 - 以为历史会自己变长:不会。你不
append,列表永远停在原地;你不传列表,模型永远只看到当前一句。
2. 封装对话循环:三行代码的「自动记忆」
比喻:上一节的三步循环——入账、调用、回账——就像每次泡茶都要现烧水、现洗杯、现抓茶叶。在家这么讲究没人管你,但店员一天要接待几十位会员,这套动作必须变成肌肉记忆:写一个函数,泡茶变按键。
作用:把「追加用户消息 → 调用模型 → 追加回复」封装成一次调用。很多老教程会把这一步交给框架组件(把整套历史自动注入链的包装器),LangChain 1.x 的态度变了:这点活儿自己写比学一套旧 API 更快,也更透明——出了问题,每一行都是你自己的,看得见摸得着。
签名(自己动手写的「自动记忆」):
def chat(messages: list[AnyMessage], user_input: str) -> str:
"""一轮对话:入账 → 调用 → 回账。"""
messages.append(HumanMessage(user_input))
reply = model.invoke(messages)
messages.append(reply)
return reply.content
三行,没有魔法。传进去的是列表本身(引用),函数内的两次 append 都落在同一个本子上——调用方不需要接收返回值来「更新历史」。
最小示例——书屋同时接待两位会员,记录本各管各的:
sessions: dict[str, list[AnyMessage]] = {}
def get_history(session_id: str) -> list[AnyMessage]:
"""取该会员的对话记录本;第一次来,发一本新的。"""
if session_id not in sessions:
sessions[session_id] = []
return sessions[session_id]
# 小雅的会话
xiaoya = get_history("member-2048")
print(chat(xiaoya, "上周借的《山月记》什么时候到期?"))
# 小周的会话,和小雅完全隔离
zhou = get_history("member-1024")
print(chat(zhou, "推荐一本适合通勤路上读的书"))
# 同一会话接着聊,历史自然延续
print(chat(xiaoya, "那我续借一次可以吗?"))
常见坑:
- 全员共用一个列表:把
sessions写成全局单个messages,小雅刚说完会员号,小周来问「我是谁」,模型把她当成小雅。多用户场景,会话隔离是底线——哪怕只是个dict。 - 函数内
return messages式的画蛇添足:列表是引用传递,返回它没有意义;真正会出 bug 的是函数内对列表整体赋值(messages = [...]),那会切断引用。 - 并发写同一本记录:两个请求同时往一个
list里append,顺序会乱(甚至触发 CPython 里罕见的异常)。单进程小流量随便用;真的要上并发,先加锁,或者直接上第 6 节的数据库。 - 别去找「自动注入历史」的旧组件了:老教程里那套「给链包一层、自动读写历史」的包装器,在新版里已经被官方归入遗留路线,新项目不再推荐。它的两个必填参数——「用户输入字段名」和「历史占位符字段名」——曾经是新手重灾区(名字对不上就静默失败)。手写方案里这两个坑天然不存在:历史就摊在列表里,没有占位符要你对齐。
3. trim_messages:给记忆开一扇滑动的窗
比喻:记录本越写越厚,翻起来越来越慢,直到某天递给模型时超了上下文窗口,直接被拒收。滑动窗口的做法很朴素:白板只有这么长——写满一行,最上面的一行就被顶出去,桌上永远只摊最近几行。旧的一行不是被撕掉了,只是不摊给模型看了。
作用:trim_messages 是 LangChain 1.x 官方维护的核心工具,把消息列表裁到预算以内:从尾部往前装(strategy="last",保最近的),装不下的旧消息直接丢弃。它是老教程里「窗口记忆」的现成替代品。
签名:
from langchain_core.messages import trim_messages
def trim_messages(
messages: Iterable[MessageLikeRepresentation] | PromptValue,
*,
max_tokens: int, # 预算上限(含义随 token_counter 变)
token_counter: Callable[[list[BaseMessage]], int]
| Callable[[BaseMessage], int]
| BaseLanguageModel
| Literal["approximate"],
strategy: Literal["first", "last"] = "last", # 从哪头保(记忆用 last)
allow_partial: bool = False, # 装不下的那条要不要拦腰截断
include_system: bool = False, # system 是否无条件保留
start_on: str | type[BaseMessage] | None = None, # 强制窗口以谁开头
end_on: str | type[BaseMessage] | None = None,
) -> list[BaseMessage]: ...
token_counter 是计数器,决定「预算」按什么数。三档常用姿势:
# ① 按条数:一条消息算 1,max_tokens 的含义变成「最多留几条」
trim_messages(messages, max_tokens=4, token_counter=len)
# ② 近似 token 数(推荐日常使用):对中英混合文本做粗略估算,不依赖模型
trim_messages(messages, max_tokens=800, token_counter="approximate")
# ③ 精确计数:把模型实例传进去,用它自带的分词器数(最准,也最慢)
trim_messages(messages, max_tokens=800, token_counter=model)
最小示例:
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, AIMessage, trim_messages
system = SystemMessage("你是拾光书屋的店员,语气亲切,回答简洁。")
messages: list[AnyMessage] = [system]
for i in range(10): # 假装聊了 10 轮
messages.append(HumanMessage(f"第 {i + 1} 个问题:书屋有什么新书?"))
messages.append(AIMessage(f"第 {i + 1} 个回答:这周新到一批散文集。"))
# 窗口记忆:只留最近 4 条,但 system 保住、且窗口从 human 开头
window = trim_messages(
messages,
max_tokens=4, # 配合 len 计数器 = 最多 4 条
token_counter=len,
strategy="last", # 保最近的
include_system=True, # 店员守则不能丢
start_on="human", # 窗口第一条必须是用户消息(见下方坑 2)
)
print(len(window)) # 3(system + 最近 1 组问答,见下方实测说明)
# 用窗口替代全量历史去调用
# reply = model.invoke(window)
上例有个值得掰扯的细节(实测):21 条历史裁完只剩 3 条,不是直觉里的「4 + 1」。因为 token_counter=len 眼里 system 也占一个名额,预算 4 个位置被 system、A10、H10、A9 分掉,得到 [S, A9, H10, A10];start_on="human" 再把开头的 A9 掐掉,最终 [S, H10, A10]。换句话说,system 占预算、start_on 会额外丢一条——想留满 4 条正文,预算要开到 5 以上。
include_system=True 和 start_on="human" 这两个参数为什么重要,看下面前两个坑——都是实测踩过的。
常见坑:
include_system默认是False,裁着裁着人设没了。实测:带 system 的 6 条历史,max_tokens=40裁完,默认参数下结果以AIMessage开头且 system 消失——店员守则被当旧账扔了,模型下一轮秒变「路人」。带上人设的对话,include_system=True几乎是必选项。- 裁完的列表可能以
AIMessage开头——非法历史。实测:5 条消息[H, A, H, A, H]按 token 裁剪,结果可能是[A, H, A, H]。多数聊天模型要求历史以 human 开头(或 system + human),AI 冷不丁起头会让部分模型直接报错、部分模型行为怪异。start_on="human"一劳永逸——它会把开头的 AI 消息再丢一层,保证第一条(不算 system)是用户消息。 allow_partial=True会把消息拦腰截断。预算差一口气时,它允许把装不下的那条切一半塞进去:模型看到半句话,比看不到更迷惑。除非你明确知道自己在干什么,保持默认False。token_counter=len时,max_tokens名字是骗人的——它数的是条数不是 token。代码评审时看到max_tokens=4别慌,先看token_counter是谁。trim_messages返回新列表,不动原件。想真正替换历史,要显式messages[:] = window(切片赋值,保持引用不变);直接messages = window会让外部引用指向旧列表,白裁了。
4. 长程记忆:把旧对话压成一张便签
比喻:滑动窗够用,但有个天生缺陷——它是个「越新越好、越旧越忘」的家伙。会员号是第一轮说的,聊到第 20 轮早被顶出窗外,等她问「我的会员号是多少」,店员一脸茫然。长程记忆的解法像给白板贴便签:聊了十几页之后,把前面十几页开个「会议纪要」,压成一张便签贴在白板最上方——「此会员:小雅,会员号 2048,偏好短篇,正在等《山月记》到期提醒」。原稿归档收起,桌上只留便签 + 最近几页。信息量缩了十倍,关键事实还在。
作用:当历史超过阈值时,调用模型把旧消息压缩成一条摘要消息,用「摘要 + 最近几条」替代全量历史。窗口管「近处的清晰」,摘要管「远处的轮廓」,两者拼起来才是完整的长程记忆。
签名(手写实现——LangChain 1.x 没有现成摘要记忆类,而这也正是本章想让你体会的:它不难,值得亲手写一次):
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
SUMMARY_TEMPLATE = (
"你是拾光书屋的店长助理。请把下面这段店员与会员的对话压缩成一份摘要,"
"务必保留:会员称呼、会员号、阅读偏好、借阅/预约状态、答应过会员的事。"
"不超过 5 句话。\n\n【对话记录】\n{history}"
)
summarizer = (
ChatPromptTemplate.from_template(SUMMARY_TEMPLATE)
| model
| StrOutputParser()
)
def compress_history(
messages: list[AnyMessage],
keep_recent: int = 4,
trigger: int = 10,
) -> list[AnyMessage]:
"""历史超过 trigger 条时:旧消息 → 摘要便签,最近 keep_recent 条原样保留。"""
if len(messages) <= trigger:
return messages # 还不够长,不折腾
system = messages[0] if isinstance(messages[0], SystemMessage) else None
body = messages[1:] if system else messages
old, recent = body[:-keep_recent], body[-keep_recent:]
summary_text = summarizer.invoke(
{"history": get_buffer_string(old)} # 把旧消息渲染成纯文本喂给摘要链
)
note = SystemMessage(f"【此前对话摘要】{summary_text}")
return ([system] if system else []) + [note] + recent
其中 get_buffer_string(同样住在 langchain_core.messages)把消息列表渲染成 Human: ...\nAI: ... 的纯文本——正好是摘要模型想吃的形态。
最小示例:
from langchain_core.messages import get_buffer_string
# 造一段 16 条的长历史(实际场景中它来自 chat() 循环的日积月累)
long_history: list[AnyMessage] = [system]
for i in range(7):
long_history.append(HumanMessage(f"第 {i + 1} 轮:我上次问的《山月记》借期是多久?"))
long_history.append(AIMessage(f"第 {i + 1} 轮回答:借期 14 天,可以续借一次。"))
compressed = compress_history(long_history, keep_recent=4, trigger=10)
print(f"压缩前 {len(long_history)} 条 → 压缩后 {len(compressed)} 条")
# 16 → 6:system + 摘要便签 + 最近 4 条
「摘要便签」也可以不另起一条消息,而是合并进店员守则:把摘要文本拼到 system 提示词末尾。两种做法都常见,后者少一条消息、前者结构更清晰,按口味选。
常见坑:
- 摘要丢事实,等于白干。会员号 2048 被模型「概括」没了,下一轮照样答不上。对策有两层:摘要提示词里点名要保留的字段(业务字段写死在提示词里,如上面那句「务必保留:会员称呼、会员号……」);真正不能丢的关键事实(会员等级、办卡信息)压根别放进摘要链,直接写死在 system 或外部数据库里。
- 失真会累积——复印件的复印件。渐进式摘要(旧摘要 + 新挤出的消息 → 再摘要)每压一次都有损耗,几轮下来早期细节可能面目全非。触发阈值别设太低(别聊三轮就压缩),给「近处的原文」留足窗口。
- 每轮都跑摘要 = 每轮多一次模型调用。摘要只在「超过阈值」时触发,平时零开销——上面
compress_history开头那行if就是干这个的。 - 摘要消息的角色:用
SystemMessage承载摘要最省心(模型天然把它当背景资料);有教程用HumanMessage开头写「以下是之前对话的总结」也行,但别用AIMessage——历史里凭空多出一段「模型自言自语」。 - 压缩完记得替换原列表(同第 3 节最后一个坑):
messages[:] = compress_history(messages)。
5. 持久化:打烊前把白板拍照存档
比喻:内存里的记录本,进程一退出就烧成灰——服务器半夜一重启,三百位会员的对话记录集体蒸发。持久化就是打烊前把白板拍照存档:明早开店(进程重启),照片一放(文件读回),店员接着昨天的上下文继续聊。
作用:把 messages 列表落成磁盘上的 JSON 文件,重启后原样读回。存取离不开一对现成的序列化工具:messages_to_dict 把消息列表变成可 JSON 化的 dict 列表,messages_from_dict 负责逆向还原(消息的 id 等元数据都能完整往返)。
签名:
from langchain_core.messages import messages_to_dict, messages_from_dict
def save_messages(path: str, messages: list[AnyMessage]) -> None:
"""原子写入:先写临时文件,写完整体换页。"""
tmp = f"{path}.tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(messages_to_dict(messages), f, ensure_ascii=False, indent=2)
f.flush()
os.fsync(f.fileno()) # 确保数据真的落了盘
os.replace(tmp, path) # 原子换页:要么旧文件、要么新文件
def load_messages(path: str) -> list[AnyMessage]:
"""读回历史;文件不存在就发新本子,坏了就报警并兜底。"""
if not os.path.exists(path):
return []
try:
with open(path, encoding="utf-8") as f:
return messages_from_dict(json.load(f))
except json.JSONDecodeError:
print(f"[警告] {path} 已损坏,启用新记录本")
return []
messages_to_dict 的产物长这样(一条用户消息):
{"type": "human", "data": {"content": "我叫小雅,会员号 2048", "type": "human",
"additional_kwargs": {}, "id": None, ...}}
最小示例——重启后记忆失而复得:
import os
from langchain_core.messages import AIMessage, HumanMessage
path = "demo/history_member-2048.json"
save_messages(path, [
HumanMessage("我是小雅,会员号 2048,上周借了《山月记》"),
AIMessage("已登记!《山月记》借期 14 天,下周三到期。"),
])
# —— 模拟进程重启:内存清空 ——
restored = load_messages(path)
print(restored[0].content) # '我是小雅,会员号 2048,上周借了《山月记》'
# 接着聊:restored 就是新的 messages 列表,chat(restored, ...) 无缝续杯
为什么非要「临时文件 + os.replace」? 这是文件写入的原子化:直接 open(path, "w") 写原文件,等于站在梯子上现场改招牌——写到一半断电、崩溃、磁盘满,路人(下次 load 的你)看到的是「拾光书 屋」的半成品,json.load 直接抛 JSONDecodeError,整段会话报废。而「先写 .tmp、再整体换页」的套路里,os.replace 在操作系统层面是原子的:换页的那一瞬间只有新旧两种状态,不存在半新半旧的中间态。哪怕写 .tmp 时断电,坏的是临时文件,原档毫发无伤。
什么时候写盘? 不必每轮都写(磁盘和数据库都会嫌烦),常见的触发时机:
| 触发时机 | 适合场景 |
|---|---|
| 每轮对话后 | 客服/咨询等记录珍贵的场景,量级可控时最稳妥 |
| 每隔 N 轮 / N 分钟 | 普通闲聊,容忍丢最后几轮,换 I/O 大减 |
| 会话结束 / 进程退出 | 打烊统一存档,配 atexit 或信号处理兜底 |
| 用户主动保存 | 「保存本次咨询记录」这类显式动作 |
常见坑:
open(path, "w")裸写原文件——上面刚说完,这是持久化里最疼的一跤。标准姿势永远是:临时文件写好、os.replace换页。- 忘了
ensure_ascii=False:json.dump默认把中文转成\uXXXX转义。文件能读回没错,但你要人工翻看排查时,满屏转义码等于给自己上刑。 - 多进程同时写一个文件:两个 worker 都在「写 tmp → 换页」,后换的会覆盖先换的,对话记录悄悄丢半截。单进程随便玩;多进程要么加文件锁,要么直接升级到第 6 节的数据库方案。
- 坏文件没有兜底:
json.JSONDecodeError不捕获,一位会员的坏文件能让整个服务起不来。load里 try/except 是廉价保险——坏了就备份原文件、发新本子,别让一条坏数据放倒全场。
6. 第三方存储:记忆住进总部档案室
比喻:JSON 文件是单店账本——放在这家店的抽屉里,店关了账本还在,但隔壁分店看不到。拾光书屋开了三家分店(多进程/多机部署)之后,会员在小雅在 A 店聊的记录,走到 B 店就得从头再来。这时候记录本要搬进总部档案室(Redis / Postgres):所有分店共享一个档案室,谁来都能调阅。
作用(本节是概念课,不写完整代码):当出现以下任一信号,就该把第 5 节的 JSON 文件升级成数据库了——
- 多进程:Web 服务开了不止一个 worker(比如
uvicorn --workers 4),进程内存互不相通,第 2 节的sessionsdict 各自为政,会员的请求被分到不同 worker 就「间歇性失忆」; - 多机:服务部署在多台服务器上,没有任何一台的本地文件是「全局」的;
- 要审计:聊天记录要留档、要检索、要给运营分析——文件翻起来太原始;
- 写冲突管不过来:第 5 节说的文件锁已经成了新瓶颈。
存的东西没变,变的是抽屉。无论存哪,记忆的形状始终是「消息列表序列化后的结构」(就是 messages_to_dict 产出的那种 dict);换存储换的只是读和写两个函数的实现:
- Redis:内存数据库,读写极快,天生适合「正在进行的会话」这种热数据。给每个
session_id一个 key,value 存序列化后的消息列表;设个 TTL(比如 30 天),不来的会员记录自动过期,档案室不会被僵尸 key 撑爆。 - Postgres:磁盘上的关系型数据库,适合长期留存、审计、分页翻阅历史——「调出这位会员过去三个月的全部对话」这种活儿它最在行。
真正自己动手前,先认识一个词:LangGraph。它是官方给「记忆自动化」指定的继任方向——所谓 checkpointer(检查点器)就是框架级的「边聊边存」:你在图里每走一步,它自动把当时的完整状态(含消息列表)存进 InMemorySaver、SqliteSaver、RedisSaver 等容器之一。手写方案练的是内功,工程里想省事可以直接用它。第 11 章搭 Agent 时会再遇到它。
常见坑:
- 教程项目急着上 Redis:单机跑得欢、用户个位数,JSON 文件绰绰有余,多出来的运维成本(连接、超时、部署)纯属给自己加戏。升级信号(上面四条)没亮之前,别动。
- TTL 忘了设:Redis 里的会话 key 只进不出,一年后内存被几十万个「聊过一次就再没来」的僵尸会话吃光。
- 把历史存进了 SQL 却不建索引:按
session_id查会话列表,没索引就是全表扫描——数据库不背这个锅。
7. 新旧对照表:看老教程不迷路
LangChain 曾经有一整个 Memory 家族,老教程、老博客、老视频里到处是它们的身影。1.x 时代官方态度很明确:这些类已废弃或移出核心,正路只剩一条——messages 列表自己管,长了自己裁(第 3 节)、自己压(第 4 节)、自己存(第 5 节)。这张表帮你把新老世界对上号:
| 老教程里的名字 | 它当年干什么 | 1.x 里去哪了 | 本章对应做法 |
|---|---|---|---|
ConversationBufferMemory |
全量记忆,一轮不落 | langchain_classic.memory(遗留) |
手动维护 messages 列表(第 1、2 节) |
ConversationBufferWindowMemory |
滑动窗口,只记最近 k 轮 | langchain_classic.memory(遗留) |
trim_messages + token_counter=len(第 3 节) |
ConversationSummaryMemory |
摘要记忆,全程压缩 | langchain_classic.memory(遗留) |
手写摘要链(第 4 节) |
ConversationSummaryBufferMemory |
摘要 + 窗口混合 | langchain_classic.memory(遗留) |
第 4 节的「摘要便签 + 最近 N 条」正是它 |
ChatMessageHistory |
消息历史容器 | 核心里还有 InMemoryChatMessageHistory,但不再唱主角 |
直接用 list[AnyMessage](第 1 节) |
FileChatMessageHistory |
历史存文件 | classic / 社区包 | 自己 JSON 读写 + 原子写入(第 5 节) |
RedisChatMessageHistory |
历史存 Redis | classic + redis 依赖 | 自存 Redis(第 6 节概念) |
RunnableWithMessageHistory |
给链自动注入历史 | core 里还在,官方文档已归入遗留路线 | 封装 chat()(第 2 节);框架级上 LangGraph checkpointer |
一眼识别的口诀:看到以 Memory 结尾的类名,基本可以断定是旧时代的遗产。遇到老教程不必绕道走——它的思路(缓冲、窗口、摘要、持久化)全都是好思路,本章的每一节就是这些思路在 1.x 里的转世。要的只是翻译:看到 ConversationBufferWindowMemory,心里默念「trim_messages,token_counter=len」。
8. 本章小结
一张速查表带走本章:
| 需求 | 做法 |
|---|---|
| 记住多轮对话 | messages 列表:append(HumanMessage) → invoke(messages) → append(reply) |
| 封装对话循环 | 三行 chat() 函数;多会话用 dict[session_id, list] |
| 滑动窗口 | trim_messages(messages, max_tokens=k, token_counter=len, strategy="last") |
| 按 token 预算裁 | token_counter="approximate"(近似)或 token_counter=model(精确) |
| 保住人设与合法开头 | include_system=True + start_on="human" |
| 替换原历史 | messages[:] = window(切片赋值保引用) |
| 长程摘要 | 摘要链 + get_buffer_string(old) → SystemMessage 便签 + 最近 N 条 |
| 落盘 | messages_to_dict + json.dump + os.replace 原子换页 |
| 读回 | messages_from_dict(json.load(...)),坏文件 try/except 兜底 |
| 多进程 / 多机 | Redis(热会话 + TTL)/ Postgres(留存审计);框架级用 LangGraph checkpointer |
| 看到老 API | 对照第 7 节的表翻译——Memory 结尾 = 旧时代遗产 |
配套 demo(本章 demo/ 目录):
demo_memory.py:单文件走完记忆全流程——手动维护 messages → 多会话隔离 →trim_messages开窗(含include_system/start_on的对比实验)→ 摘要压缩 → JSON 原子化持久化(含坏文件兜底演示)。历史管理逻辑不需要模型也能完整演示:本机 Ollama 未启动时,店员回复换成内置剧本(它真的会从历史里翻出会员号来回答——记忆机制本身被当场验证),摘要换成截断式假摘要;启动 Ollama 后重跑即为真实模型版。
到这里,拾光书屋的店员已经「记得住」了。但读者接下来的请求越来越具体:「帮我查一下《山月记》还有没有库存」「把我的读书会报名改到下周」。店员嘴上答应得再漂亮,也变不出一个真实的查询动作——会说话与会干活之间,隔着「工具调用」这座桥。下一章,我们给店员发工具。
下一章:10 · 工具调用