AI开发-python-langchain框架（3-19-智能问答-保留会话历史）

WI8LbH788

201人浏览 · 2026-06-22 20:35:30

WI8LbH788 · 2026-06-22 20:35:30 发布

于RunnableWithMessageHistory实现多会话隔离与持久化记忆

在构建LLM对话应用时，多会话隔离与对话历史持久化是从demo走向生产级的关键。本文基于ReAct智能体实践，拆解RunnableWithMessageHistory的会话隔离实现，及可扩展的数据库持久化方案，助力快速落地稳定对话应用。

一、核心需求：为什么需要会话隔离与持久化记忆？

LLM本身无状态，无法记住历史交互，实际应用中需解决两个核心问题：

　　1. 多用户会话隔离：确保不同用户对话互不干扰，每个用户拥有独立“记忆空间”；

　　2. 对话历史持久化：保障服务重启后历史不丢失，用户可跨会话续聊，无需重复说明背景。

二、实践核心：用RunnableWithMessageHistory实现多会话隔离

RunnableWithMessageHistory是会话管理核心工具，可自动关联历史、注入提示词，简化开发，核心逻辑如下：

　　1. 分配唯一会话ID（session_id），作为区分不同会话的核心标识；

　　2. 用户发起对话时，传入session_id，工具自动读取对应历史，填充至提示词{chat_history}占位符；

　　3. 对话结束后，自动将新消息（用户输入+AI回复）追加至对应会话历史，实现记忆延续。

优势：开发者无需关注历史拼接、存储逻辑，可专注智能体核心功能开发。

三、扩展升级：对话历史持久化（数据库落地方案）

基础场景的字典内存存储，重启服务会丢失数据，需替换为持久化数据库，核心是修改会话历史获取函数，适配不同存储场景。

（一）主流持久化方案选型

根据业务场景灵活选择，无需局限单一方案：

　　1. Redis：适配高并发、低延迟场景（如在线客服），读写快、支持过期策略，通过langchain-redis快速集成；

　　2. 关系型数据库（PostgreSQL/MySQL）：适配数据安全、需审计场景（如企业内部系统），支持事务，通过langchain-community集成；

　　3. SQLite：适配小型应用、本地部署/测试，轻量无独立服务，快速落地降低成本；

　　4. MongoDB：适配非结构化数据场景（如多模态对话），文档型存储，通过langchain-mongodb集成。

（二）持久化通用实现步骤

所有方案流程一致，可直接复用：

　　1. 安装对应存储依赖包，确保环境齐全；

　　2. 重写会话历史获取函数：根据session_id从数据库查询，无记录则初始化并写入；

　　3. 将新函数传入RunnableWithMessageHistory，替换内存存储逻辑；

　　4. 配置存储连接参数（地址、端口等），适配生产环境。

（三）持久化核心设计要点

避免数据混乱、性能瓶颈，需注意4点：

　　1. 会话与消息映射：通过session_id关联，保持消息时间顺序，还原完整对话；

　　2. 异常处理：数据库连接失败时，降级为内存存储，保障服务可用并记录日志；

　　3. 数据清理：设置会话过期策略（如30天未活跃删除），支持用户主动删除，符合合规；

　　4. 序列化兼容：将LLM消息对象（HumanMessage等）序列化为JSON存储，读取时反序列化。

四、生产级落地建议

聚焦性能、安全、可扩展性，确保应用稳定：

　　1. 性能优化：高并发配置数据库连接池，关系型数据库建立索引，用对话摘要压缩历史减少token消耗；

　　2. 安全合规：加密存储敏感信息，明确数据生命周期，支持用户删除历史；

　　3. 可扩展性：继承BaseChatMessageHistory实现自定义存储，通过配置动态切换存储方案，增加监控指标。

五、总结

RunnableWithMessageHistory通过session_id实现多会话隔离，自动管理历史注入，大幅简化开发；对话历史持久化核心是替换存储层，根据业务场景选择合适数据库，即可快速落地生产级记忆能力，实现LLM“有记忆、多会话”的连续对话。

代码实现（要持久化存储修改如下代码中存储和获取会话历史的方法即可实现，需要的自行修改，这里就演示核心的功能）：

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

from langchain_openai import ChatOpenAI

from langchain_core.tools import Tool

from langchain.agents import create_react_agent # 改用 ReAct 智能体

from langchain.agents import AgentExecutor

from langchain_core.prompts import PromptTemplate # ReAct 用 PromptTemplate 而非 ChatPromptTemplate

