前言

这期可以看作一期“特别篇”。
为了理解这节课的代码，我花了快两天的时间查资料，在代码上进行注释，勉强搞懂了这些对我来说挺难理解的东西。但是我相信，如果你的理解能力比我更好一点的话，对你来说或许不会那么难弄懂。
本期只有一段源代码，本身不算长，但是你会发现它看起来变长了，原因是我加了很长的注释。为了大家的观感，我把没有注释的源代码放在末尾了，需要自取。
奇怪的是，我理解它的每一部分，但是当我想要从头开始写它时，脑袋很奇怪地变得很乱，无从下手，大概率还是基础不牢。
（找个时间，我要把在文件夹里吃灰的python基础教程拿出来看看了）

源代码

# 导入依赖（模型，模板和占位符，转换函数和基础消息类，用于继承的消息基础类，字符串输出解析器，Runnable类，json，os，序列）
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import message_to_dict, messages_from_dict, BaseMessage
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableWithMessageHistory
import json,os
from typing import Sequence

# 定义最重要的类：FileChatMessageHistory，其中要定义两个方法（添加历史消息，清空历史消息），一个属性（转换消息）
class FileChatMessageHistory(BaseChatMessageHistory):
    # 初始化方法，后续的方法可以直接调用其中的变量
    def __init__(self, session_id, storage_path):
        # 会话ID
        self.session_id = session_id
        # 不同会话ID的存储文件
        self.storage_path = storage_path
        # 完整的路径（os.path.join 用于跨平台拼接文件/目录路径）
        self.file_path = os.path.join(self.storage_path, self.session_id)

        # 确保文件存在
        # os.path.dirname(self.file_path)用于获取一个路径的目录部分。
        # os.makedirs(path, exist_ok=True)的作用：创建目录树（多级目录）。
        # exist_ok=True 确保即使目录已存在也不会报错，程序继续运行。
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)

    # 定义方法：添加消息
    # messages: Sequence[BaseMessage]是一个类型注解，表示messages应该是这样的类型。
    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        # 调用类自己的属性messages：从历史中读取一条历史消息（BaseMessage），并转换成列表list
        all_messages = list(self.messages)
        # 新的和旧的融合，all_messages变成一个新的消息（这里的messages（形参）和用self读取的messages（属性）不是一个东西）
        all_messages.extend(messages)
        # 用循环把all_messages洗成字典类型，赋给new_messages
        new_messages = [message_to_dict(msg) for msg in all_messages]
        # 写入文件，使用w模式，表示覆写
        with open(self.file_path, 'w',encoding="UTF-8") as f:
            # 把已经转为字典格式的new_messages写进文件中（注意不是dumps，dumps用来返回JSON字符串而不是写进去）
            json.dump(new_messages, f)

    # 使用property修饰，定义成成员属性
    @property
    def messages(self) -> list[BaseMessage]:
        try:
            with open(self.file_path, "r", encoding="UTF-8") as f:
                messages_data = json.load(f)    # 返回值就是：list[字典]
                # 把字典列表转成消息
                return messages_from_dict(messages_data)
        # 要是找不到文件，返回空
        except FileNotFoundError:
            return []

    def clear(self) -> None:
        with open(self.file_path, "w", encoding="UTF-8") as f:
            # 要清除记录很简单，往文件夹内覆写一个空即可。
            json.dump([], f)
            
# 接下来的部分和临时记忆差不多
model = ChatTongyi(model="qwen3-max")
# prompt = PromptTemplate.from_template(
#     "你需要根据会话历史回应用户问题。对话历史：{chat_history}，用户提问：{input}，请回答"
# )
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题。对话历史："),
        MessagesPlaceholder("chat_history"),
        ("human", "请回答如下问题：{input}")
    ]
)

str_parser = StrOutputParser()


def print_prompt(full_prompt: Sequence[BaseMessage]):
    print("="*20, full_prompt.to_string(), "="*20)
    return full_prompt


base_chain = prompt | print_prompt | model | str_parser

def get_history(session_id):
    # 调用刚刚创建的类，输入会话序号和历史消息的路径
    return FileChatMessageHistory(session_id, "./chat_history")

