当前位置：首页 > news >正文

LangChain-6-消息持久化

news 2025/7/17 8:40:48

在 LangChain 中，ChatMessageHistory 是一个关键组件，用于管理和持久化聊天消息。它为构建聊天机器人和对话式应用程序提供了强大的消息管理功能，下面将对其进行详细介绍。

1. 概述

ChatMessageHistory 是一个抽象基类，定义了一系列用于管理聊天消息历史记录的接口和方法。它允许你添加用户和 AI 的消息到历史记录中，检索历史消息，以及清理历史记录。不同的子类实现了不同的持久化策略，例如内存存储、文件存储、数据库存储等。

from langchain_community.chat_message_histories import ChatMessageHistoryhistory = ChatMessageHistory()
#向聊天历史记录中添加用户发送的消息
history.add_user_message("你好今天过得怎么样")
#向聊天历史记录中添加 AI 生成的回复消息
history.add_ai_message("我没有真实的生活体验，但随时为你服务！")
#获取聊天历史记录中的所有消息，返回一个消息列表。
print(history.messages)
#清空聊天历史记录中的所有消息。
history.clear()
print(len(history.messages))

常见子类及使用场景

`InMemoryChatMessageHistory`

将聊天消息存储在内存中，适用于短期会话或测试环境。

from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistoryfrom model.deepseek import deepseek_llmprompt = ChatPromptTemplate.from_messages([("system", "你是一个翻译助手，请将用户的输入翻译成英语"),MessagesPlaceholder(variable_name="history", optional=True),("human", "{input}")
])
runnable = prompt | deepseek_llm#存储到内存中
store = {}#获取会话历史
def get_session_history(session_id : str)-> BaseChatMessageHistory:if session_id not in store:store[session_id] = InMemoryChatMessageHistory()return store.get(session_id)#添加会话历史
with_history_runnable = RunnableWithMessageHistory(runnable,get_session_history,input_messages_key="input",history_messages_key="history"
)while True:user_input = input("请输入用户输入：")if user_input == "exit":breaksession_id = input("请输入会话ID：")response = with_history_runnable.invoke({"input": user_input}, config={"configurable": {"session_id": session_id}})print(response.content)print(store)

`RedisChatMessageHistory`

将聊天消息存储在 Redis 数据库中，适用于分布式系统或需要高并发访问的场景。

from urllib.parse import quotefrom langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.prompts import MessagesPlaceholder, ChatPromptTemplate
from langchain_core.runnables import RunnableWithMessageHistoryfrom model.deepseek import deepseek_llmprompt = ChatPromptTemplate.from_messages([("system", "你是一个翻译助手，请将用户的输入翻译成英语"),MessagesPlaceholder(variable_name="history", optional=True),("human", "{input}")
])password = "密码"
encoded_password = quote(password, safe='')
def get_messages_history(session_id: str)->RedisChatMessageHistory:return RedisChatMessageHistory(url=f"redis://:{encoded_password}@192.168.0.120:16379/0", session_id=session_id)with_history_runnable = RunnableWithMessageHistory(prompt | deepseek_llm,get_messages_history,input_messages_key="input",history_messages_key="history"
)resp = with_history_runnable.invoke({"input": "中午吃什么"}, config={"configurable": {"session_id": "123"}})
print(resp.content)

FileChatMessageHistory

将聊天消息存储在文件中，适用于需要持久化存储且数据量较小的场景。FileChatMessageHistory 并不支持多会话隔离，也就是说多个会话的聊天记录都会写入同一个文件中，导致数据混乱。所以我们每个session保存为一个文件

from langchain_community.chat_message_histories import FileChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory, InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistoryfrom model.deepseek import deepseek_llmprompt = ChatPromptTemplate.from_messages([("system", "你是一个翻译助手，请将用户的输入翻译成英语"),MessagesPlaceholder(variable_name="history", optional=True),("human", "{input}")
])
runnable = prompt | deepseek_llm#获取会话历史
def get_session_history(session_id : str)-> BaseChatMessageHistory:file_path = f"chat_history_{session_id}.json"return FileChatMessageHistory(file_path=file_path)#添加会话历史
with_history_runnable = RunnableWithMessageHistory(runnable,get_session_history,input_messages_key="input",history_messages_key="history"
)while True:user_input = input("请输入用户输入：")if user_input == "exit":breaksession_id = input("请输入会话ID：")response = with_history_runnable.invoke({"input": user_input}, config={"configurable": {"session_id": session_id}})print(response.content)

配置会话唯一键

在 LangChain 里，为了在多用户或者多会话场景下准确区分和管理不同的会话，需要配置会话唯一键。默认的会话id是session_id，也可以通过history_factory_config配置：

from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableWithMessageHistory, ConfigurableFieldSpec
from langchain_community.chat_message_histories import ChatMessageHistoryfrom model.deepseek import deepseek_llmprompt = ChatPromptTemplate.from_messages([("system", "你是一个翻译助手，请将用户的输入翻译成英语"),MessagesPlaceholder(variable_name="history", optional=True),("human", "{input}")
])
runnable = prompt | deepseek_llm#存储到内存中
store = {}#获取会话历史
def get_session_history(user_id: str, conversation_id: str) -> BaseChatMessageHistory:if (user_id, conversation_id) not in store:store[(user_id, conversation_id)] = ChatMessageHistory()return store[(user_id, conversation_id)]#添加会话历史
with_history_runnable = RunnableWithMessageHistory(runnable,get_session_history,input_messages_key="input",history_messages_key="history",history_factory_config=[ConfigurableFieldSpec(id="user_id",annotation=str,name="User ID",description="用户的唯一标识符。",default="",is_shared=True,),ConfigurableFieldSpec(id="conversation_id",annotation=str,name="Conversation ID",description="对话的唯一标识符。",default="",is_shared=True,),],
)while True:user_input = input("请输入用户输入：")if user_input == "exit":breakuser_id = input("请输入会话ID：")response = with_history_runnable.invoke({"input": user_input}, config={"configurable": {"user_id": user_id, "conversation_id": "345"}})print(response.content)print(store)

查看全文

http://www.dtcms.com/a/282846.html