LangChain 指南 · 第 09 章

对话记忆

本章配套可运行示例:demo/demomemory.py。demo 从「手动维护 messages 列表」起步,一路跑到多会话隔离、trimmessages 开窗、摘要压缩、JSON 原子化持久化;本机没启动 Ollam…

约 38 分钟15,158 字14 个代码示例

本章配套可运行示例: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 节的 sessions dict 各自为政,会员的请求被分到不同 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 · 工具调用