LangChain大模型应用开发：消息管理与聊天历史存储

介绍

大家好，博主又来给大家分享知识了。今天要给大家分享的是LangChain中的消息管理与聊天历史存储。
在LangChain里，消息管理可精细区分用户、助手、系统等不同角色消息，有序调度处理，让交互更顺畅。而聊天历史存储则赋予模型 “记忆”，多轮对话时能参考过往记录，理解意图更精准，回复更连贯。二者相辅相成，为构建智能对话应用筑牢基础。希望大家能通过我本次的分享，对它们有更清晰的认知。

消息存储在内存

在LangChain的应用场景中，消息存储是保障多轮对话连贯性的关键环节。它主要负责留存聊天过程中的历史信息，使得模型在后续交互时能够参考过往对话，更精准地理解用户意图并生成回复。

内存存储是消息存储的方式之一。将消息存储在内存里，数据访问速度快，能即时响应对话需求，适用于一些对实时性要求高、会话规模相对较小的场景。不过，它也存在局限性，如内存容量有限，可能因数据量过大导致性能问题。

下面我们展示一个简单的示例，其中聊天历史保存在内存中，此处通过全局Python字典实现。

我们构建一个名为get_session_history的可调用对象，引用此字典以返回ChatMessageHistory实例。通过在运行时向RunnableWithMessageHistory传递配置，可以指定可调用对象的参数。默认情况下，期望配置参数是一个字符串session_id 。可以通过history_factory_config关键字参数进行调整。

开启新会话

此次调用开启了一个新的会话。用户询问了 "行列式是什么意思？"，由于是新会话，消息历史为空，模型会基于系统提示和当前问题进行回复。

完整代码

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI# 用于存储不同会话的聊天消息历史
store = {}# 初始化OpenAI的聊天模型，使用gpt-3.5-turbo
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo")# 创建聊天提示模板，包含系统指令、消息历史占位和用户输入
prompt = ChatPromptTemplate.from_messages([("system", "你是一名得力助手并且非常擅长{ability}"),MessagesPlaceholder(variable_name="history"),("human", "{input}"),
])# 将提示模板和聊天模型组合成一个可执行链
chain = prompt | chat_model# 定义函数，根据会话ID获取对应的聊天消息历史，若不存在则创建
def get_session_history(session_id: str) -> BaseChatMessageHistory:if session_id not in store:store[session_id] = ChatMessageHistory()return store[session_id]# 创建带消息历史管理功能的可运行对象
with_message_history = RunnableWithMessageHistory(chain,get_session_history,input_messages_key="input",history_messages_key="history",
)# 调用带消息历史管理的对象，传入用户问题和会话配置并获取结果
result = with_message_history.invoke({"ability": "数学", "input": "行列式是什么意思？"},config={"configurable": {"session_id": "qaz123"}},
)# 打印聊天模型针对用户问题给出的回复结果
print(result)

运行结果

content='在线性代数中，行列式是一个针对一个方阵的函数，它将一个方阵映射到一个标量。行列式可以通过方阵的元素来定义，用来描述方阵的性质和特征。行列式的值可以提供有关于方阵是否可逆、方阵的秩以及方阵的特征值等信息。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 105, 'prompt_tokens': 40, 'total_tokens': 145, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-e5230a21-a183-4df5-9a1f-2aedeef4fba0-0' usage_metadata={'input_tokens': 40, 'output_tokens': 105, 'total_tokens': 145, 'input_token_details': {}, 'output_token_details': {}}进程已结束，退出代码为 0

在同一会话中进行追问

再次使用相同的会话ID，用户询问 "什么?"。此时模型会结合之前的消息历史，也就是第一次询问的 "行列式是什么意思？" 以及对应的回复，来理解用户的追问意图并给出基于上下文的回答。

完整代码

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI# 用于存储不同会话的聊天消息历史
store = {}# 初始化OpenAI的聊天模型，使用gpt-3.5-turbo
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo")# 创建聊天提示模板，包含系统指令、消息历史占位和用户输入
prompt = ChatPromptTemplate.from_messages([("system", "你是一名得力助手并且非常擅长{ability}"),MessagesPlaceholder(variable_name="history"),("human", "{input}"),
])# 将提示模板和聊天模型组合成一个可执行链
chain = prompt | chat_model# 定义函数，根据会话ID获取对应的聊天消息历史，若不存在则创建
def get_session_history(session_id: str) -> BaseChatMessageHistory:if session_id not in store:store[session_id] = ChatMessageHistory()return store[session_id]# 创建带消息历史管理功能的可运行对象
with_message_history = RunnableWithMessageHistory(chain,get_session_history,input_messages_key="input",history_messages_key="history",
)# 调用带消息历史管理的对象，传入用户问题和会话配置并获取结果
result = with_message_history.invoke({"ability": "math", "input": "行列式是什么意思？"},config={"configurable": {"session_id": "qaz123"}},
)# 打印聊天模型针对用户问题给出的回复结果
print(result)# 记住
result = with_message_history.invoke({"ability": "数学", "input": "什么?"},config={"configurable": {"session_id": "qaz123"}},
)print(result)

 运行结果