# 创建一个新的链，对原有链增强功能：自动附加历史消息
conversation_chain = RunnableWithMessageHistory(
    base_chain,     # 被增强的原有chain
    get_history,    # 通过会话id获取InMemoryChatMessageHistory类对象
    input_messages_key="input",             # 表示用户输入在模板中的占位符
    history_messages_key="chat_history"     # 表示用户输入在模板中的占位符
)


if __name__ == '__main__':
    # 固定格式，添加LangChain的配置，为当前程序配置所属的session_id
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }

    res = conversation_chain.invoke({"input": "小明有2个猫"}, session_config)
    print("第1次执行：", res)

    res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    print("第2次执行：", res)

    res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
    print("第3次执行：", res)

你说看起来不是很难？嗯，那也许你能更快理解它。
好了，我们从哪里开始说起呢？

依赖部分

按层次分的话，所有依赖可以如下分：
应用层
├── ChatTongyi (模型)
├── RunnableWithMessageHistory (记忆管家)
├── StrOutputParser (解析输出)
│
模板层
├── ChatPromptTemplate（聊天提示词模板）
└── MessagesPlaceholder（占位符）
│
历史存储层
├── BaseChatMessageHistory (接口)
├── FileChatMessageHistory (自定义实现)
├── message_to_dict / messages_from_dict (转换)
├── BaseMessage (消息基类)
│
底层支持
├── json, os, typing.Sequence

结合我找大模型要的资料，我们一点点解释：

应用层（直接参与对话流程）

ChatTongyi：通义千问聊天模型，负责根据提示词生成回答。是对话链的“大脑”。

RunnableWithMessageHistory：包装器，给普通链自动添加“读写历史消息”的能力。它负责调用 get_history 获取存储器，并在调用前后自动注入/保存历史。

StrOutputParser：输出解析器，将模型返回的 AIMessage 对象转换成纯字符串，方便直接打印或进一步处理。

模板层（构建提示词）

ChatPromptTemplate：创建聊天模型专用的提示词模板，输出是一个消息列表（包含 SystemMessage、HumanMessage 等），比普通字符串模板更适合多轮对话。

MessagesPlaceholder：占位符，运行时会被替换为一系列消息（如历史对话记录）。让模板中可以灵活插入不定数量的历史消息。

历史存储层（管理聊天记录）

BaseChatMessageHistory：抽象基类，定义了历史存储器必须实现的接口（messages 属性、add_messages、clear）。你的 FileChatMessageHistory 继承它，保证了与框架的兼容性。

FileChatMessageHistory：你自定义的实现类，将历史消息以 JSON 格式永久保存在硬盘文件中，实现长期记忆。

message_to_dict / messages_from_dict：转换函数。message_to_dict 将 BaseMessage 对象转为字典（便于 JSON 序列化）；messages_from_dict 将字典列表还原为消息对象列表。用于读写文件时的格式转换。

BaseMessage：所有消息类型（HumanMessage、AIMessage、SystemMessage）的基类。用于类型注解，确保你操作的是标准的消息对象。

底层支持（基础工具）

json：标准库，用于将消息字典列表写入文件（json.dump）和从文件读取（json.load）。

os：标准库，用于处理文件路径（os.path.join）和创建目录（os.makedirs）。
typing.Sequence 类型注解，表示一个有序序列（如列表、元组）。用于标注 add_messages 的参数类型，增强代码可读性。

代码部分1：定义最重要的类

FileChatMessageHistory是我们自定的一个基于BaseChatMessageHistory的类，它可以将聊天对话的历史记录永久保存到本地文件中，即使程序关闭或重启，历史也不会丢失。

