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

基于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“有记忆、多会话”的连续对话。 代码实现（要持久化存储修改如下代码中存储和获取会话历史的方法即可实现，需要的自行修改，这里就演示核心的功能）：

1. from langchain_openai import ChatOpenAI
2. from langchain_core.tools import Tool
3. from langchain.agents import create_react_agent  # 改用 ReAct 智能体
4. from langchain.agents import AgentExecutor
5. from langchain_core.prompts import PromptTemplate  # ReAct 用 PromptTemplate 而非 ChatPromptTemplate
6. from langchain_core.messages import AIMessage, HumanMessage
7. from langchain_community.chat_message_histories import ChatMessageHistory
8. from langchain_core.runnables.history import RunnableWithMessageHistory
10. # 1. 初始化 LLM（保持不变）
11. DEEPSEEK_API_KEY = "123"  # 替换为实际的 API Key
12. llm = ChatOpenAI(
13.     api_key=DEEPSEEK_API_KEY,
14.     base_url="http://172.25.133.51:8085/v1",
15.     model="qwen3.5-27b-awq",
16.     temperature=0.3,
17.     max_tokens=1024,
18. )
20. # 2. 工具函数
21. def huawei_mall_search(query: str) -> str:
22.     """华为商城搜索工具"""
23.     print(f"[DEBUG] 工具被调用！搜索关键词：{query}")
24.     search_results = {
25.         "众测活动": "华为商城众测活动是让用户体验新品并反馈意见的活动。目前有Mate 60系列众测，参与可赢取礼品。",
26.         "手机": "华为商城最新手机：Mate 60系列、P60系列、nova系列等。",
27.         "笔记本": "华为MateBook X Pro、MateBook D系列笔记本电脑。",
28.         "手表": "华为Watch 4、Watch GT系列智能手表。",
29.         "默认": "请在华为商城官网查看详细信息或联系客服。"
30.     }
31.     for keyword in search_results:
32.         if keyword in query:
33.             return f"华为商城搜索结果：{search_results[keyword]}"
34.     return search_results["默认"]
36. # 3. 创建工具
37. huawei_tool = Tool(
38.     name="huawei_mall_search",
39.     description="查询华为商城相关信息，包括产品、活动、政策等",
40.     func=huawei_mall_search,
41. )
42. tools = [huawei_tool]
44. # 4. 定义 ReAct 提示词模板（关键修改！）
45. react_prompt = PromptTemplate.from_template("""
46. Answer the following questions as best you can. You have access to the following tools:
48. {tools}
50. Use the following format:
52. Question: the input question you must answer
53. Thought: you should always think about what to do
54. Action: the action to take, should be one of [{tool_names}]
55. Action Input: the input to the action
56. Observation: the result of the action
57. ... (this Thought/Action/Action Input/Observation can repeat N times)
58. Thought: I now know the final answer
59. Final Answer: the final answer to the original input question
61. Begin!
63. Previous conversation history:
64. {chat_history}
66. Question: {input}
67. Thought: {agent_scratchpad}
68. """)
70. #上面提示词中{chat_history}是记录历史会话记录的不能少
72. # 5. 创建 ReAct 智能体（关键修改！）
73. try:
74.     agent = create_react_agent(llm=llm, tools=tools, prompt=react_prompt)
75.     print("Agent 创建成功")
76. except Exception as e:
77.     print(f"创建 Agent 失败: {e}")
78.     exit()
80. # 6. 创建执行器（保持不变）
81. agent_executor = AgentExecutor(
82.     agent=agent,
83.     tools=tools,
84.     verbose=True,
85.     max_iterations=3,
86.     handle_parsing_errors=True,
87.     return_intermediate_steps=True
88. )
90. # 7. 测试（保持不变）
91. print("\n" + "=" * 60)
92. print("测试 Agent 工具调用")
93. print("=" * 60)
97. # 1. 用字典存储每个 session_id 的独立历史
98. session_histories = {}
100. # 2. 会话存储函数：为每个 session_id 创建/返回独立的历史
101. def get_session_history(session_id: str) -> ChatMessageHistory:
102.     if session_id not in session_histories:
103.         session_histories[session_id] = ChatMessageHistory()
104.     return session_histories[session_id]
106. # 3. 创建带历史的智能体
107. agent_with_chat_history = RunnableWithMessageHistory(
108.     agent_executor,
109.     get_session_history,  # 使用新的会话存储函数
110.     input_messages_key="input",
111.     history_messages_key="chat_history",
112. )
115. # 4. 测试不同 session_id
116. chat1 = agent_with_chat_history.invoke(
117.     {"input": "你好，我是小老虎"},
118.     config={"configurable": {"session_id": "tiger"}},
119. )
121. chat2 = agent_with_chat_history.invoke(
122.     {"input": "你好，我是小兔子"},
123.     config={"configurable": {"session_id": "rabbit"}},
124. )
126. # 5. 查看特定 session_id 的会话记录
127. print("tiger 的会话记录：")
128. for msg in session_histories["tiger"].messages:
129.     print(f"{msg.type}: {msg.content}")
131. print("\nrabbit 的会话记录：")
132. for msg in session_histories["rabbit"].messages:
133.     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.


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

Final Answer: 你好，小老虎！很高兴认识你。我是你的智能助手，有什么我可以帮助你的吗？比如查询华为商城的产品信息、优惠活动或者相关政策等，都可以告诉我哦！

> Finished chain.
Parent run b6d86ca8-bcb5-4e4e-8182-080f8381338e not found for run a05acc86-23a1-4fcc-9d89-ea8209b9a02e. Treating as a root run.


> Entering new AgentExecutor chain...
r.


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

Final Answer: 你好，小兔子！很高兴认识你～有什么我可以帮助你的吗？
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！