content='在线性代数中，行列式是一个针对一个方阵的函数，它将一个方阵映射到一个标量。行列式可以通过方阵的元素来定义，用来描述方阵的性质和特征。行列式的值可以提供有关于方阵是否可逆、方阵的秩以及方阵的特征值等信息。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 105, 'prompt_tokens': 40, 'total_tokens': 145, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-e5230a21-a183-4df5-9a1f-2aedeef4fba0-0' usage_metadata={'input_tokens': 40, 'output_tokens': 105, 'total_tokens': 145, 'input_token_details': {}, 'output_token_details': {}}
content='抱歉，我的回答可能有点复杂了。简单来说，行列式就是一个针对方阵的数学工具，用来描述和分析方阵的性质和特征。行列式的值可以提供有关方阵性质的重要信息。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 79, 'prompt_tokens': 156, 'total_tokens': 235, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-2dbeed9e-9ab6-4910-a220-c9aadf83bfba-0' usage_metadata={'input_tokens': 156, 'output_tokens': 79, 'total_tokens': 235, 'input_token_details': {}, 'output_token_details': {}}进程已结束，退出代码为 0

再次开启新会话

使用新的会话ID，用户再次询问 "什么?"。由于是新会话，消息历史为空，模型无法获取之前关于行列式的对话信息，会将此次询问当作全新的问题进行处理，回复可能与之前不同。

完整代码

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI# 用于存储不同会话的聊天消息历史
store = {}# 初始化OpenAI的聊天模型，使用gpt-3.5-turbo
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo")# 创建聊天提示模板，包含系统指令、消息历史占位和用户输入
prompt = ChatPromptTemplate.from_messages([("system", "你是一名得力助手并且非常擅长{ability}"),MessagesPlaceholder(variable_name="history"),("human", "{input}"),
])# 将提示模板和聊天模型组合成一个可执行链
chain = prompt | chat_model# 定义函数，根据会话ID获取对应的聊天消息历史，若不存在则创建
def get_session_history(session_id: str) -> BaseChatMessageHistory:if session_id not in store:store[session_id] = ChatMessageHistory()return store[session_id]# 创建带消息历史管理功能的可运行对象
with_message_history = RunnableWithMessageHistory(chain,get_session_history,input_messages_key="input",history_messages_key="history",
)# 调用带消息历史管理的对象，传入用户问题和会话配置并获取结果
result = with_message_history.invoke({"ability": "math", "input": "行列式是什么意思？"},config={"configurable": {"session_id": "qaz123"}},
)# 打印聊天模型针对用户问题给出的回复结果
print(result)# 记住
result = with_message_history.invoke({"ability": "数学", "input": "什么?"},config={"configurable": {"session_id": "qaz123"}},
)print(result)# 新的session_id-->不记得了。
result = with_message_history.invoke({"ability": "math", "input": "什么?"},config={"configurable": {"session_id": "wsx234"}},
)print(result)

运行结果 

content='在线性代数中，行列式是一个针对一个方阵的函数，它将一个方阵映射到一个标量。行列式可以通过方阵的元素来定义，用来描述方阵的性质和特征。行列式的值可以提供有关于方阵是否可逆、方阵的秩以及方阵的特征值等信息。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 105, 'prompt_tokens': 40, 'total_tokens': 145, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-e5230a21-a183-4df5-9a1f-2aedeef4fba0-0' usage_metadata={'input_tokens': 40, 'output_tokens': 105, 'total_tokens': 145, 'input_token_details': {}, 'output_token_details': {}}
content='抱歉，我的回答可能有点复杂了。简单来说，行列式就是一个针对方阵的数学工具，用来描述和分析方阵的性质和特征。行列式的值可以提供有关方阵性质的重要信息。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 79, 'prompt_tokens': 156, 'total_tokens': 235, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-2dbeed9e-9ab6-4910-a220-c9aadf83bfba-0' usage_metadata={'input_tokens': 156, 'output_tokens': 79, 'total_tokens': 235, 'input_token_details': {}, 'output_token_details': {}}
content='抱歉，我是不是给你带来了困惑？需要我为你解答什么数学问题吗？' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 36, 'prompt_tokens': 34, 'total_tokens': 70, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-ac1e77d1-8956-447d-8282-db2a2821734e-0' usage_metadata={'input_tokens': 34, 'output_tokens': 36, 'total_tokens': 70, 'input_token_details': {}, 'output_token_details': {}}进程已结束，退出代码为 0

