在与大语言模型（LLM）进行多轮对话时，我们常常希望模型能够“记住”上一轮说过的话。例如，你告诉模型“我叫小明”，下一句问“我叫什么？”，模型应当回答“小明”。然而，LLM 本质上是无状态的，每次调用都是独立的。
为了解决这个问题，LangChain 提供了强大的会话记忆机制。今天我们将深入探讨如何利用 LangChain 实现两种核心的记忆模式：临时会话记忆与长期会话记忆。

一、临时会话记忆

临时会话记忆是最基础的记忆形式，它将聊天历史存储在内存中。这意味着只要程序在运行，记忆就存在；一旦程序重启，记忆就会消失。

1. 核心组件解析

在代码实现中，我们主要依赖以下几个核心组件：

• InMemoryChatMessageHistory: 这是一个轻量级的内存存储类，用于在 Python 字典中保存聊天记录。
• RunnableWithMessageHistory: 这是 LangChain 中的“包装器”，它通过 LCEL（LangChain Expression Language）为现有的链增加记忆管理功能。它负责在调用模型前自动读取历史消息，并在调用后自动保存新的对话。
• MessagesPlaceholder: 在 Prompt 模板中预留一个位置，用于动态插入历史对话记录。

2. 代码实现详解

下面我们通过一段完整的代码来演示如何构建一个具备临时记忆的对话链。

from langchain_community.chat_models import ChatTongyi
from langchain_core.chat_history import InMemoryChatMessageHistory, BaseChatMessageHistory
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableLambda, RunnableConfig
from langchain_core.runnables.history import RunnableWithMessageHistory
# 初始化大模型，这里使用通义千问模型
model = ChatTongyi(model="qwen3-max-preview")
# 定义提示词模板
# 关键点：MessagesPlaceholder 用于承载历史对话
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据历史对话回复用户问题"),
        MessagesPlaceholder(variable_name="chat_history"), # 历史消息占位符
        ("human", "请回复如下问题{input}"),
    ]
)
# 定义一个调试函数，方便观察最终传给模型的 Prompt 结构
def print_prompt(prompt_value):
    """打印调试信息，显示完整的 Prompt 结构"""
    print("\n" + "=" * 30 + " PROMPT DEBUG " + "=" * 30)
    for msg in prompt_value.messages:
        print(f"[{msg.type.upper()}]: {msg.content}")
    print("=" * 74 + "\n")
    return prompt_value
# 构建基础处理链：Prompt -> 调试打印 -> 模型调用 -> 输出解析
base_chain = prompt | RunnableLambda(print_prompt) | model | StrOutputParser()
# 存储会话历史的全局字典
# Key: session_id, Value: InMemoryChatMessageHistory 对象
chat_history_store = {}
def get_chat_history(session_id: str) -> BaseChatMessageHistory:
    """
    根据 session_id 获取对应的 History 对象。
    如果不存在，则创建一个新的 InMemoryChatMessageHistory。
    """
    if session_id not in chat_history_store:
        chat_history_store[session_id] = InMemoryChatMessageHistory()
    return chat_history_store[session_id]
# 创建带有记忆功能的链
conversation_chain = RunnableWithMessageHistory(
    base_chain,  # 被增强的基础链
    get_chat_history,  # 获取历史记录的函数
    input_messages_key="input",  # 用户输入在模板中的变量名
    history_messages_key="chat_history", # 历史记录在模板中的变量名
)
if __name__ == "__main__":
    # 配置会话 ID，用于区分不同用户的会话
    session_config: RunnableConfig = {
        "configurable": {"session_id": "user_001"}
    }
    # 测试对话序列
    questions = [
        "小明有1只猫",
        "小明有4只狗",
        "宠物总数是多少"
    ]
    for q in questions:
        print(f"\nUser: {q}")
        # invoke 时传入 config，链会自动处理历史的读取与写入
        response = conversation_chain.invoke(
            {"input": q},
            config=session_config
        )
        print(f"AI: {response}")

3. 关键逻辑解读

1. 历史注入流程：
   当我们调用 conversation_chain.invoke 时，RunnableWithMessageHistory 会拦截调用。它首先通过 get_chat_history 获取 user_001 的历史记录，并将其填充到 Prompt 模板的 chat_history 位置。
2. 自动保存：
   模型生成回复后，包装器会自动将用户的提问（HumanMessage）和模型的回答（AIMessage）追加到 InMemoryChatMessageHistory 中，无需手动管理。