from langchain_core.messages import AIMessage, HumanMessage

from langchain_community.chat_message_histories import ChatMessageHistory

from langchain_core.runnables.history import RunnableWithMessageHistory

# 1. 初始化 LLM（保持不变）

DEEPSEEK_API_KEY = "123" # 替换为实际的 API Key

llm = ChatOpenAI(

api_key=DEEPSEEK_API_KEY,

base_url="http://172.25.133.51:8085/v1",

model="qwen3.5-27b-awq",

temperature=0.3,

max_tokens=1024,

)

# 2. 工具函数

def huawei_mall_search(query: str) -> str:

"""华为商城搜索工具"""

print(f"[DEBUG] 工具被调用！搜索关键词：{query}")

search_results = {

"众测活动": "华为商城众测活动是让用户体验新品并反馈意见的活动。目前有Mate 60系列众测，参与可赢取礼品。",

"手机": "华为商城最新手机：Mate 60系列、P60系列、nova系列等。",

"笔记本": "华为MateBook X Pro、MateBook D系列笔记本电脑。",

"手表": "华为Watch 4、Watch GT系列智能手表。",

"默认": "请在华为商城官网查看详细信息或联系客服。"

}

for keyword in search_results:

if keyword in query:

return f"华为商城搜索结果：{search_results[keyword]}"

return search_results["默认"]

# 3. 创建工具

huawei_tool = Tool(

name="huawei_mall_search",

description="查询华为商城相关信息，包括产品、活动、政策等",

func=huawei_mall_search,

)

tools = [huawei_tool]

# 4. 定义 ReAct 提示词模板（关键修改！）

react_prompt = PromptTemplate.from_template("""

Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer

Thought: you should always think about what to do

Action: the action to take, should be one of [{tool_names}]

Action Input: the input to the action

Observation: the result of the action

... (this Thought/Action/Action Input/Observation can repeat N times)

Thought: I now know the final answer

Final Answer: the final answer to the original input question

Begin!

Previous conversation history:

{chat_history}

Question: {input}

Thought: {agent_scratchpad}

""")

#上面提示词中{chat_history}是记录历史会话记录的不能少

# 5. 创建 ReAct 智能体（关键修改！）

try:

agent = create_react_agent(llm=llm, tools=tools, prompt=react_prompt)

print("Agent 创建成功")

except Exception as e:

print(f"创建 Agent 失败: {e}")

exit()

# 6. 创建执行器（保持不变）

agent_executor = AgentExecutor(

agent=agent,

tools=tools,

verbose=True,

max_iterations=3,

handle_parsing_errors=True,

return_intermediate_steps=True

)

# 7. 测试（保持不变）

print("\n" + "=" * 60)

print("测试 Agent 工具调用")

print("=" * 60)

# 1. 用字典存储每个 session_id 的独立历史

session_histories = {}

# 2. 会话存储函数：为每个 session_id 创建/返回独立的历史

def get_session_history(session_id: str) -> ChatMessageHistory:

if session_id not in session_histories:

session_histories[session_id] = ChatMessageHistory()

return session_histories[session_id]

# 3. 创建带历史的智能体

agent_with_chat_history = RunnableWithMessageHistory(

agent_executor,

get_session_history, # 使用新的会话存储函数

input_messages_key="input",

history_messages_key="chat_history",

)

# 4. 测试不同 session_id

chat1 = agent_with_chat_history.invoke(

{"input": "你好，我是小老虎"},

config={"configurable": {"session_id": "tiger"}},

)

chat2 = agent_with_chat_history.invoke(

{"input": "你好，我是小兔子"},

config={"configurable": {"session_id": "rabbit"}},

)

# 5. 查看特定 session_id 的会话记录

print("tiger 的会话记录：")

for msg in session_histories["tiger"].messages:

print(f"{msg.type}: {msg.content}")

print("\nrabbit 的会话记录：")

for msg in session_histories["rabbit"].messages:

print(f"{msg.type}: {msg.content}")

输出结果：

Agent 创建成功

============================================================
测试 Agent 工具调用
============================================================
Parent run 5a054763-47a5-4360-b529-78012bbde271 not found for run 7b1f1b02-d9ac-4da8-b935-034ee28d0b7c. Treating as a root run.

> Entering new AgentExecutor chain...
l.
</think>

Thought: 用户只是在打招呼自我介绍，这是一个简单的问候，不需要使用华为商城搜索工具来查询信息。我应该友好地回应用户的问候。