综上，这段代码主要实现了一个基于LangChain的聊天系统，该系统可以管理不同会话的聊天消息历史，并利用这些历史信息为用户提供上下文感知的回复。

具体来说，它创建了一个带有消息历史管理功能的可运行对象with_message_history，通过不同的session_id来区分不同的会话，在每次调用invoke方法时，会根据session_id维护对应的消息历史，并将其作为上下文传递给语言模型。

配置会话唯一键

我们可以通过向history_factory_config参数传递一个ConfigurableFieldSpec对象列表来自定义跟踪消息历史的配置参数。下面我们使用了两个参数：user_id和conversation_id。

我们配置user_id和 conversation_id作为会话唯一键。

完整代码

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI
from langchain_core.runnables import ConfigurableFieldSpec# 初始化一个字典用于存储不同用户和对话的消息历史
store = {}# 初始化OpenAI的聊天模型，使用gpt-3.5-turbo
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo")# 创建聊天提示模板，包含系统指令、消息历史占位和用户输入
prompt = ChatPromptTemplate.from_messages([("system", "你是一名得力助手并且非常擅长{ability}"),MessagesPlaceholder(variable_name="history"),("human", "{input}"),
])# 将提示模板和聊天模型组合成一个可执行链
chain = prompt | chat_model# 定义函数，根据用户ID和对话ID获取或创建对应的消息历史
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)]# 创建带消息历史管理功能的可运行对象，并配置用户和对话ID相关参数
with_message_history = RunnableWithMessageHistory(chain,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,),],
)# 第一次调用，传入用户123在对话1中询问行列式的含义并打印结果
result = with_message_history.invoke({"ability": "数学", "input": "行列式是什么意思"},config={"configurable": {"user_id": "123", "conversation_id": "1"}},
)
print(result)# 第二次调用，同一用户在同一会话中追问并打印结果
result = with_message_history.invoke({"ability": "数学", "input": "什么？"},config={"configurable": {"user_id": "123", "conversation_id": "1"}},
)
print(result)# 第三次调用，不同用户在同一会话中追问并打印结果
result = with_message_history.invoke({"ability": "数学", "input": "什么？"},config={"configurable": {"user_id": "456", "conversation_id": "1"}},
)
print(result)

运行结果

content='行列式是矩阵特有的一种性质，是一个方阵（行数与列数相等的矩阵）所具有的一个标量值。行列式的计算方法比较复杂，但简单来说，行列式的值表示了矩阵线性变换所导致的尺度变化的比例系数。行列式的值为零表示矩阵不可逆，非零表示矩阵可逆。行列式在代数、几何学、物理学等领域有着广泛应用。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 160, 'prompt_tokens': 40, 'total_tokens': 200, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-9974ad1b-452f-4878-ab3d-916dc729a9de-0' usage_metadata={'input_tokens': 40, 'output_tokens': 160, 'total_tokens': 200, 'input_token_details': {}, 'output_token_details': {}}
content='抱歉，我的解释可能有点晦涩难懂。简单来说，行列式可以看作是一个方阵的重要属性，它的值可以帮助我们了解这个矩阵的特性，比如是否可逆、线性变换的比例系数等。行列式的计算方法可能有点复杂，但在数学和其他领域中扮演着非常重要的角色。希望这次解释更容易理解一些！如果你有任何具体的问题或者需要进一步的解释，请随时告诉我。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 165, 'prompt_tokens': 212, 'total_tokens': 377, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-cf46774c-e380-4695-9989-9661a2d02748-0' usage_metadata={'input_tokens': 212, 'output_tokens': 165, 'total_tokens': 377, 'input_token_details': {}, 'output_token_details': {}}
content='对不起，我想说的是我是一名得力助手，而且非常擅长数学。有什么数学问题我可以帮你解答吗？' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 50, 'prompt_tokens': 35, 'total_tokens': 85, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-e8ac3267-d81f-41ad-99fd-9b81bb8d9b0e-0' usage_metadata={'input_tokens': 35, 'output_tokens': 50, 'total_tokens': 85, 'input_token_details': {}, 'output_token_details': {}}进程已结束，退出代码为 0

我们看到上述代码基于LangChain库构建了一个具备消息历史管理功能的智能聊天系统，能与 OpenAI的gpt-3.5-turbo模型进行交互。我们看到第三次调用，不同的用户问同一个会话问题会变成一个新的会话。

