【LangChain】P7 对话记忆完全指南：从原理到实战（下）

本篇博文为【LangChain】系列“对话记忆完全指南：从原理到实战”第三篇博文，前两篇围绕基础介绍、简单示例以及一个酒店服务的实例展开，本篇内容将围绕进阶内容展开。

详细内容请访问：

• 【LangChain】P5 对话记忆完全指南：从原理到实战（上）
• 【LangChain】P6 对话记忆完全指南：从原理到实战（中）

第五部分：进阶技巧与生产实践

5.1 持久化存储：对话不丢失

问题：内存中的数据会丢失

# 当前的实现
bot = HotelServiceBot(llm)
bot.chat("user_a", "我叫李明")# 如果程序重启...
# ❌ 所有对话历史都丢失了！

解决方案：存储到数据库

import json
from pathlib import Path
from datetime import datetimeclass PersistentConversationManager:"""支持持久化的对话管理器"""def __init__(self, llm, session_id: str, storage_path: str = "./conversations"):"""参数说明：- session_id: 用户的唯一标识（如 user_123）- storage_path: 对话文件存储目录"""self.llm = llmself.session_id = session_idself.storage_path = Path(storage_path)self.storage_path.mkdir(exist_ok=True)  # 创建目录self.messages = []# 程序启动时，自动加载历史对话self._load_history()def _get_file_path(self) -> Path:"""获取当前用户的对话文件路径"""return self.storage_path / f"{self.session_id}.json"def _load_history(self):"""从文件加载历史对话"""file_path = self._get_file_path()if not file_path.exists():print(f"[系统] 用户 {self.session_id} 是新用户，创建新会话")return# 读取 JSON 文件with open(file_path, 'r', encoding='utf-8') as f:data = json.load(f)# 重建消息对象for msg_data in data['messages']:msg_type = msg_data['type']content = msg_data['content']if msg_type == 'system':self.messages.append(SystemMessage(content=content))elif msg_type == 'human':self.messages.append(HumanMessage(content=content))elif msg_type == 'ai':self.messages.append(AIMessage(content=content))print(f"[系统] 已加载用户 {self.session_id} 的历史对话（{len(self.messages)} 条消息）")def _save_history(self):"""保存历史对话到文件"""file_path = self._get_file_path()# 序列化消息messages_data = []for msg in self.messages:# 判断消息类型if isinstance(msg, SystemMessage):msg_type = 'system'elif isinstance(msg, HumanMessage):msg_type = 'human'elif isinstance(msg, AIMessage):msg_type = 'ai'else:continue  # 忽略未知类型messages_data.append({'type': msg_type,'content': msg.content,'timestamp': datetime.now().isoformat()})# 写入 JSON 文件with open(file_path, 'w', encoding='utf-8') as f:json.dump({'messages': messages_data}, f, ensure_ascii=False,  # 支持中文indent=2  # 格式化输出)print(f"[系统] 已保存对话历史到 {file_path}")def chat(self, user_input: str, system_prompt: str = None) -> str:"""发送消息并自动保存"""# 如果是新会话且有系统提示if not self.messages and system_prompt:self.messages.append(SystemMessage(content=system_prompt))# 添加用户消息self.messages.append(HumanMessage(content=user_input))# 调用模型response = self.llm.invoke(self.messages)# 添加 AI 回复self.messages.append(AIMessage(content=response.content))# 💾 关键：每次对话后自动保存self._save_history()return response.contentdef clear_history(self):"""清空对话历史"""self.messages = []self._save_history()print(f"[系统] 用户 {self.session_id} 的对话已清空")

使用示例：持久化的威力

print("===== 第一次运行程序 =====")
manager1 = PersistentConversationManager(llm=chat_model,session_id="user_123",storage_path="./hotel_conversations"
)response1 = manager1.chat("你好，我叫李明，住在 301 房间",system_prompt="你是酒店客服小悦"
)
print(f"小悦: {response1}")response2 = manager1.chat("我需要一个加湿器")
print(f"小悦: {response2}")print("\n===== 模拟程序重启 =====")
# 重新创建管理器（模拟程序重启）
manager2 = PersistentConversationManager(llm=chat_model,session_id="user_123",  # 同一个用户IDstorage_path="./hotel_conversations"
)# 继续之前的对话
response3 = manager2.chat("你还记得我的名字和房间号吗？")
print(f"小悦: {response3}")

