网站导航
> 技术文档 > 【LangChain】langchain_community.chat_message_histories.ChatMessageHistory 类：在内存中存储和管理对话消息历史_langchain chatmessagehistory

【LangChain】langchain_community.chat_message_histories.ChatMessageHistory 类：在内存中存储和管理对话消息历史_langchain chatmessagehistory

网友: 技术文档 2025-07-30 11:58:34

langchain_community.chat_message_histories.ChatMessageHistory 类是 LangChain 社区库中用于管理聊天消息历史的工具类，旨在存储和跟踪对话中的消息（如用户输入和模型响应），支持多轮对话和上下文管理。

本文基于 LangChain 0.3.x ，详细介绍 ChatMessageHistory 的定义、参数、方法和典型场景，并提供一个独立示例，展示如何使用 ChatMessageHistory 结合 ChatOpenAI 实现人工智能主题的多轮对话，示例突出消息历史在上下文保留中的作用。

langchain_community.chat_message_histories.ChatMessageHistory 简介

ChatMessageHistory 是 LangChain 社区库中的内存消息历史存储类，用于在内存中存储和管理对话消息（HumanMessage、AIMessage 等）。它是一个轻量级的消息历史实现，适合临时对话管理，常用于测试或简单应用。相比其他历史存储（如 RedisChatMessageHistory），它无需外部数据库，但数据不持久化。

核心功能：

  • 存储和管理对话消息（用户输入和模型输出）。
  • 支持添加、获取、清除消息。
  • 提供消息列表访问，集成到对话链。
  • 支持多轮对话的上下文保留。

适用场景：

  • 构建多轮对话系统（如聊天机器人、问答助手）。
  • 测试对话模型的上下文理解能力。
  • 临时存储对话历史，快速原型开发。
  • 与 LLM 链结合，实现上下文感知的交互。

与其他消息历史类对比：

  • ChatMessageHistory：内存存储，适合临时使用。
  • RedisChatMessageHistory：持久化到 Redis，适合生产环境。
  • SQLChatMessageHistory：存储到 SQL 数据库，适合长期记录。

类定义和初始化

以下是 ChatMessageHistory 的定义，基于 LangChain 社区库源码（langchain_community/chat_message_histories/in_memory.py）和官方文档（ChatMessageHistory)。

类签名
class ChatMessageHistory(BaseChatMessageHistory): def __init__(self) -> None
  • 参数：无额外参数，直接初始化内存存储。
  • 属性：
    • messages（List[BaseMessage]）：存储的消息列表。
  • 功能：
    • 在内存中维护消息历史。
    • 提供添加、获取、清除消息的接口。
初始化示例
from langchain_community.chat_message_histories import ChatMessageHistoryhistory = ChatMessageHistory()

常用方法

ChatMessageHistory 继承自 langchain_core.chat_history.BaseChatMessageHistory，提供以下核心方法。