系统会根据不同的用户ID(user_id)和对话ID(conversation_id)来管理和维护每个用户在不同对话场景下的消息历史，在每次与模型交互时，会将对应的消息历史作为上下文信息传递给模型，使得模型可以根据历史对话内容生成更符合当前语境的回复。

修改聊天历史

修改存储的聊天消息可以帮助我们的聊天机器人处理各种情况。

裁剪消息

LLM和聊天模型都有有限的上下文窗口，即使我们没有直接达到限制，我们可能也希望限制模型处理的干扰量。一种解决方案是只加载和存储最近的N条消息。让我们使用一个带有一些预加载消息的示例历史记录：

预加载完整代码

# 导入LangChain中用于管理聊天消息历史的模块
from langchain_community.chat_message_histories import ChatMessageHistory
# 导入用于创建人类消息和AI消息的类
from langchain_core.messages import HumanMessage, AIMessage
# 导入用于创建聊天提示模板和消息占位符的类
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# 导入用于处理带有消息历史的可运行对象的类
from langchain_core.runnables.history import RunnableWithMessageHistory
# 导入用于与OpenAI聊天模型交互的类
from langchain_openai import ChatOpenAI# 初始化OpenAI的聊天模型，使用gpt-3.5-turbo作为模型
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo")# 创建一个聊天消息历史对象，用于存储聊天记录
temp_chat_history = ChatMessageHistory()
temp_chat_history.add_user_message("我叫Ethan，你好")
temp_chat_history.add_ai_message("你好")
temp_chat_history.add_user_message("我今天心情特别好")
temp_chat_history.add_ai_message("你今天心情怎么样")
temp_chat_history.add_user_message("我上午在踢足球")
temp_chat_history.add_ai_message("你上午在做什么")# 直接修改聊天历史记录，以列表形式重新设置聊天消息
temp_chat_history.messages = [HumanMessage(content='我叫Ethan，你好'), AIMessage(content='你好'), HumanMessage(content='我今天心情特别好'), AIMessage(content='你今天心情怎么样'), HumanMessage(content='我上午在踢足球'), AIMessage(content='你上午在做什么'), HumanMessage(content='我今天心情怎么样?'), AIMessage(content='你今天心情特别好。')]# 根据消息列表创建一个聊天提示模板
prompt = ChatPromptTemplate.from_messages([("system","你是一个乐于助人的助手。尽力回答所有问题。提供的聊天历史包括与您交谈的用户的事实。",),MessagesPlaceholder(variable_name="chat_history"),("human", "{input}"),]
)# 将提示模板和聊天模型组合成一个可运行的链
chain = prompt | chat_model# 创建一个带有消息历史功能的可运行对象
chain_with_message_history = RunnableWithMessageHistory(# 传入前面组合好的链chain,# 一个函数，用于根据会话ID获取聊天历史记录，这里直接返回之前创建的聊天历史对象lambda session_id: temp_chat_history,# 指定输入消息的键名input_messages_key="input",# 指定历史消息的键名history_messages_key="chat_history",
)# 调用带有消息历史功能的可运行对象，传入用户新输入和配置信息
result = chain_with_message_history.invoke(# 用户新输入的问题{"input": "我今天心情如何?"},# 配置信息，指定会话ID{"configurable": {"session_id": "unused"}},
)# 打印AI针对用户新输入问题的回复结果
print(result)

预加载运行结果

content='你今天心情特别好。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 9, 'prompt_tokens': 159, 'total_tokens': 168, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-5ccbbbfc-abf0-4344-90cb-9a33bc476951-0' usage_metadata={'input_tokens': 159, 'output_tokens': 9, 'total_tokens': 168, 'input_token_details': {}, 'output_token_details': {}}进程已结束，退出代码为 0

从上面可以看到，Chain记住了预加载的地点。

但是假设我们有一个非常小的上下文窗口，并且我们想要将传递给链的消息数量减少到最近的2 条。我们可以使用clear方法来删除消息并重新将它们添加到历史记录中。我们不一定要这样做，但让我们将这个方法放在链的最前面，以确保它总是被调用：

近两条完整代码

