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

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

news 来源：原创 2025/5/1 16:08:01

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字。"
            ),
            # 历史消息相关的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={}
			)
		]
	)
}