langchain学习笔记之消息存储在内存中的实现方法

引言

本节将介绍$\text{langchain}$将历史消息存储在内存中的实现方法。

背景

在与大模型交互过程中，经常出现消息管理方面的问题。一些大模型的上下文窗口包含的$\text{token}$数量往往是有限的，如何将有限的大模型$\text{memory}$合理利用是十分关键的问题。

再比如，在与大模型交互时，可能会产生一系列连续的对话，这些对话往往不是独立的，我们更期望大模型能够合理找出这些对话之间的语义联系，并从而给出更符合要求的答案。

消息存储在内存的实现方法

基于上述背景，第一个朴素的想法是：与大模型连续交互的过程中，大模型能够认识到若干个prompt之间的语义联系，从而给出合理的回复。

• 准备工作：将模型、prompt以及初始chain的部分进行定义，其中使用MessagePlaceholder给交互过程中产生的历史信息留下位置：

from langchain_community.llms import Tongyi
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder	def get_model():return Tongyi(model_name="tongyi-7b-chinese",temperature=0.5,max_tokens=100,)def get_runnable_chain():chat_prompt = ChatPromptTemplate.from_messages([("system","你是一个擅长{field_input}的智能助理,返回结果不超过200字。"),# 历史消息相关的placeholderMessagesPlaceholder(variable_name="history"),("human","{prompt_input}")])llm = get_model()# 创建一个chain式调用,和历史信息相关的可运行chainrunnable = chat_prompt | llmreturn runnable

• 创建一个store_message字典，在与大模型交互交互过程中，将历史会话记录存储在字典中：

store_message = {}

• 在用户prompt过程中，设置一个名为session_id的参数，目的是将相同session_id的prompt归结为具有语义联系的prompt。定义函数：get_session_history，该函数的目的是：将store_message中当前session_id包含的所有交互信息获取出来。若未获取消息，即session_id第一次出现在store_message中，则需要新创建一个ChatMessageHistory的对象，将消息存入其中：

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistorydef get_session_history(session_id: str) -> InMemoryChatMessageHistory:if session_id not in store_message:store_message[session_id] = ChatMessageHistory()return store_message[session_id]

• 创建一个包含历史会话记录的运行器：通过RunnableWithMessageHistory类，通过history的标识，将get_session_history中获取的历史会话记录，结合当前交互步骤的prompt_input(该部分中包含session_id)，映射在MessagePlaceHolder中，并最终生成包含交互历史记录的runnable_chain:

from langchain_core.runnables.history import RunnableWithMessageHistorymessage_history_runnable = RunnableWithMessageHistory(runnable=get_runnable_chain(),get_session_history=get_session_history,input_messages_key="prompt_input",history_messages_key="history")

• 最终使用该runnable_chain进行交互。示例：

response_1 = message_history_runnable.stream(input={"field_input": "历史科普","prompt_input": "简单介绍一下李白"},config={"configurable": {"session_id": "libai_introduction"}})for chunk in response_1:print(chunk, end="", flush=True)

首先是通过field_input对大模型进行角色定义，并提出prompt以及当前交互步骤的session_id。返回结果如下：

李白（701－762），字太白，号青莲居士，唐代著名诗人。出生于中亚碎叶城，少年时迁居四川。他性格豪放不羁，好饮酒作乐，游历名山大川，留下大量诗篇。其诗歌风格飘逸洒脱、意境开阔，充满浪漫主义色彩，善于运用夸张手法和奇特想象。李白与杜甫并称“李杜”，代表作品有《将进酒》、《静夜思》、《望庐山瀑布》等，对后世影响深远。安史之乱爆发后，因参与永王李璘起兵而获罪流放，晚年生活困顿，在当涂病逝。

• 继续执行，第二次交互的respense_2表示如下：

response_2 = message_history_runnable.stream(input={"field_input": "历史科普","prompt_input": "他具体受到哪些政治迫害?"},config={"configurable": {"session_id": "libai_introduction"}})

需要注意的是，仅从response_2交互自身，我们无法知晓prompt_input中的他描述的具体是谁，但由于与response_1共享同一个session_id，结合历史会话信息，能够得到这个他描述的是李白。返回结果如下：

李白在安史之乱期间因卷入永王李璘的起兵事件而遭受政治迫害。756年，永王李璘起兵东下，李白应邀加入其幕府。然而，李璘与唐肃宗争夺帝位失败，李白因此获罪被捕入狱。虽经友人营救得以免死，但仍被流放夜郎（今贵州一带）。后因朝廷大赦，李白途中遇赦返回，但晚年生活穷困潦倒，最终客死当涂。这次政治牵连对李白的晚年生活和创作产生了重大影响。

创建一个反例：基于response_2，若prompt_input不变，但调整session_id：

response_3 = message_history_runnable.stream(input={"field_input": "历史科普","prompt_input": "他具体受到哪些政治迫害?"},config={"configurable": {"session_id": "libai_politics"}})

预期结果是：大模型不清楚这个他指代的是谁。返回结果如下：

你提到的政治迫害对象不明确呢。如果你是指历史上某个特定人物遭受的政治迫害，比如屈原，他因谗言被楚怀王疏远，放逐汉北；或岳飞被秦桧以“莫须有”的罪名陷害等，你可以具体说说你关注的人物哦，这样我能更准确作答。

至少和李白没什么关系~

消息完整存储：完整代码

# 引入聊天信息历史记录
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_community.llms import Tongyi# 用于储存历史会话记录
store_message = {}def get_model():return Tongyi(model_name="tongyi-7b-chinese",temperature=0.5,max_tokens=100,)def get_runnable_chain():chat_prompt = ChatPromptTemplate.from_messages([("system","你是一个擅长{field_input}的智能助理,返回结果不超过200字。"),# 历史消息相关的placeholderMessagesPlaceholder(variable_name="history"),("human","{prompt_input}")])llm = get_model()runnable = chat_prompt | llmreturn runnable# 获取历史会话
def get_session_history(session_id: str) -> InMemoryChatMessageHistory:if session_id not in store_message:# 没有查到session_id,使用ChatMessageHistory做一个初始化store_message[session_id] = ChatMessageHistory()return store_message[session_id]message_history_runnable = RunnableWithMessageHistory(runnable=get_runnable_chain(),get_session_history=get_session_history,input_messages_key="prompt_input",history_messages_key="history")if __name__ == '__main__':response_1 = message_history_runnable.stream(input={"field_input": "历史科普","prompt_input": "简单介绍一下李白"},config={"configurable": {"session_id": "libai_introduction"}})for chunk in response_1:print(chunk, end="", flush=True)print("\n")print("-----" * 30)response_2 = message_history_runnable.stream(input={"field_input": "历史科普","prompt_input": "他具体受到哪些政治迫害?"},config={"configurable": {"session_id": "libai_introduction"}})for chunk in response_2:print(chunk, end="", flush=True)print("\n")print("-----" * 30)response_3 = message_history_runnable.stream(input={"field_input": "历史科普","prompt_input": "他具体受到哪些政治迫害?"},config={"configurable": {"session_id": "libai_politics"}})for chunk in response_3:print(chunk, end="", flush=True)

观察一下执行了三次交互后的store_message:

{'libai_introduction': InMemoryChatMessageHistory(messages=[HumanMessage(content='简单介绍一下李白', additional_kwargs={}, response_metadata={}), AIMessage(content='李白（701-762），字太白，号青莲居士，唐代伟大诗人。生于绵州昌隆，祖籍陇西成纪。其诗风豪放飘逸，想象丰富，语言流转自然，音律和谐多变。他喜好饮酒作乐，常与友人畅饮赋诗，留下“斗酒诗百篇”的佳话。代表作有《静夜思》《望庐山瀑布》等。李白一生游历名山大川，交友广泛，曾任翰林供奉，后因卷入永王李璘事件被流放夜郎，途中遇赦返回，晚年生活困顿，病逝于当涂。他与杜甫并称为“李杜”，对后世影响深远。', additional_kwargs={}, response_metadata={}), HumanMessage(content='他具体受到哪些政治迫害?', additional_kwargs={}, response_metadata={}), AIMessage(content='李白在安史之乱期间，因卷入永王李璘的起兵事件而遭受政治迫害。永王李璘是唐玄宗之子，在安禄山叛乱时，他试图争夺帝位，李白误以为他是中兴之主，便加入其幕府。然而，永王兵败后，李白被指控参与谋反，获罪下狱，后被判流放夜郎（今贵州一带）。幸而在流放途中遇赦免，得以返回。这次政治风波对李白晚年生活影响极大，也使他失去了仕途机会。', additional_kwargs={}, response_metadata={})]), 'libai_politics': InMemoryChatMessageHistory(messages=[HumanMessage(content='他具体受到哪些政治迫害?', additional_kwargs={}, response_metadata={}), AIMessage(content='你提到的政治迫害对象不明确呢。如果你是指历史上某个特定人物遭受的政治迫害，比如屈原，他因谗言被楚怀王疏远，放逐汉北；或岳飞被秦桧以“莫须有”的罪名陷害等，你可以具体说说你关注的人物哦，这样我能更准确作答。', additional_kwargs={}, response_metadata={})])
}

很明显，store_message中的两个session_id：libai_introduction和libai_politics相互独立。