模型效果：成功加载历史消息

===== 第一次运行程序 =====
[系统] 用户 user_123 是新用户，创建新会话
[系统] 已保存对话历史到 hotel_conversations/user_123.json
小悦: 您好，李明先生！很高兴为您服务。请问有什么可以帮您的吗？
[系统] 已保存对话历史到 hotel_conversations/user_123.json
小悦: 好的，李明先生。我马上为您安排将加湿器送到301房间，预计10分钟内送达。请问您还有其他需要吗？===== 模拟程序重启 =====
[系统] 已加载用户 user_123 的历史对话（5 条消息）
[系统] 已保存对话历史到 hotel_conversations/user_123.json
小悦: 当然记得，您是住在301房间的李明先生。请问加湿器使用起来还满意吗？

查看存储：分类存储本地

完整内容：

{"messages": [{"type": "system","content": "你是酒店客服小悦","timestamp": "2025-10-02T00:06:38.343860"},{"type": "human","content": "你好，我叫李明，住在 301 房间","timestamp": "2025-10-02T00:06:38.343895"},{"type": "ai","content": "您好，李明先生！很高兴为您服务。请问有什么可以帮您的吗？","timestamp": "2025-10-02T00:06:38.343911"},{"type": "human","content": "我需要一个加湿器","timestamp": "2025-10-02T00:06:38.343920"},{"type": "ai","content": "好的，李明先生。我马上为您安排将加湿器送到301房间，预计10分钟内送达。请问您还有其他需要吗？","timestamp": "2025-10-02T00:06:38.343930"},{"type": "human","content": "你还记得我的名字和房间号吗？","timestamp": "2025-10-02T00:06:38.343938"},{"type": "ai","content": "当然记得，您是住在301房间的李明先生。请问加湿器使用起来还满意吗？","timestamp": "2025-10-02T00:06:38.343949"}]
}

---

5.2 Token 管理策略：控制成本

问题：Token 消耗快速增长

# 假设每条消息平均 50 tokens
第 1 轮：系统提示(100) + 用户(50) + AI(50) = 200 tokens
第 2 轮：200 + 用户(50) + AI(50) = 300 tokens
第 3 轮：300 + 用户(50) + AI(50) = 400 tokens
...
第 20 轮：2000 tokens（已经很贵了）
第 50 轮：5000 tokens（成本翻倍）

策略 1：按 Token 数量截断

def trim_messages_by_token(messages: list, max_tokens: int = 4000) -> list:"""根据 token 数量智能截断消息优点：精确控制成本缺点：需要额外的 token 计算"""try:from tiktoken import encoding_for_modelenc = encoding_for_model("gpt-3.5-turbo")except ImportError:print("⚠️ 需要安装 tiktoken: pip install tiktoken")return messages# 永远保留系统消息system_msgs = [m for m in messages if isinstance(m, SystemMessage)]other_msgs = [m for m in messages if not isinstance(m, SystemMessage)]# 计算系统消息的 tokentotal_tokens = sum(len(enc.encode(m.content)) for m in system_msgs)# 从后往前添加消息，直到达到限制trimmed_msgs = []for msg in reversed(other_msgs):msg_tokens = len(enc.encode(msg.content))if total_tokens + msg_tokens > max_tokens:break  # 达到限制，停止添加trimmed_msgs.insert(0, msg)  # 插入到开头（保持顺序）total_tokens += msg_tokensprint(f"[Token管理] 截断前: {len(other_msgs)} 条, 截断后: {len(trimmed_msgs)} 条, Token: {total_tokens}")return system_msgs + trimmed_msgs

策略 2：智能总结压缩