3. 运行结果预期：
   在第三个问题“宠物总数是多少”时，模型能够通过 chat_history 看到前两轮的对话内容（“1只猫”和“4只狗”），从而正确计算出总数为5。

二、长期会话记忆

临时记忆虽然简单，但无法应对程序重启或服务重新部署的场景。为了实现持久化存储，我们需要自定义存储后端。这里我们实现一个基于本地文件的 FileChatMessageHistory 类。

1. 设计思路

我们需要继承 BaseChatMessageHistory 基类，并实现三个核心方法：

• add_messages: 将消息列表写入文件。
• messages: 从文件读取消息列表。
• clear: 清空文件内容。
  LangChain 提供了 message_to_dict 和 messages_from_dict 工具函数，帮助我们将消息对象序列化为 JSON 格式，方便存储。

2. 代码实现详解

import json
import os
from typing import Sequence
from langchain_community.chat_models import ChatTongyi
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import message_to_dict, messages_from_dict, BaseMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableLambda, RunnableWithMessageHistory, RunnableConfig
# 自定义基于文件的会话历史记录类
class FileChatMessageHistory(BaseChatMessageHistory):
    def __init__(self, session_id: str, store_path: str = "./chat_history"):
        self.session_id = session_id  # 会话id作为文件名
        self.store_path = store_path  # 存储目录
        self.file_path = os.path.join(self.store_path, self.session_id)
        # 确保目录存在
        os.makedirs(os.path.dirname(self.file_path), exist_ok=True)
    def add_message(self, message: BaseMessage) -> None:
        # 单条消息添加接口，内部调用批量添加接口
        self.add_messages([message])
    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        # 1. 读取现有消息
        all_messages = list(self.messages)
        # 2. 追加新消息
        all_messages.extend(messages)
        # 3. 序列化并写入文件
        # 注意：这里采用覆盖写入模式，适用于中小规模历史记录
        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 = json.load(f)
                return messages_from_dict(messages)
        except FileNotFoundError:
            # 文件不存在时返回空列表
            return []
    def clear(self):
        """清空会话记录"""
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump([], f)
# --- 以下链的定义与临时记忆部分基本一致 ---
model = ChatTongyi(model="qwen3-max-preview")
prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你需要根据历史对话回复用户问题"),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "请回复如下问题{input}"),
    ]
)
def print_prompt(prompt_value):
    """打印调试信息，显示完整的 Prompt 结构"""
    print("\n" + "=" * 30 + " PROMPT DEBUG " + "=" * 30)
    for msg in prompt_value.messages:
        print(f"[{msg.type.upper()}]: {msg.content}")
    print("=" * 74 + "\n")
    return prompt_value
base_chain = prompt | RunnableLambda(print_prompt) | model | StrOutputParser()
def get_chat_history(session_id: str) -> BaseChatMessageHistory:
    """
    关键变化：这里返回的是 FileChatMessageHistory 实例
    不同的 session_id 对应不同的物理文件
    """
    return FileChatMessageHistory(session_id)
# 创建带有持久化记忆功能的链
conversation_chain = RunnableWithMessageHistory(
    base_chain,
    get_chat_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)
if __name__ == "__main__":
    session_config: RunnableConfig = {
        "configurable": {"session_id": "user_001"}
    }
    questions = [
        "小明有1只猫",
        "小明有4只狗",
        "宠物总数是多少"
    ]
    for q in questions:
        print(f"\nUser: {q}")
        response = conversation_chain.invoke(
            {"input": q},
            config=session_config
        )
        print(f"AI: {response}")

3. 关键技术点解析

1. 序列化机制：
   BaseMessage 对象无法直接存入 JSON 文件。代码中使用了 message_to_dict 将消息对象转换为字典（包含 type 和 content 等字段），读取时再通过 messages_from_dict 还原为对象。
2. 文件读写策略：
   在 add_messages 方法中，我们采用了“读取 -> 追加 -> 全量覆盖写入”的策略。这在并发场景下可能存在风险，但在单线程演示中是可行的。在生产环境中，通常会使用数据库（如 Redis、SQL）来替代文件存储。
3. 无缝切换：
   对比临时记忆的代码，你会发现只有 get_chat_history 函数的内部实现发生了变化（从 InMemoryChatMessageHistory() 变为 FileChatMessageHistory(session_id)）。这就是 LangChain 抽象层的强大之处——业务逻辑（Chain）不需要改变，只需替换存储后端即可。

此时，当你一次运行后，历史会话将被存在本地，当注释掉questions中的前两句对话时，再次运行大模型仍然能给出正确答案