(LangChain 实战14)：基于 ChatMessageHistory 自定义实现对话记忆功能

在 LangChain 开发对话机器人时，对话记忆是实现上下文连贯交互的核心，而ChatMessageHistory是 LangChain 中管理对话消息的底层核心类，专门用于存储、追加用户与 AI 的对话消息，是实现自定义记忆功能的最优选择之一。

本文将基于ChatMessageHistory从零实现对话记忆功能，全程使用简洁的原生代码，不依赖高阶封装的记忆组件，详细拆解每一步实现逻辑，适合 LangChain 入门者理解记忆功能的底层原理，代码可直接复制运行，最终实现机器人记住对话上下文的效果。

核心实现思路

本次基于ChatMessageHistory实现记忆功能的核心逻辑仅有 3 步，简单易懂且贴合 LangChain 的设计规范：

1. 定义全局的 ChatMessageHistory 实例作为消息存储容器，确保跨轮对话共享同一份历史消息，不会被清空；
2. 向实例中手动添加系统消息（定义 AI 角色 / 规则），并通过其内置方法add_user_message/add_ai_message自动追加每轮的用户、AI 消息；
3. 构建提示模板时使用MessagesPlaceholder占位符，将ChatMessageHistory中的消息列表传入占位符，让模型获取完整的对话上下文。

环境准备

首先安装本次开发所需的 Python 依赖库，执行以下命令即可完成安装：

pip install langchain-core langchain-openai langchain-community python-dotenv

各依赖的核心作用：

• langchain-core：LangChain 核心库，提供提示模板、消息类型、占位符等基础组件；
• langchain-openai：对接 OpenAI / 兼容 OpenAI 接口的大模型（本文以深度求索 DeepSeek 为例）；
• langchain-community：提供ChatMessageHistory消息管理类；
• python-dotenv：加载.env文件中的环境变量，保护 API_KEY 等敏感信息。

同时在项目根目录创建.env文件，配置大模型的基础地址和 API_KEY（其他兼容 OpenAI 接口的模型仅需修改model和base_url）：

# .env文件配置内容
OPENAI_BASE_URL=https://api.deepseek.com
OPENAI_API_KEY=你的DeepSeek_API_KEY

代码分步拆解讲解

下面将逐段讲解基于ChatMessageHistory实现记忆功能的代码，每一步都标注核心作用，确保理解每一行代码的意义，代码结构保持简洁，无冗余逻辑。

第一步：导入所需模块与类

import os
from dotenv import load_dotenv
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.messages import SystemMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

核心类 / 模块说明：

• os + load_dotenv：加载系统环境变量和.env文件中的自定义配置，避免硬编码敏感信息；
• ChatMessageHistory：核心消息管理类，用于存储对话消息，提供便捷的消息追加、清空方法；
• SystemMessage：系统消息类，用于定义 AI 的角色、回复规则，是对话的基础配置；
• ChatPromptTemplate + MessagesPlaceholder：构建对话式提示模板，MessagesPlaceholder为历史消息预留动态插入位置；
• ChatOpenAI：对接兼容 OpenAI 接口的大模型，实现模型调用。

第二步：加载环境变量并初始化大模型

# 加载.env文件中的环境变量
load_dotenv()

# 构建大模型实例
llm = ChatOpenAI(
    model="deepseek-chat",  # 调用的大模型名称（DeepSeek对话模型）
    base_url=os.getenv('OPENAI_BASE_URL'),  # 从环境变量读取模型基础地址
    api_key=os.getenv('OPENAI_API_KEY'),    # 从环境变量读取API_KEY
)

• load_dotenv()：执行后可通过os.getenv()直接获取.env文件中的配置项，是保护敏感信息的标准写法；
• ChatOpenAI初始化：参数与兼容 OpenAI 接口的模型通用，更换模型时仅需修改model和base_url，核心代码无需改动。

第三步：初始化全局 ChatMessageHistory 并添加系统消息

