21. LangChain组件——memory临时会话记忆
本帖最后由 糖逗 于 2026-3-8 08:07 编辑1. 如何临时会话记忆,让大模型的回答能基于上文内容?
LangChain提供了History功能,帮助模型在有历史记忆的情况下回答:
[*]基于RunnableWithMessageHistory在原有链的基础上创建带有历史记录功能的新链(新Runnable实例)
[*]基于InMemoryChatMessageHistory为历史记录提供内存存储(临时用)
注:①这里的会话ID是用于区分不同用户的不同会话内容
②临时会话记忆的缺点是一旦程序退出,则记忆丢失
2. 代码实战
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate, ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
model = ChatTongyi(model="qwen3-max")
# prompt = PromptTemplate.from_template(
# "你需要根据会话历史回应用户问题。对话历史:{chat_history},用户提问:{input},请回答"
# )
prompt = ChatPromptTemplate.from_messages(
[
("system", "你需要根据会话历史回应用户问题。对话历史:"),
MessagesPlaceholder("chat_history"),
("human", "请回答如下问题:{input}")
]
)
str_parser = StrOutputParser()
def print_prompt(full_prompt):
print("="*20, full_prompt.to_string(), "="*20)
return full_prompt
base_chain = prompt | print_prompt | model | str_parser
store = {} # key就是session,value就是InMemoryChatMessageHistory类对象
# 实现通过会话id获取InMemoryChatMessageHistory类对象
def get_history(session_id):
if session_id not in store:
store = InMemoryChatMessageHistory()
return store
# 创建一个新的链,对原有链增强功能:自动附加历史消息
conversation_chain = RunnableWithMessageHistory(
base_chain, # 被增强的原有chain
get_history, # 通过会话id获取InMemoryChatMessageHistory类对象
input_messages_key="input", # 表示用户输入在模板中的占位符
history_messages_key="chat_history" # 表示用户输入在模板中的占位符
)
if __name__ == '__main__':
# 固定格式,添加LangChain的配置,为当前程序配置所属的session_id
session_config = {
"configurable": {
"session_id": "user_001"
}
}
res = conversation_chain.invoke({"input": "小明有2个猫"}, session_config)
print("第1次执行:", res)
res = conversation_chain.invoke({"input": "小刚有1只狗"}, session_config)
print("第2次执行:", res)
res = conversation_chain.invoke({"input": "总共有几个宠物"}, session_config)
print("第3次执行:", res)
学习视频:【黑马程序员大模型RAG与Agent智能体项目实战教程,基于主流的LangChain技术从大模型提示词到实战项目】 https://www.bilibili.com/video/BV1yjz5BLEoY/?p=37&share_source=copy_web&vd_source=792a2cb63a1882bff4ed856eadc41a71
如何临时会话记忆,让大模型的回答能基于上文内容?
LangChain提供了History功能,帮助模型在有历史记忆的情况下回答:
基于RunnableWithMessageHistory在原有链的基础上创建带有历史记录功能的新链(新Runnable实例)
基于InMemoryChatMessageHistory为历史记录提供内存存储(临时用)
注:这里的会话ID是用于区分不同用户的不同会话内容
实现让大模型基于上文内容(临时会话记忆)回答的核心步骤如下:
1. 创建一个基础的对话链(base_chain)。
2. 定义一个存储聊天历史的字典(store)和一个根据会话ID获取或创建历史记录的函数(get_history)。
3. 使用 RunnableWithMessageHistory 来增强基础链,使其能够自动管理历史消息。
4. 调用时,通过 session_config 指定会话ID,以确保同一个会话的对话历史被连贯地使用。
你提供的代码思路正确,但有几处需要修正和完善的地方:
主要修正点:
[*] get_history 函数中,当 session_id 不存在时,需要将创建的 InMemoryChatMessageHistory 对象存入 store 字典。
[*] ChatPromptTemplate.from_messages 的内容是空的,需要补充消息结构。
[*] 对话链的配置参数名称需要根据你使用的 PromptTemplate 进行调整。
以下是修正和优化后的完整代码示例:
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import InMemoryChatMessageHistory
# 1. 初始化模型
model = ChatTongyi(model="qwen-max", temperature=0)
# 2. 构建Prompt模板,使用 MessagesPlaceholder 来预留历史消息的位置
prompt = ChatPromptTemplate.from_messages([
# 系统提示词,可以设定AI的角色或行为
("system", "你是一个乐于助人的助手。请根据对话历史和当前问题来回答。"),
# 这是历史消息的占位符,RunnableWithMessageHistory会自动将历史对话填充到这里
MessagesPlaceholder(variable_name="chat_history"),
# 用户最新问题的占位符
("human", "{input}")
])
# 3. 初始化输出解析器
str_parser = StrOutputParser()
# 4. 一个用于调试、打印当前完整Prompt的函数
def print_prompt(full_prompt):
print("=" * 40)
print(" 当前完整的Prompt内容:")
print(full_prompt.to_string())
print("=" * 40)
return full_prompt
# 5. 构建基础链
base_chain = prompt | print_prompt | model | str_parser
# 6. 创建一个字典来存储不同会话的历史记录
# key: session_id (字符串), value: InMemoryChatMessageHistory 对象
store = {}
def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
"""根据session_id获取或创建对应的聊天历史记录对象"""
if session_id not in store:
# 如果这个session_id是第一次出现,则为它创建一个新的历史记录存储对象
store = InMemoryChatMessageHistory()
return store
# 7. 使用 RunnableWithMessageHistory 增强基础链,使其具备历史记忆功能
conversation_chain = RunnableWithMessageHistory(
runnable=base_chain, # 要被增强的基础链
get_session_history=get_session_history, # 获取历史记录的函数
input_messages_key="input", # 用户最新输入在Prompt模板中对应的变量名
history_messages_key="chat_history" # 历史消息在Prompt模板中对应的变量名
)
if __name__ == '__main__':
# 定义会话配置,`session_id` 用于区分不同的对话上下文
config = {"configurable": {"session_id": "user_001"}}
# 第一轮对话
response_1 = conversation_chain.invoke({"input": "小明有2个猫"}, config=config)
print("第1次回答:", response_1)
# 第二轮对话
response_2 = conversation_chain.invoke({"input": "小刚有1只狗"}, config=config)
print("第2次回答:", response_2)
# 第三轮对话:此时模型应该能“记住”前面提到的猫和狗
response_3 = conversation_chain.invoke({"input": "他们总共有几个宠物?"}, config=config)
print("第3次回答:", response_3)
# 验证不同session_id的隔离性
print("\n--- 测试不同会话ID的隔离 ---")
another_config = {"configurable": {"session_id": "user_002"}}
response_new_session = conversation_chain.invoke({"input": "总共有几个宠物?"}, config=another_config)
print("新会话(user_002)的回答:", response_new_session)
代码要点解释:
[*]ChatPromptTemplate:消息模板清晰地分为系统指令、历史消息和用户当前输入三部分。MessagesPlaceholder 是容纳历史对话的关键。
[*]store 和 get_session_history:在内存中为每个 session_id 维护独立的对话历史。这是实现“临时”记忆的基础。
[*]RunnableWithMessageHistory:这个包装器自动完成繁琐的历史消息管理。每次调用时,它会:
[*]根据 session_id 从 get_session_history 函数中获取对应的历史记录对象。
[*]将历史记录中的所有消息填充到Prompt模板的 chat_history 占位符。
[*]将用户本次的输入 input 填充到对应位置。
[*]调用基础链生成回复后,将本轮的用户输入和AI回复自动添加到历史记录对象中,供下次使用。
[*]session_id 的重要性:它是区分不同对话上下文的唯一标识。例如,“user_001”和“user_002”拥有完全独立、互不干扰的对话历史。在实际应用中,这可以对应不同的用户或不同的聊天窗口。
[*]运行结果预期:在同一个 session_id(如“user_001”)下,第三次询问“总共有几个宠物?”时,模型能基于前两次对话的记忆,给出“3个”或类似的答案。而对于新的 session_id(如“user_002”),由于没有历史,模型无法直接回答数量问题。
通过以上方式,你就成功地为LangChain应用添加了临时的、基于会话的记忆功能。
以上回复来自 FishC AI 小助理,如未能正确解答您的问题,请继续追问。
页:
[1]