# 定义最重要的类：FileChatMessageHistory，其中要定义两个方法（添加历史消息，清空历史消息），一个属性（转换消息）
class FileChatMessageHistory(BaseChatMessageHistory):
    # 初始化方法，后续的方法可以直接调用其中的变量
    def __init__(self, session_id, storage_path):
        # 会话ID
        self.session_id = session_id
        # 不同会话ID的存储文件
        self.storage_path = storage_path
        # 完整的路径（os.path.join 用于跨平台拼接文件/目录路径）
        self.file_path = os.path.join(self.storage_path, self.session_id)

        # 确保文件存在
        # os.path.dirname(self.file_path)用于获取一个路径的目录部分。
        # os.makedirs(path, exist_ok=True)的作用：创建目录树（多级目录）。
        # exist_ok=True 确保即使目录已存在也不会报错，程序继续运行。
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)

    # 定义方法：添加消息
    # messages: Sequence[BaseMessage]是一个类型注解，表示messages应该是这样的类型。
    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        # 调用类自己的属性messages：从历史中读取一条历史消息（BaseMessage），并转换成列表list
        all_messages = list(self.messages)
        # 新的和旧的融合，all_messages变成一个新的消息（这里的messages（形参）和用self读取的messages（属性）不是一个东西）
        all_messages.extend(messages)
        # 用循环把all_messages洗成字典类型，赋给new_messages
        new_messages = [message_to_dict(msg) for msg in all_messages]
        # 写入文件，使用w模式，表示覆写
        with open(self.file_path, 'w',encoding="UTF-8") as f:
            # 把已经转为字典格式的new_messages写进文件中（注意不是dumps，dumps用来返回JSON字符串而不是写进去）
            json.dump(new_messages, f)

    # 使用property修饰，定义成成员属性
    @property
    def messages(self) -> list[BaseMessage]:
        try:
            with open(self.file_path, "r", encoding="UTF-8") as f:
                messages_data = json.load(f)    # 返回值就是：list[字典]
                # 把字典列表转成消息
                return messages_from_dict(messages_data)
        # 要是找不到文件，返回空
        except FileNotFoundError:
            return []

    def clear(self) -> None:
        with open(self.file_path, "w", encoding="UTF-8") as f:
            # 要清除记录很简单，往文件夹内覆写一个空即可。
            json.dump([], f)

它需要定义三个内部方法，包括一个初始化方法，两个自定义方法。还有一个方法需要包装成属性。
init：初始化方法
add_messages：接收消息类，与历史消息中最近一次记录融合，转成JSON格式后存进文件夹内。
clear：清除记录，原理是往历史记录文件夹内覆写一个空列表。
messages：用@property修饰，变成一个类自带的属性，用来从历史消息中读取信息并转化为消息格式。

代码部分2：定义部分

这部分主要把一些需要用的工具进行基础定义，并对链进行一次“升级”。要注意如何往链中加入历史消息存储的FileChatMessageHistory类。
源代码：

model = ChatTongyi(model="qwen3-max")
# prompt = PromptTemplate.from_template(
#     "你需要根据会话历史回应用户问题。对话历史：{chat_history}，用户提问：{input}，请回答"
# )
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题。对话历史："),
        MessagesPlaceholder("chat_history"),
        ("human", "请回答如下问题：{input}")
    ]
)

str_parser = StrOutputParser()


def print_prompt(full_prompt: Sequence[BaseMessage]):
    print("="*20, full_prompt.to_string(), "="*20)
    return full_prompt


base_chain = prompt | print_prompt | model | str_parser

def get_history(session_id):
    # 调用刚刚创建的类，输入会话序号和历史消息的路径
    return FileChatMessageHistory(session_id, "./chat_history")

# 创建一个新的链，对原有链增强功能：自动附加历史消息
conversation_chain = RunnableWithMessageHistory(
    base_chain,     # 被增强的原有chain
    get_history,    # 通过会话id获取FieldChatMessageHistory类对象
    input_messages_key="input",             # 表示用户输入在模板中的占位符
    history_messages_key="chat_history"     # 表示用户输入在模板中的占位符
)

提示：这里用通用模板PromptTemplate也可以，但是还是建议用聊天模板。
model：定义聊天模型

prompt：定义聊天模板，同时需要占位符

str_parser：字符串输出解析器，可以把模型回复的内容转换为字符串类型，方便打印

print_prompt：打印提示词用的小函数，可以在模型回复你之前输出这次给了模型什么消息（可以看到历史消息被一次次加入）