# 定义全局的ChatMessageHistory实例，作为对话消息的存储容器
history = ChatMessageHistory()
# 向历史容器中添加系统消息，定义AI角色（仅添加一次，全程生效）
history.messages.append(SystemMessage(content="你是一个AI助手，你的名字叫小智"))

这是实现记忆功能的关键步骤，核心说明：

1. 全局实例：将history定义在函数外部，确保所有函数调用共享同一个实例，跨轮对话的消息会持续追加，不会被清空；如果定义在函数内部，每次调用都会新建实例，导致历史消息丢失，无法实现记忆；
2. 添加系统消息：ChatMessageHistory无专门的系统消息追加方法，需手动导入SystemMessage并通过history.messages.append()添加，系统消息仅需添加一次，后续所有对话都会携带该规则，避免重复添加造成 token 浪费；
3. history.messages：是ChatMessageHistory的核心属性，为列表类型，存储所有消息实例（系统、用户、AI），后续直接将该列表传入提示模板的占位符即可。

第四步：自定义函数实现对话记忆核心逻辑

这部分是代码的核心，整合了提示模板构建、模型调用、消息追加三大功能，实现从接收用户问题到返回 AI 带上下文回复的完整流程：

# 自定义记忆功能函数，接收用户问题，返回AI带上下文的回复
def chat_message_history(msg: str) -> str:
    # 1. 构建提示模板：仅保留历史消息占位符和用户当前问题
    chat_prompt_template = ChatPromptTemplate.from_messages([
        MessagesPlaceholder(variable_name="history"),  # 历史消息占位符，关联全局history
        ("human", "{question}"),                      # 用户当前输入的问题
    ])

    # 2. 构建LangChain管道链，实现「提示模板→大模型」的流水线调用
    chain = chat_prompt_template | llm

    # 3. 调用模型，传入历史消息和用户问题，生成带上下文的回复
    response = chain.invoke({"history": history.messages, "question": msg})

    # 4. 追加本轮消息到全局历史容器，实现记忆持久化
    history.add_user_message(msg)        # 追加用户消息，底层自动封装为HumanMessage
    history.add_ai_message(response.content)  # 追加AI回复，底层自动封装为AIMessage

    # 返回AI的纯文本回复内容
    return response.content

核心细节讲解：

1. 提示模板简化：因系统消息已存入history，模板无需再单独写system行，仅需通过MessagesPlaceholder加载历史消息即可，简洁且符合 LangChain 设计规范；
2. 管道链构建：使用|管道符语法，是 LangChain 的经典写法，等价于将提示模板的输出作为大模型的输入，简化了手动传参的繁琐逻辑；
3. 模型调用：invoke方法中传入history.messages（全局历史消息列表）和question（当前用户问题），模型会基于完整的上下文生成回复；
4. 消息追加：ChatMessageHistory内置了add_user_message和add_ai_message便捷方法，无需手动封装消息类，直接传入文本内容即可，底层会自动转换为HumanMessage和AIMessage并追加到history.messages中。

第五步：主函数实现交互式持续对话

if __name__ == '__main__':
    # 无限循环实现持续对话，直到用户输入退出指令
    while True:
        input_message = input("请输入问题: ")
        # 定义退出条件，兼容中文和英文指令
        if input_message.lower() in ['退出', 'exit', 'quit', 'q']:
            print("再见！")
            break
        # 调用自定义记忆函数，获取AI带上下文的回复
        output_message = chat_message_history(input_message)
        # 打印AI回复内容
        print(output_message)

• 无限循环while True：实现持续的对话交互，满足日常聊天的使用场景；
• 多条件退出：兼容退出/exit/quit/q四种指令，提升使用体验；
• 函数调用：将用户输入传入chat_message_history函数，直接获取并打印 AI 的带上下文回复，逻辑清晰。

完整可运行代码

将以上所有步骤整合，就是基于ChatMessageHistory实现记忆功能的完整代码，可直接复制使用（只需修改.env文件中的 API_KEY）：