def summarize_if_needed(llm, messages: list, threshold: int = 20) -> list:"""当消息过多时，自动总结旧对话优点：保留关键信息，大幅减少 token缺点：总结本身需要一次 LLM 调用"""if len(messages) <= threshold:return messages  # 还没到阈值，不需要总结# 保留系统消息和最近的对话system_msgs = [m for m in messages if isinstance(m, SystemMessage)]old_msgs = [m for m in messages[1:-10] if not isinstance(m, SystemMessage)]recent_msgs = messages[-10:]  # 最近 5 轮对话if not old_msgs:return messages  # 没有旧消息需要总结# 格式化旧消息history_text = "\n".join([f"{'用户' if isinstance(m, HumanMessage) else 'AI'}: {m.content}"for m in old_msgs])# 调用 LLM 生成总结summary_prompt = f"""请简要总结以下对话的关键信息（不超过 150 字）：{history_text}总结要点：
1. 用户的基本信息和需求
2. 讨论的主要问题
3. 达成的结论或待办事项"""summary = llm.invoke([HumanMessage(content=summary_prompt)])summary_msg = SystemMessage(content=f"【历史对话摘要】\n{summary.content}")print(f"[智能总结] 已将 {len(old_msgs)} 条消息压缩为摘要")return system_msgs + [summary_msg] + recent_msgs

策略 3：混合方案

class TokenAwareConversationManager:"""Token 感知的对话管理器"""def __init__(self, llm, max_tokens: int = 3000, summary_threshold: int = 20):self.llm = llmself.messages = []self.max_tokens = max_tokensself.summary_threshold = summary_thresholddef chat(self, user_input: str) -> str:# 添加用户消息self.messages.append(HumanMessage(content=user_input))# 智能管理历史self._manage_history()# 调用模型response = self.llm.invoke(self.messages)self.messages.append(AIMessage(content=response.content))return response.contentdef _manage_history(self):"""智能历史管理：先尝试总结，再按 token 截断"""# 步骤 1：如果消息太多，先总结if len(self.messages) > self.summary_threshold:self.messages = summarize_if_needed(self.llm, self.messages, self.summary_threshold)# 步骤 2：确保不超过 token 限制self.messages = trim_messages_by_token(self.messages, self.max_tokens)

---

5.3 异步处理：提升并发性能

问题：同步调用效率低

# 同步方式：依次处理每个用户
for user_id in ["user_1", "user_2", "user_3"]:response = bot.chat(user_id, "你好")# 每个请求需要 1-2 秒# 总耗时：3-6 秒

解决方案：异步并发

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessageclass AsyncConversationManager:"""支持异步的对话管理器"""def __init__(self, llm):self.llm = llmself.sessions = {}async def chat(self, session_id: str, user_input: str, system_prompt: str = None) -> str:"""异步处理消息"""# 获取或创建会话if session_id not in self.sessions:self.sessions[session_id] = []if system_prompt:self.sessions[session_id].append(SystemMessage(content=system_prompt))messages = self.sessions[session_id]messages.append(HumanMessage(content=user_input))# 🚀 关键：使用 ainvoke 进行异步调用response = await self.llm.ainvoke(messages)messages.append(AIMessage(content=response.content))return response.content# 使用示例
async def main():llm = ChatOpenAI(model="gpt-3.5-turbo")manager = AsyncConversationManager(llm)# 🎯 并发处理多个用户请求tasks = [manager.chat("user_1", "推荐一本书", "你是图书管理员"),manager.chat("user_2", "今天天气怎么样", "你是气象助手"),manager.chat("user_3", "晚餐吃什么", "你是美食顾问")]# 等待所有请求完成responses = await asyncio.gather(*tasks)for i, resp in enumerate(responses, 1):print(f"用户{i}: {resp[:50]}...")# 运行异步任务
# asyncio.run(main())

性能对比：简单对比

同步方式：3 个请求，每个 1.5 秒 = 总计 4.5 秒
异步方式：3 个请求，并发执行 = 总计 1.6 秒（提速 65%）

📢 注意：上述内容需要读者注重 Python 异步通信能力，包含 async 以及 ainvoke 能力。该部分能力将在 Python 系列博文中展开。

---

5.4 监控与日志：生产必备

import logging
from datetime import datetime# 配置日志系统
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler('conversation.log'),  # 写入文件logging.StreamHandler()  # 同时输出到控制台]
)class MonitoredConversationManager:"""带监控的对话管理器"""def __init__(self, llm):self.llm = llmself.messages = []self.logger = logging.getLogger(__name__)# 统计指标self.stats = {'total_requests': 0,'total_tokens': 0,'avg_response_time': 0}def chat(self, user_input: str) -> str:"""带监控的对话"""start_time = datetime.now()try:# 记录请求self.logger.info(f"用户输入: {user_input[:100]}")# 添加用户消息self.messages.append(HumanMessage(content=user_input))# 调用模型response = self.llm.invoke(self.messages)self.messages.append(AIMessage(content=response.content))# 计算耗时elapsed = (datetime.now() - start_time).total_seconds()# 更新统计self.stats['total_requests'] += 1self.stats['avg_response_time'] = ((self.stats['avg_response_time'] * (self.stats['total_requests'] - 1) + elapsed)/ self.stats['total_requests'])# 记录响应self.logger.info(f"AI回复: {response.content[:100]}, "f"耗时: {elapsed:.2f}秒, "f"总请求数: {self.stats['total_requests']}")# ⚠️ 性能告警if elapsed > 3.0:self.logger.warning(f"⚠️ 响应时间过长: {elapsed:.2f}秒")return response.contentexcept Exception as e:self.logger.error(f"❌ 对话出错: {str(e)}", exc_info=True)raisedef get_stats(self):"""获取统计信息"""return self.stats

日志输出示例

2025-10-01 10:30:00,123 - __main__ - INFO - 用户输入: 你好，我叫李明
2025-10-01 10:30:01,456 - __main__ - INFO - AI回复: 你好李明！很高兴认识你, 耗时: 1.33秒, 总请求数: 1
2025-10-01 10:30:05,789 - __main__ - INFO - 用户输入: 推荐一本书
2025-10-01 10:30:09,012 - __main__ - WARNING - ⚠️ 响应时间过长: 3.22秒

📢 注意：上述内容需要读者注重 Python 监控日志能力，包含 logging 等能力。该部分能力将在 Python 系列博文中展开。

---

总结与思考

核心要点回顾

让我们回顾一下本文的核心知识点：

1. 大模型的记忆本质

┌─────────────────────────────────────┐
│  错误认知：大模型有记忆    	          │
│  "它记住了我的名字！"      	          │
└─────────────────────────────────────┘❌┌─────────────────────────────────────┐
│  正确理解：外部系统管理历史     	      │
│  "我们把历史消息一起传给它"     	      │
└─────────────────────────────────────┘✅

关键结论：

• 大模型本身无状态，每次调用都是独立的
• "记忆"是通过显式传递历史消息实现的
• 开发者需要自己管理消息历史和上下文

2. 技术方案演进

阶段	方案	特点	适用场景
传统	LangChain Memory	自动化程度高，但不够灵活	❌ 已弃用
现代	手动管理消息列表	代码透明，完全可控	✅ 新项目首选
进阶	LangGraph	复杂状态管理	🎯 复杂工作流

3. 实战的关键设计

多用户支持：

sessions = {"user_a": [消息列表],"user_b": [消息列表]
}

历史管理：

# 策略 1：保留最近 N 轮
messages = messages[-(max_turns * 2):]# 策略 2：总结 + 保留
summary + recent_messages

持久化：

# 保存到文件/数据库
json.dump(messages, file)# 程序重启后加载
messages = json.load(file)

最佳实践建议

1. 从简单开始

# 第一步：理解基本原理messages = []messages.append(HumanMessage(content="你好"))response = llm.invoke(messages)

2. 逐步增加功能

# 第二步：封装成类
class SimpleChat:def __init__(self, llm):self.llm = llmself.messages = []def chat(self, text):self.messages.append(HumanMessage(content=text))response = self.llm.invoke(self.messages)self.messages.append(response)return response.content

3. 添加必要的管理

• 限制历史长度（避免成本失控）
• 添加系统提示（定义 AI 角色）
• 实现会话隔离（多用户场景）

进阶：性能优化

• 使用异步处理提升并发能力
• 实现智能缓存减少重复调用
• 按 token 数量精确控制成本

生产级特性

✅ 持久化存储（数据库/文件）
✅ 完善的日志和监控
✅ 异常处理和降级策略
✅ Token 使用统计和告警
✅ 安全性（输入验证、敏感信息过滤）

架构设计原则

┌─────────────────────────────────────┐
│         应用层（业务逻辑）             │
└──────────────┬──────────────────────┘│
┌──────────────▼──────────────────────┐
│      对话管理层（消息历史）         	  │
│      • 会话隔离              		  │
│      • 历史管理           		      │
│      • 上下文压缩       		      │
└──────────────┬──────────────────────┘│
┌──────────────▼──────────────────────┐
│      存储层（持久化）            	  │
│      • 数据库                        │
│      • 缓存            		      │
└──────────────┬──────────────────────┘│
┌──────────────▼──────────────────────┐
│      模型层（LLM API）      		  │
└─────────────────────────────────────┘

---

常见问题解答

Q1：为什么不直接用 ChatGPT 的网页版，而要自己管理消息？

答： 网页版确实方便，但有局限：

• ❌ 无法集成到自己的应用中
• ❌ 无法自定义对话逻辑
• ❌ 无法批量处理或自动化
• ❌ 数据不在自己手中

使用 LangChain 的优势：

• ✅ 完全控制对话流程
• ✅ 与业务系统深度集成
• ✅ 自定义存储和分析
• ✅ 支持复杂的工作流

Q2: 历史消息应该保留多少轮？

答： 取决于具体场景：

场景类型	推荐轮数	原因
简单问答	3-5 轮	问题通常独立，不需要太多上下文
客服咨询	5-10 轮	需要记住用户信息和问题细节
深度对话	10-20 轮	话题连贯性强，需要更多历史
代码助手	5-15 轮	需要记住代码上下文和需求

通用建议：

# 根据 token 限制动态调整
if model == "gpt-3.5-turbo":  # 4K contextmax_history = 5-8
elif model == "gpt-4":  # 8K contextmax_history = 10-15
elif model == "claude-2":  # 100K contextmax_history = 50+

Q3: 如何判断是否需要总结历史？

判断标准：

需要总结的信号：
✅ 对话超过 20 轮
✅ Token 使用接近上限（如超过 3000）
✅ 早期对话信息仍然重要（如用户资料）
✅ 响应时间变慢不需要总结的场景：
❌ 短期对话（< 10 轮）
❌ 话题变化快（旧信息已不相关）
❌ 成本敏感（总结本身需要一次 LLM 调用）

实现策略：

if len(messages) > 20 and contains_important_info(messages[:10]):summary = summarize(messages[:10])messages = [summary] + messages[10:]

Q4: 多用户场景下如何避免信息泄露？

严格的会话隔离：

class SecureConversationManager:def __init__(self):self.sessions = {}def chat(self, user_id: str, message: str) -> str:# ✅ 正确：每个用户独立的会话if user_id not in self.sessions:self.sessions[user_id] = []messages = self.sessions[user_id]messages.append(HumanMessage(content=message))response = llm.invoke(messages)return response.content# ❌ 错误示范：共享会话# def chat(self, message: str):#     self.shared_messages.append(...)  # 所有用户共享！

安全检查清单：

✅ 每个用户有唯一的 session_id
✅ 消息列表按 user_id 隔离存储
✅ 敏感信息（如用户名、手机号）需要加密存储
✅ 定期清理过期会话
✅ 实现访问控制（用户只能访问自己的历史）

---

大模型未来展望

短期趋势（1-2 年）

更长的上下文窗口

当前：GPT-4 Turbo = 128K tokens
未来：1M+ tokens影响：
- 可以保留更完整的对话历史
- 减少总结的需求
- 支持更复杂的应用场景（如分析整本书）

原生多模态支持

# 未来的对话可能是这样的
messages = [HumanMessage(content="这是我的设计图", image="design.png"),AIMessage(content="我看到了，建议修改配色"),HumanMessage(content="改成这样如何？", image="design_v2.png")
]

流式对话优化

# 实时流式响应
async for chunk in llm.astream(messages):print(chunk, end="", flush=True)

中期展望（3-5 年）

内置记忆机制

# 模型 API 可能会提供原生的会话管理
response = llm.chat(user_id="user_123",message="你好",# 不需要手动传递历史！
)

个性化长期记忆

模型能够：
- 记住用户的长期偏好
- 跨会话共享知识
- 自动提取和存储关键信息

智能上下文管理

模型自动：
- 识别重要信息（无需人工标注）
- 决定何时总结、何时遗忘
- 优化 token 使用

长期愿景（5 年+）

持续学习能力

模型可以从每次交互中学习：
- 个性化回复风格
- 记住用户的纠正和反馈
- 不断优化对话质量

情感与关系记忆

不仅记住事实，还能记住：
- 用户的情绪状态
- 对话的情感基调
- 长期的关系建立

---

2025.10.02 吉林·松原