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

langchain学习笔记之消息存储在内存中的实现方法 引言背景消息存储在内存的实现方法消息完整存储：完整代码

引言

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

背景

在与大模型交互过程中，经常出现消息管理方面的问题。一些大模型的上下文窗口包含的 token \text{token} token数量往往是有限的，如何将有限的大模型 memory \text{memory} 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字。" ), # 历史消息相关的placeholder MessagesPlaceholder( variable_name="history" ), ( "human", "{prompt_input}" ) ] ) llm = get_model() # 创建一个chain式调用,和历史信息相关的可运行chain runnable = chat_prompt | llm return 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 InMemoryChatMessageHistory def 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 RunnableWithMessageHistory message_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字。" ), # 历史消息相关的placeholder MessagesPlaceholder( variable_name="history" ), ( "human", "{prompt_input}" ) ] ) llm = get_model() runnable = chat_prompt | llm return 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相互独立。