import os

from dotenv import load_dotenv
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.messages import SystemMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

# 加载env环境变量
load_dotenv()

# 构建模型
llm = ChatOpenAI(
    model="deepseek-chat",
    base_url=os.getenv('OPENAI_BASE_URL'),
    api_key=os.getenv('OPENAI_API_KEY'),
)

# 定义全局history存储历史消息，实现跨轮对话共享
history = ChatMessageHistory()
# 添加入系统消息，定义AI角色和名称
history.messages.append(SystemMessage(content="你是一个AI助手，你的名字叫小智"))

# chat_message_history 实现记忆功能
def chat_message_history(msg: str) -> str:
    # 定义消息模版：历史占位符 + 用户当前问题
    chat_prompt_template = ChatPromptTemplate.from_messages([
        MessagesPlaceholder(variable_name="history"),
        ("human", "{question}"),
    ])

    # 构建管道链，实现提示模板到模型的流水线调用
    chain = chat_prompt_template | llm

    # 调用模型，传入历史消息和用户问题
    response = chain.invoke({"history": history.messages, "question": msg})

    # 追加本轮用户消息和AI消息到全局历史，实现记忆
    history.add_user_message(msg)
    history.add_ai_message(response.content)

    # 返回AI回复内容
    return response.content


if __name__ == '__main__':
    while True:
        input_message = input("请输入问题: ")
        # 添加退出条件，支持多指令退出
        if input_message.lower() in ['退出', 'exit', 'quit', 'q']:
            print("再见！")
            break
        # 调用记忆函数并打印回复
        output_message = chat_message_history(input_message)
        print(output_message)

实际运行结果展示

直接运行上述完整代码，按照控制台提示输入问题，即可看到带上下文记忆的对话效果，以下是本地实际运行的完整输出结果，与代码逻辑完全匹配：

E:\pythonWorkSpace\pythonProject1\.venv1\Scripts\python.exe "E:\pythonWorkSpace\LangChain\(14)LangChain中Memory设计思路ChatMessageHistory.py" 
请输入问题: 我是Bruk,很高兴认识你
你好，Bruk！很高兴认识你！我是小智，随时准备为你提供帮助或解答问题。有什么我可以为你做的吗？ 😊
请输入问题: 我是谁?
你是Bruk呀！我们刚刚互相介绍过呢！如果还有其他问题或需要帮助的地方，随时告诉我哦～ 😄
请输入问题:

运行结果分析

1. 第一次输入我是Bruk,很高兴认识你：全局history中仅有系统消息，模型基于系统规则进行初次问候，随后该用户消息和 AI 回复被自动追加到history中；
2. 第二次输入我是谁?：模型通过history.messages获取到上一轮的对话历史，准确识别出用户的名字是 Bruk，实现了上下文记忆，回复贴合对话场景；
3. 全程无冗余打印，AI 回复连贯自然，完全符合对话记忆的开发需求。

核心要点总结

本文基于ChatMessageHistory实现 LangChain 的自定义对话记忆功能，核心知识点只需记住 3 点，即可灵活运用：

1. 全局实例是关键：ChatMessageHistory必须定义为函数外部的全局变量，确保跨轮对话共享消息，否则会因实例重新创建导致历史丢失；
2. 系统消息的添加方式：通过history.messages.append(SystemMessage(content="..."))手动添加，仅添加一次，避免重复造成 token 浪费；
3. 消息追加的便捷方法：使用ChatMessageHistory内置的add_user_message和add_ai_message，直接传入文本即可，无需手动封装消息类，简化代码；
4. 占位符关联历史：提示模板中通过MessagesPlaceholder预留位置，模型调用时传入history.messages，即可让模型获取完整的对话上下文。

这种实现方式是 LangChain 记忆功能的底层写法，理解后再学习高阶的ConversationBufferMemory等封装组件会更加轻松，是 LangChain 入门者的必学内容。