1. add_message
def add_message(self, message: BaseMessage) -> None
  • 功能：添加单条消息到历史。
  • 输入：message（BaseMessage），如 HumanMessage 或 AIMessage。
  • 示例： 
    from langchain_core.messages import HumanMessage, AIMessagehistory.add_message(HumanMessage(content=\"什么是人工智能？\"))history.add_message(AIMessage(content=\"人工智能是计算机科学的一个分支。\"))
2. add_user_message
def add_user_message(self, message: str) -> None
  • 功能：添加用户消息（HumanMessage）。
  • 输入：message（str），用户输入文本。
  • 示例： 
    history.add_user_message(\"AI 有哪些应用？\")
3. add_ai_message
def add_ai_message(self, message: str) -> None
  • 功能：添加 AI 消息（AIMessage）。
  • 输入：message（str），模型响应文本。
  • 示例： 
    history.add_ai_message(\"AI 应用于医疗、自动驾驶等地方。\")
4. clear
def clear(self) -> None
  • 功能：清除所有消息历史。
  • 示例： 
    history.clear()print(len(history.messages)) # 0
5. messages 属性
  • 功能：访问存储的消息列表。
  • 输出：List[BaseMessage]，所有消息。
  • 示例： 
    for msg in history.messages: print(f\"{msg.__class__.__name__}: {msg.content}\")
6. async 方法
  • 功能：异步版本方法（如 aclear、aadd_message）。
  • 使用场景：高并发场景。
  • 示例： 
    import asyncioawait history.aclear()

使用方式

以下是使用 ChatMessageHistory 的步骤。

1. 安装依赖
pip install --upgrade langchain langchain-openai langchain-community
2. 设置 OpenAI API 密钥
export OPENAI_API_KEY=\"your-api-key\"

或在代码中：

import osos.environ[\"OPENAI_API_KEY\"] = \"your-api-key\"
3. 初始化 ChatMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistoryhistory = ChatMessageHistory()
4. 构建对话链
from langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderllm = ChatOpenAI(model=\"gpt-3.5-turbo\")prompt = ChatPromptTemplate.from_messages([ (\"system\", \"你是一个人工智能专家，回答用户问题。\"), MessagesPlaceholder(variable_name=\"history\"), (\"human\", \"{input}\")])chain = prompt | llm
5. 调用链并存储历史
history.add_user_message(\"什么是人工智能？\")response = chain.invoke({\"history\": history.messages, \"input\": \"什么是人工智能？\"})history.add_ai_message(response.content)print(response.content)

使用 ChatMessageHistory 的示例

以下是一个独立示例，展示如何使用 ChatMessageHistory 结合 ChatOpenAI 实现人工智能主题的多轮对话，保留上下文并回答用户查询。示例使用 ConversationChain 简化对话管理，突出消息历史的作用。

准备环境：

  • 获取 OpenAI API 密钥：OpenAI Platform。
  • 设置环境变量： 
    export OPENAI_API_KEY=\"your-api-key\"
  • 安装依赖： 
    pip install --upgrade langchain langchain-openai langchain-community

代码：

from langchain_community.chat_message_histories import ChatMessageHistoryfrom langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.runnables.history import RunnableWithMessageHistory# 初始化 ChatMessageHistoryhistory = ChatMessageHistory()# 初始化 ChatOpenAIllm = ChatOpenAI(model=\"gpt-3.5-turbo\", temperature=0.7)# 定义提示模板prompt = ChatPromptTemplate.from_messages([ (\"system\", \"你是一个人工智能专家，回答用户问题。\"), MessagesPlaceholder(variable_name=\"history\"), (\"human\", \"{input}\")])# 定义输出解析器parser = StrOutputParser()# 创建对话链chain = prompt | llm | parser# 包装链以支持消息历史chain_with_history = RunnableWithMessageHistory( runnable=chain, get_session_history=lambda session_id: history, input_messages_key=\"input\", history_messages_key=\"history\")# 测试 ChatMessageHistory 和对话链print(\"测试 ChatMessageHistory 和对话链：\")try: session_id = \"ai_chat_001\" # 第一轮对话 question1 = \"什么是人工智能？\" result1 = chain_with_history.invoke( {\"input\": question1}, config={\"configurable\": {\"session_id\": session_id}} ) print(f\"问题 1: {question1}\") print(f\"回答 1: {result1}\") # 第二轮对话（依赖上下文） question2 = \"它有哪些应用？\" result2 = chain_with_history.invoke( {\"input\": question2}, config={\"configurable\": {\"session_id\": session_id}} ) print(f\"\\n问题 2: {question2}\") print(f\"回答 2: {result2}\") # 显示消息历史 print(\"\\n消息历史：\") for i, msg in enumerate(history.messages): print(f\"消息 {i+1} ({msg.__class__.__name__}): {msg.content}\")except Exception as e: print(f\"错误: {e}\")

输出示例（实际输出取决于模型和 API 响应）：

测试 ChatMessageHistory 和对话链：问题 1: 什么是人工智能？回答 1: 人工智能（AI）是计算机科学的一个分支，旨在模拟人类智能，如学习、推理和问题解决。问题 2: 它有哪些应用？回答 2: 人工智能的应用包括医疗诊断、自动驾驶、语音识别、推荐系统和金融分析等地方。消息历史：消息 1 (HumanMessage): 什么是人工智能？消息 2 (AIMessage): 人工智能（AI）是计算机科学的一个分支，旨在模拟人类智能，如学习、推理和问题解决。消息 3 (HumanMessage): 它有哪些应用？消息 4 (AIMessage): 人工智能的应用包括医疗诊断、自动驾驶、语音识别、推荐系统和金融分析等地方。
代码说明
  1. ChatMessageHistory：
    • 初始化 history 用于存储对话消息。
  2. LLM 初始化：
    • 使用 ChatOpenAI 调用 gpt-3.5-turbo，设置 temperature=0.7。
  3. 提示模板：
    • ChatPromptTemplate 包含系统消息、历史占位符和用户输入。
    • MessagesPlaceholder 注入 history.messages。
  4. 对话链：
    • 使用 LCEL 组合 prompt、llm 和 parser。
    • RunnableWithMessageHistory 包装链，绑定 history。
  5. 测试：
    • 进行两轮对话：定义 AI 和询问应用。
    • 使用 session_id 跟踪同一会话。
    • 显示消息历史，展示上下文保留。
  6. 错误处理：
    • 使用 try-except 捕获 API 或链错误。

运行要求：

  • 有效的 OpenAI API 密钥： 
    export OPENAI_API_KEY=\"your-api-key\"
  • 安装依赖： 
    pip install --upgrade langchain langchain-openai langchain-community
  • 网络连接：访问 https://api.openai.com.

注意事项

  1. API 密钥：
    • 确保 OPENAI_API_KEY 已设置： 
      echo $OPENAI_API_KEY
    • 或在代码中设置： 
      llm = ChatOpenAI(api_key=\"your-api-key\")
  2. 消息历史管理：
    • 内存存储不持久化，重启后丢失： 
      print(len(history.messages)) # 重启后为 0
    • 限制历史长度以节省上下文： 
      if len(history.messages) > 10: history.messages = history.messages[-10:]
  3. 会话 ID：
    • 使用 session_id 区分对话： 
      config = {\"configurable\": {\"session_id\": \"ai_chat_001\"}}
    • 不同 session_id 使用同一 history 对象可能导致混淆： 
      history.clear() # 清空前确认
  4. 性能优化：
    • 异步调用：使用 ainvoke： 
      result = await chain_with_history.ainvoke(input, config)
    • 持久化历史：切换到 RedisChatMessageHistory： 
      from langchain_community.chat_message_histories import RedisChatMessageHistoryhistory = RedisChatMessageHistory(session_id=\"ai_chat_001\", redis_url=\"redis://localhost:6379\")
  5. 错误调试：
    • 消息丢失：
      • 检查历史： 
        print(history.messages)
      • 验证输入： 
        print(input)
    • API 错误：
      • 检查密钥： 
        print(os.environ.get(\"OPENAI_API_KEY\"))
      • 增加超时： 
        llm = ChatOpenAI(timeout=30)
    • 上下文溢出：
      • 检查消息数量： 
        print(len(history.messages))
      • 截断历史： 
        history.messages = history.messages[-4:]

常见问题

Q1：如何持久化消息历史？
A：使用 RedisChatMessageHistory 或 SQLChatMessageHistory：

from langchain_community.chat_message_histories import RedisChatMessageHistoryhistory = RedisChatMessageHistory(session_id=\"ai_chat_001\", redis_url=\"redis://localhost:6379\")

Q2：如何限制上下文长度？
A：截断消息历史：

max_messages = 6if len(history.messages) > max_messages: history.messages = history.messages[-max_messages:]

Q3：如何与 RAG 结合？
A：添加检索上下文：

from langchain.vectorstores import FAISSvectorstore = FAISS.from_documents(docs, OpenAIEmbeddings())retriever = vectorstore.as_retriever()prompt = ChatPromptTemplate.from_messages([ (\"system\", \"根据上下文和历史回答：\\n{context}\"), MessagesPlaceholder(variable_name=\"history\"), (\"human\", \"{input}\")])chain = {\"context\": retriever, \"history\": lambda x: history.messages, \"input\": lambda x: x[\"input\"]} | prompt | llm

Q4：如何支持开源模型？
A：使用 ChatOllama：

from langchain_ollama import ChatOllamallm = ChatOllama(model=\"llama3\")chain = prompt | llm | parserchain_with_history = RunnableWithMessageHistory( runnable=chain, get_session_history=lambda session_id: history, input_messages_key=\"input\", history_messages_key=\"history\")

总结

langchain_community.chat_message_histories.ChatMessageHistory 是 LangChain 中管理对话历史的轻量工具，核心功能包括：

  • 定义：在内存中存储消息历史。
  • 初始化：简单创建，无需配置。
  • 常用方法：add_message、add_user_message、add_ai_message、clear。
  • 适用场景：多轮对话、上下文保留、快速原型。
网络标签：

相关问题