# 导入LangChain中用于管理聊天消息历史的模块
from langchain_community.chat_message_histories import ChatMessageHistory
# 导入用于创建人类消息和AI消息的类
from langchain_core.messages import HumanMessage, AIMessage
# 导入用于创建聊天提示模板和消息占位符的类
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# 导入用于处理带有消息历史的可运行对象的类
from langchain_core.runnables.history import RunnableWithMessageHistory
# 导入用于与OpenAI聊天模型交互的类
from langchain_openai import ChatOpenAI
# 导入RunnablePassthrough类，该类用于在链式调用中直接传递输入，可结合其他操作对输入进行处理，常用于构建复杂的可运行对象链
from langchain_core.runnables import RunnablePassthrough# 初始化OpenAI的聊天模型，使用gpt-3.5-turbo作为模型
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo")# 创建一个聊天消息历史对象，用于存储聊天记录
temp_chat_history = ChatMessageHistory()
temp_chat_history.add_user_message("我叫Ethan，你好")
temp_chat_history.add_ai_message("你好")
temp_chat_history.add_user_message("我今天心情特别好")
temp_chat_history.add_ai_message("你今天心情怎么样")
temp_chat_history.add_user_message("我上午在踢足球")
temp_chat_history.add_ai_message("你上午在做什么")# 根据消息列表创建一个聊天提示模板
prompt = ChatPromptTemplate.from_messages([("system","你是一个乐于助人的助手。尽力回答所有问题。提供的聊天历史包括与您交谈的用户的事实。",),MessagesPlaceholder(variable_name="chat_history"),("human", "{input}"),]
)# 将提示模板和聊天模型组合成一个可运行的链
chain = prompt | chat_model# 创建一个带有消息历史功能的可运行对象
chain_with_message_history = RunnableWithMessageHistory(# 传入前面组合好的链chain,# 一个函数，用于根据会话ID获取聊天历史记录，这里直接返回之前创建的聊天历史对象lambda session_id: temp_chat_history,# 指定输入消息的键名input_messages_key="input",# 指定历史消息的键名history_messages_key="chat_history",
)# 定义一个函数，用于裁剪聊天历史记录，该函数接收链的输入作为参数
def trim_messages(chain_input):stored_messages = temp_chat_history.messagesif len(stored_messages) <= 2:return Falsetemp_chat_history.clear()for message in stored_messages[-2:]:temp_chat_history.add_message(message)return True# 创建一个新的可运行对象链，该链集成了消息裁剪和消息历史功能
chain_with_trimming = (RunnablePassthrough.assign(messages_trimmed=trim_messages)| chain_with_message_history
)# 调用带有消息裁剪功能和消息历史的链，传入用户输入和配置信息
result = chain_with_trimming.invoke(# 用户输入的新问题{"input": "我上午在做什么"},# 配置信息，指定会话ID{"configurable": {"session_id": "unused"}},
)# 打印AI针对用户新输入问题的回复结果
print(result)
print(temp_chat_history.messages)

近两条运行结果

content='你上午在踢足球。踢得开心吗？' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 20, 'prompt_tokens': 93, 'total_tokens': 113, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-032774e1-4e86-4309-a639-15a924f510e3-0' usage_metadata={'input_tokens': 93, 'output_tokens': 20, 'total_tokens': 113, 'input_token_details': {}, 'output_token_details': {}}
[HumanMessage(content='我上午在踢足球', additional_kwargs={}, response_metadata={}), AIMessage(content='你上午在做什么', additional_kwargs={}, response_metadata={}), HumanMessage(content='我上午在做什么', additional_kwargs={}, response_metadata={}), AIMessage(content='你上午在踢足球。踢得开心吗？', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 20, 'prompt_tokens': 93, 'total_tokens': 113, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None}, id='run-032774e1-4e86-4309-a639-15a924f510e3-0', usage_metadata={'input_tokens': 93, 'output_tokens': 20, 'total_tokens': 113, 'input_token_details': {}, 'output_token_details': {}})]进程已结束，退出代码为 0

我们重新调用这个新链并检查消息发现大模型已经知道我们在做什么了。

我们可以看到我们的历史记录已经删除了两条最旧的消息，同时在末尾添加了最近的对话。下次调用链时，trim_messages将再次被调用，只有最近的两条消息将被传递给模型。在这种情况下，这意味着下次调用时模型将忘记我们给它的名字。

删除询问完整代码