base_chain：一个基础链，用来把刚刚这些东西先串在一起，但是还没有加入最重要的历史储存。（注意print_prompt要放在模型model之前，同时检查一下它的返回值要和输入值一样是提示词模板）

get_history：创建一个FileChatMessageHistory类（会话序号和储存路径需要你提供），返回值是这个类的对象

conversation_chain：“升级版”的链，可以用RunnableWithMessageHistory把之前的基础链加上历史消息存储。RunnableWithMessageHistory 会把它包起来，在调用前自动往提示词里插入历史消息，在调用后自动把本次对话存进历史。
需要传入的参数有：

链对象（基础的）：升级用

通过会话id获取FieldChatMessageHistory对象的方法get_history：告诉包装器“根据会话 ID 去哪里找对应的聊天记录”，RunnableWithMessageHistory 会在每次调用时调用这个函数，然后：
调用 .messages 属性读取历史。
调用 .add_messages() 保存新消息。

input_messages_key：告诉包装器从传入的字典中取出键名为 “input” 的值，作为本次用户的问题。

history_messages_key：历史消息往哪个位置插？

代码部分3：执行部分

if __name__ == '__main__':
    # 固定格式，添加LangChain的配置，为当前程序配置所属的session_id
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }

    res = conversation_chain.invoke({"input": "小明有2个猫"}, session_config)
    print("第1次执行：", res)

    res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    print("第2次执行：", res)

    res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
    print("第3次执行：", res)

这部分就是配置会话ID和执行的部分。
session_config 是传递给 RunnableWithMessageHistory 的配置字典，名字可以随便取，但是内部的字典是结构固定的，而且第二层必须有一个键叫“session_id”用来传入之前说的get_history函数，经过FieldChatMessageHistory类的处理后，就能返回对应这一次会话的历史消息存储体。

至于执行部分，可以先执行前两个问题，然后注释这两个问题，单独执行第三个问题，看看模型还记不记得。
同时由于这是长期记忆，所以可以执行一次后，关闭编译器再打开，单独问第三个问题看看。

纯净源代码（无注释）

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import message_to_dict, messages_from_dict, BaseMessage
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableWithMessageHistory
import json
import os
from typing import Sequence


class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, session_id, storage_path):
        self.session_id = session_id
        self.storage_path = storage_path
        self.file_path = os.path.join(self.storage_path, self.session_id)
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        all_messages = list(self.messages)
        all_messages.extend(messages)
        new_messages = [message_to_dict(msg) for msg in all_messages]
        with open(self.file_path, 'w', encoding="UTF-8") as f:
            json.dump(new_messages, f)

    @property
    def messages(self) -> list[BaseMessage]:
        try:
            with open(self.file_path, "r", encoding="UTF-8") as f:
                messages_data = json.load(f)
                return messages_from_dict(messages_data)
        except FileNotFoundError:
            return []

    def clear(self) -> None:
        with open(self.file_path, "w", encoding="UTF-8") as f:
            json.dump([], f)


model = ChatTongyi(model="qwen3-max")

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据会话历史回应用户问题。对话历史："),
        MessagesPlaceholder("chat_history"),
        ("human", "请回答如下问题：{input}")
    ]
)

str_parser = StrOutputParser()


def print_prompt(full_prompt: Sequence[BaseMessage]):
    print("="*20, full_prompt.to_string(), "="*20)
    return full_prompt


base_chain = prompt | print_prompt | model | str_parser


def get_history(session_id):
    return FileChatMessageHistory(session_id, "./chat_history")


conversation_chain = RunnableWithMessageHistory(
    base_chain,
    get_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)


if __name__ == '__main__':
    session_config = {
        "configurable": {
            "session_id": "user_001"
        }
    }

    res = conversation_chain.invoke({"input": "小明有2个猫"}, session_config)
    print("第1次执行：", res)

    res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
    print("第2次执行：", res)

    res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
    print("第3次执行：", res)

结尾

大概就这样了，理解这些代码需要的知识不少，这两天真是让我吃了些苦头。有什么问题可以在评论区留言。