第五讲 | 对话记忆:临时Memory与长期持久化会话
·
一、前言
普通单次问答无法保留上下文,用户连续对话时模型不知道历史聊天内容。LangChain内置Memory组件解决上下文记忆问题。本讲区分临时会话记忆与长期持久化记忆两种方案,对比两者适用场景,实现连贯多轮对话,打造具备上下文感知的问答机器人。
二、Memory临时会话记忆
如果想要封装历史记录,除了自行维护历史消息外,也可以借助LangChain内置的历史记录附加功能。
LangChain提供了History功能,帮助模型在有历史记忆的情况下回答。
-基于RunnableWithMessageHistory在原有链的基础上创建带有历史记录功能的新链(新Runnable实例)
-基于InMemoryChatMessageHistory为历史记录提供内存存储(临时用)
from langchain_core.runnables.history import RunnableWithMessageHistory
# 通过RunnalbeWithMessageHistory获取一个新的带有历史记录功能的chain
conversation_chain = RunnableWithMessageHistory(
some_chain, # 被附加历史消息的Runnable,通常是chain
None, # 获取指定会话ID的历史会话的函数
input_messages_key="input", # 声明用户输入消息在模板中的占位
history_messages_key="chat_history" # 声明历史消息在模板中的占位符
)
# 获取指定会话ID的历史会话记录函数
chat_history_store = {} # 存放多个会话ID所对应的历史会话记录
# 函数传入会话ID(字符串类型)
# 函数要求返回BaseChatMessageHistory的子类
# BaseChatMessageHistory类专用于存放某个会话的历史记录
# InMemoryChatMessageHistory是官方自带的基于内存存放历史纪录的类
def get_history(session_id):
if session_id not in chat_history_store:
# 返回一个新的实例
chat_history_store[session_id] = InMemoryChatMessageHistory()
return chat_history_store[session_id]
from dotenv import load_dotenv
load_dotenv()
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
llm = ChatTongyi(model="qwen3-max")
prompt = ChatPromptTemplate.from_messages([
("system", "你是AI助手,结合历史对话回答问题"),
MessagesPlaceholder("history"),
("human", "{input}")
])
chain = prompt | llm
# 获取会话历史回调函数
def get_history(session_id):
return FileChatMessageHistory(session_id=session_id)
# 包装带记忆的链
with_history_chain = RunnableWithMessageHistory(
chain,
get_history,
input_messages_key="input",
history_messages_key="history"
)
# 测试对话,指定会话ID
res1 = with_history_chain.invoke(
{"input":"我叫小明"},
config={"configurable":{"session_id":"user_001"}}
)
print(res1.content)
res2 = with_history_chain.invoke(
{"input":"我叫什么名字?"},
config={"configurable":{"session_id":"user_001"}}
)
print(res2.content)
三、Memory长期会话记忆
FileChatMessageHistory类实现,核心思路:
-基于文件存储会话记录,以session_id为文件名,不同session_id有不同文件存储消息
继承BaseChatMessageHistory实现如下3个方法:
-add_messages:同步模式,添加消息
-messages:同步模式,获取消息
-clear:同步模式,清除消息
import json
import os
from typing import Sequence
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, message_to_dict, messages_from_dict
# 自定义文件会话历史类
class FileChatMessageHistory(BaseChatMessageHistory):
def __init__(self, session_id: str, storage_path: str = "./chat_history"):
self.session_id = session_id
self.storage_path = storage_path
os.makedirs(self.storage_path, exist_ok=True)
self.file_path = os.path.join(self.storage_path, self.session_id)
@property
def messages(self) -> list[BaseMessage]:
"""读取文件,获取历史消息"""
try:
with open(self.file_path, "r", encoding="utf-8") as f:
msg_list = json.load(f)
return messages_from_dict(msg_list)
except FileNotFoundError:
return []
def add_messages(self, messages: Sequence[BaseMessage]) -> None:
"""追加消息到文件"""
all_msg = self.messages
all_msg.extend(messages)
dict_data = [message_to_dict(m) for m in all_msg]
with open(self.file_path, "w", encoding="utf-8") as f:
json.dump(dict_data, f, ensure_ascii=False, indent=2)
def clear(self) -> None:
"""清空当前会话所有消息"""
if os.path.exists(self.file_path):
os.remove(self.file_path)
四、结语
本讲解决了大模型”无记忆、断上下文“的问题。区分了临时记忆与持久化记忆的使用场景,实现了连续多轮对话能力。至此我们的模型不再是单次问答机器人,而是具备上下文理解的智能对话体。
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐



所有评论(0)