# 导入LangChain中用于管理聊天消息历史的模块
from langchain_community.chat_message_histories import ChatMessageHistory
# 导入用于创建聊天提示模板和消息占位符的类
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# 导入用于处理带有消息历史的可运行对象的类
from langchain_core.runnables.history import RunnableWithMessageHistory
# 导入用于与OpenAI聊天模型交互的类
from langchain_openai import ChatOpenAI
# 导入RunnablePassthrough类，该类用于在链式调用中直接传递输入，可结合其他操作对输入进行处理，常用于构建复杂的可运行对象链
from langchain_core.runnables import RunnablePassthrough# 初始化OpenAI的聊天模型，使用gpt-3.5-turbo作为模型
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo")# 创建一个聊天消息历史对象，用于存储聊天记录
temp_chat_history = ChatMessageHistory()
temp_chat_history.add_user_message("我叫Ethan，你好")
temp_chat_history.add_ai_message("你好")
temp_chat_history.add_user_message("我今天心情特别好")
temp_chat_history.add_ai_message("你今天心情怎么样")
temp_chat_history.add_user_message("我上午在踢足球")
temp_chat_history.add_ai_message("你上午在做什么")# 根据消息列表创建一个聊天提示模板
prompt = ChatPromptTemplate.from_messages([("system","你是一个乐于助人的助手。尽力回答所有问题。提供的聊天历史包括与您交谈的用户的事实。",),MessagesPlaceholder(variable_name="chat_history"),("human", "{input}"),]
)# 将提示模板和聊天模型组合成一个可运行的链
chain = prompt | chat_model# 创建一个带有消息历史功能的可运行对象
chain_with_message_history = RunnableWithMessageHistory(# 传入前面组合好的链chain,# 一个函数，用于根据会话ID获取聊天历史记录，这里直接返回之前创建的聊天历史对象lambda session_id: temp_chat_history,# 指定输入消息的键名input_messages_key="input",# 指定历史消息的键名history_messages_key="chat_history",
)# 定义一个函数，用于裁剪聊天历史记录，该函数接收链的输入作为参数
def trim_messages(chain_input):stored_messages = temp_chat_history.messagesif len(stored_messages) <= 2:return Falsetemp_chat_history.clear()for message in stored_messages[-2:]:temp_chat_history.add_message(message)return True# 创建一个新的可运行对象链，该链集成了消息裁剪和消息历史功能
chain_with_trimming = (RunnablePassthrough.assign(messages_trimmed=trim_messages)| chain_with_message_history
)# 调用带有消息裁剪功能和消息历史的链，传入用户输入和配置信息
result = chain_with_trimming.invoke({"input": "我叫什么名字?"},{"configurable": {"session_id": "unused"}},
)# 打印AI针对用户新输入问题的回复结果
print(result)
print(temp_chat_history.messages)

删除询问运行结果

content='很抱歉，我无法知道你的名字。如果你想告诉我，我会很高兴知道。有什么我可以帮你的吗？' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 49, 'prompt_tokens': 93, 'total_tokens': 142, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-5f2619e5-ea12-4fd7-8857-213b878d6dac-0' usage_metadata={'input_tokens': 93, 'output_tokens': 49, 'total_tokens': 142, 'input_token_details': {}, 'output_token_details': {}}
[HumanMessage(content='我上午在踢足球', additional_kwargs={}, response_metadata={}), AIMessage(content='你上午在做什么', additional_kwargs={}, response_metadata={}), HumanMessage(content='我叫什么名字?', additional_kwargs={}, response_metadata={}), AIMessage(content='很抱歉，我无法知道你的名字。如果你想告诉我，我会很高兴知道。有什么我可以帮你的吗？', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 49, 'prompt_tokens': 93, 'total_tokens': 142, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None}, id='run-5f2619e5-ea12-4fd7-8857-213b878d6dac-0', usage_metadata={'input_tokens': 93, 'output_tokens': 49, 'total_tokens': 142, 'input_token_details': {}, 'output_token_details': {}})]进程已结束，退出代码为 0

我们看到，此时大模型已经忘记我们叫什么名字了。

总结记忆

我们也可以以其他方式使用相同的模式。例如，我们可以使用额外的LLM调用来在调用链之前生成对话摘要。让我们重新创建我们的聊天历史和聊天机器人链：

重创聊天代码演示

# 创建一个聊天消息历史对象，用于存储聊天记录
temp_chat_history = ChatMessageHistory()
temp_chat_history.add_user_message("我叫Ethan，你好")
temp_chat_history.add_ai_message("你好")
temp_chat_history.add_user_message("我今天心情特别好")
temp_chat_history.add_ai_message("你今天心情怎么样")
temp_chat_history.add_user_message("我上午在踢足球")
temp_chat_history.add_ai_message("你上午在做什么")# 直接修改聊天历史记录，以列表形式重新设置聊天消息
temp_chat_history.messages = [HumanMessage(content='我叫Ethan，你好'), AIMessage(content='你好'),HumanMessage(content='我今天心情特别好'), AIMessage(content='你今天心情怎么样'),HumanMessage(content='我上午在踢足球'), AIMessage(content='你上午在做什么'),HumanMessage(content='我今天心情怎么样?'), AIMessage(content='最为一个人工智能，我无法知道您的心情。您可以告诉我您今天感觉如何，我会尽我所能提供帮助')]

我们稍微修改提示，让LLM意识到它将收到一个简短摘要而不是聊天历史。

简短摘要代码演示

# 根据消息列表创建一个聊天提示模板
prompt = ChatPromptTemplate.from_messages([("system","你是一个乐于助人的助手。尽力回答所有问题。提供的聊天历史包括与您交谈的用户的事实。",),MessagesPlaceholder(variable_name="chat_history"),("user", "{input}"),]
)# 将提示模板和聊天模型组合成一个可运行的链
chain = prompt | chat_model# 创建一个带有消息历史功能的可运行对象
chain_with_message_history = RunnableWithMessageHistory(# 传入前面组合好的链chain,# 一个函数，用于根据会话ID获取聊天历史记录，这里直接返回之前创建的聊天历史对象lambda session_id: temp_chat_history,# 指定输入消息的键名input_messages_key="input",# 指定历史消息的键名history_messages_key="chat_history",
)

现在，让我们创建一个函数，将之前的交互总结为摘要。我们也可以将这个函数添加到链的最前面。

创建摘要代码演示

# 定义一个函数，用于对聊天历史消息进行总结
def summarize_messages(chain_input):# 从聊天历史对象中获取存储的消息stored_messages = temp_chat_history.messages# 如果存储的消息数量为 0，即没有聊天记录，则直接返回 Falseif len(stored_messages) == 0:return False# 创建一个用于总结消息的提示模板，该模板会将历史聊天消息和总结要求传递给模型summarization_prompt = ChatPromptTemplate.from_messages([# 占位符，用于插入聊天历史消息MessagesPlaceholder(variable_name="chat_history"),("user",# 要求模型将上述聊天消息浓缩成一条摘要消息，并尽可能包含多个具体细节"将上述聊天消息浓缩成一条摘要消息。尽可能包含多个具体细节。",),])# 将总结提示模板和聊天模型组合成一个可运行的链，用于执行消息总结操作summarization_chain = summarization_prompt | chat_model# 调用总结链，传入聊天历史消息，得到总结后的消息summary_message = summarization_chain.invoke({"chat_history": stored_messages})# 清空原有的聊天历史记录temp_chat_history.clear()# 将总结后的消息添加到聊天历史记录中temp_chat_history.add_message(summary_message)# 表示消息总结操作成功完成，返回 Truereturn True# 构建带有消息总结功能的可运行对象链
chain_with_summarization = (RunnablePassthrough.assign(messages_summarized=summarize_messages)| chain_with_message_history
)

我们在对大模型进行提问，看看它是否还记得我们的名字。

完整代码

# 导入LangChain中用于管理聊天消息历史的模块
from langchain_community.chat_message_histories import ChatMessageHistory
# 导入用于创建人类消息和AI消息的类
from langchain_core.messages import HumanMessage, AIMessage
# 导入用于创建聊天提示模板和消息占位符的类
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# 导入用于处理带有消息历史的可运行对象的类
from langchain_core.runnables.history import RunnableWithMessageHistory
# 导入用于与OpenAI聊天模型交互的类
from langchain_openai import ChatOpenAI
# 导入RunnablePassthrough类，该类用于在链式调用中直接传递输入，可结合其他操作对输入进行处理，常用于构建复杂的可运行对象链
from langchain_core.runnables import RunnablePassthrough# 初始化OpenAI的聊天模型，使用gpt-3.5-turbo作为模型
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo")# 创建一个聊天消息历史对象，用于存储聊天记录
temp_chat_history = ChatMessageHistory()
temp_chat_history.add_user_message("我叫Ethan，你好")
temp_chat_history.add_ai_message("你好")
temp_chat_history.add_user_message("我今天心情特别好")
temp_chat_history.add_ai_message("你今天心情怎么样")
temp_chat_history.add_user_message("我上午在踢足球")
temp_chat_history.add_ai_message("你上午在做什么")# 直接修改聊天历史记录，以列表形式重新设置聊天消息
temp_chat_history.messages = [HumanMessage(content='我叫Ethan，你好'), AIMessage(content='你好'),HumanMessage(content='我今天心情特别好'), AIMessage(content='你今天心情怎么样'),HumanMessage(content='我上午在踢足球'), AIMessage(content='你上午在做什么'),HumanMessage(content='我今天心情怎么样?'), AIMessage(content='最为一个人工智能，我无法知道您的心情。您可以告诉我您今天感觉如何，我会尽我所能提供帮助')]# 根据消息列表创建一个聊天提示模板
prompt = ChatPromptTemplate.from_messages([("system","你是一个乐于助人的助手。尽力回答所有问题。提供的聊天历史包括与您交谈的用户的事实。",),MessagesPlaceholder(variable_name="chat_history"),("user", "{input}"),]
)# 将提示模板和聊天模型组合成一个可运行的链
chain = prompt | chat_model# 创建一个带有消息历史功能的可运行对象
chain_with_message_history = RunnableWithMessageHistory(# 传入前面组合好的链chain,# 一个函数，用于根据会话ID获取聊天历史记录，这里直接返回之前创建的聊天历史对象lambda session_id: temp_chat_history,# 指定输入消息的键名input_messages_key="input",# 指定历史消息的键名history_messages_key="chat_history",
)# 定义一个函数，用于对聊天历史消息进行总结
def summarize_messages(chain_input):# 从聊天历史对象中获取存储的消息stored_messages = temp_chat_history.messages# 如果存储的消息数量为 0，即没有聊天记录，则直接返回 Falseif len(stored_messages) == 0:return False# 创建一个用于总结消息的提示模板，该模板会将历史聊天消息和总结要求传递给模型summarization_prompt = ChatPromptTemplate.from_messages([# 占位符，用于插入聊天历史消息MessagesPlaceholder(variable_name="chat_history"),("user",# 要求模型将上述聊天消息浓缩成一条摘要消息，并尽可能包含多个具体细节"将上述聊天消息浓缩成一条摘要消息。尽可能包含多个具体细节。",),])# 将总结提示模板和聊天模型组合成一个可运行的链，用于执行消息总结操作summarization_chain = summarization_prompt | chat_model# 调用总结链，传入聊天历史消息，得到总结后的消息summary_message = summarization_chain.invoke({"chat_history": stored_messages})# 清空原有的聊天历史记录temp_chat_history.clear()# 将总结后的消息添加到聊天历史记录中temp_chat_history.add_message(summary_message)# 表示消息总结操作成功完成，返回 Truereturn True# 构建带有消息总结功能的可运行对象链
chain_with_summarization = (RunnablePassthrough.assign(messages_summarized=summarize_messages)| chain_with_message_history
)# 调用带有消息总结功能和消息历史的链，传入用户输入和配置信息
result = chain_with_summarization.invoke({"input": "我叫什么名字？"},{"configurable": {"session_id": "unused"}},
)# 打印AI针对用户新输入问题的回复结果
print(result)# 打印当前的聊天历史记录
print(temp_chat_history.messages)

运行结果

content='您的名字是Ethan。' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 8, 'prompt_tokens': 90, 'total_tokens': 98, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None} id='run-85e2181b-e204-472c-b95c-dedcebec56ce-0' usage_metadata={'input_tokens': 90, 'output_tokens': 8, 'total_tokens': 98, 'input_token_details': {}, 'output_token_details': {}}
[AIMessage(content='Ethan今天上午在踢足球，心情特别好。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 20, 'prompt_tokens': 174, 'total_tokens': 194, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None}, id='run-0abfa757-9811-43a7-990d-08a1e56d557f-0', usage_metadata={'input_tokens': 174, 'output_tokens': 20, 'total_tokens': 194, 'input_token_details': {}, 'output_token_details': {}}), HumanMessage(content='我叫什么名字？', additional_kwargs={}, response_metadata={}), AIMessage(content='您的名字是Ethan。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 8, 'prompt_tokens': 90, 'total_tokens': 98, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_name': 'gpt-3.5-turbo-0125', 'system_fingerprint': 'fp_0165350fbb', 'finish_reason': 'stop', 'logprobs': None}, id='run-85e2181b-e204-472c-b95c-dedcebec56ce-0', usage_metadata={'input_tokens': 90, 'output_tokens': 8, 'total_tokens': 98, 'input_token_details': {}, 'output_token_details': {}})]进程已结束，退出代码为 0

我们再次调用链式模型会生成一个新的摘要，该摘要包括初始摘要以及新的消息等。我们还可以设计一种混合方法，其中一定数量的消息保留在聊天历史记录中，而其他消息则被总结成摘要。

结束

好了，以上就是本次分享的全部内容了。不知道大家是否掌握并了解了LangChain中的消息管理与聊天历史存储。

LangChain的消息管理与聊天历史存储在对话应用开发中很关键。消息管理含多种消息类型，且可灵活处理；聊天历史存储方式有本地文件、数据库、内存等；在与语言模型交互时，能传递上下文，开发者还可制定管理策略。

那么本次分享就到这了。最后，博主还是那句话：请大家多去大胆的尝试和使用，成功总是在不断的失败中试验出来的，敢于尝试就已经成功了一半。如果大家对博主分享的内容感兴趣或有帮助，请点赞和关注。大家的点赞和关注是博主持续分享的动力🤭，博主也希望让更多的人学习到新的知识。