【LangChain】P5 对话记忆完全指南:从原理到实战(上)
你还记得我之前说过什么吗?” - AI 能准确回答“继续上次的话题” - AI 能无缝衔接重新打开 - AI 完全忘记了你是谁大模型到底有没有记忆?如果有,为什么会突然失忆?本文将带你揭开这个谜团,并教你如何用 LangChain 构建一个"记忆力"强大的 AI 应用。无论你是刚接触 AI 的新手,还是想要优化现有对话系统的开发者,都能从中获得实用的知识和代码。概念解释类比无状态每次 API 调用
本文分为上中下三篇,内容较长。所以如果读者只是为了入门,可以看该博文的上半部分,或者可以看【LangChain】系列博文中的 P4 部分内容。
前言:大模型真的有记忆吗?
当你和 ChatGPT 或其他 AI 助手聊天时,是否有过这样的体验:
- “你还记得我之前说过什么吗?” - AI 能准确回答
- “继续上次的话题” - AI 能无缝衔接
- 重新打开 - AI 完全忘记了你是谁
这些现象让人困惑:大模型到底有没有记忆?如果有,为什么会突然失忆?
本文将带你揭开这个谜团,并教你如何用 LangChain 构建一个"记忆力"强大的 AI 应用。无论你是刚接触 AI 的新手,还是想要优化现有对话系统的开发者,都能从中获得实用的知识和代码。
第一部分:揭秘大模型的"记忆"真相
1.1 大模型的工作原理:每次都是"初次见面"
让我们先理解一个核心事实:大模型本身没有记忆。
想象一下,你去一家餐厅:
- 第一次去: 服务员问你"吃点什么?"
- 第二次去: 服务员还是问"吃点什么?"(他不记得你)
- 带着笔记本去: 你拿出笔记本说"上次我点了宫保鸡丁,今天想换个菜",服务员看了笔记本才知道你的历史
大模型就像这位服务员,每次调用都是独立的。所谓的"记忆",其实是你(或系统)把之前的对话内容(笔记本)一起传给它看。
1.2 一个简单的实验:证明大模型无记忆
让我们用代码验证这个事实:
from langchain_openai import ChatOpenAI
# 初始化模型
chat_model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 第一次对话
response1 = chat_model.invoke("你好,我叫李明,今年25岁")
print(f"第一次: {response1.content}")
# 第二次对话(没有传入历史)
response2 = chat_model.invoke("你还记得我叫什么名字吗?")
print(f"第二次: {response2.content}")
应用 DeepSeek 模型回复:
第一次: 李明你好!很高兴认识你!25岁正是充满活力和无限可能的年纪呢。请问今天有什么可以帮你的吗?无论是关于职业发展、生活建议,还是想聊聊兴趣爱好,我都很乐意与你交流~ 😊
第二次: 很抱歉,我无法记住用户的个人信息呢!😅 每次对话对我来说都是全新的开始,我没有保存之前聊天记录的能力。
不过如果你愿意的话,可以现在告诉我你的名字,我会很开心地在这段对话中称呼你!这样我们聊天的时候就会更亲切啦~你想让我怎么称呼你呢?
看到了吗?模型完全"忘记"了你的名字,因为第二次调用时,它根本不知道第一次发生了什么。
1.3 实现"记忆"的秘密:显式传递历史消息
要让模型"记住"之前的对话,我们需要手动把历史消息一起传给它:
from langchain_core.messages import HumanMessage, AIMessage
# 手动构建对话历史
messages = [
HumanMessage(content="你好,我叫李明,今年25岁"),
AIMessage(content="你好,李明!很高兴认识你。"),
HumanMessage(content="你还记得我叫什么名字吗?")
]
# 把整个历史传给模型
response = chat_model.invoke(messages)
print(f"带历史的回复: {response.content}")
带有记忆内容的回复:
带历史的回复: 当然记得!你刚才提到过你叫**李明**。名字很好听,寓意着光明与智慧,很高兴能继续和你交流!😊 如果有什么想聊的话题或需要帮助的地方,随时告诉我哦~
现在模型"记住"了!但实际上,是我们把"笔记本"(消息历史)递给它看的。
1.4 核心概念总结
理解以下几点,你就掌握了大模型对话的本质:
概念 | 解释 | 类比 |
---|---|---|
无状态 | 每次 API 调用都是独立的 | 服务员每次都忘记你 |
消息历史 | 之前所有对话的记录 | 你的笔记本 |
上下文窗口 | 模型一次能"看"多少内容 | 笔记本的页数限制 |
记忆管理 | 决定保留哪些历史消息 | 选择给服务员看笔记本的哪几页 |
第二部分:传统方案 - LangChain Memory(已弃用)
⚠️ 重要提示:LangChain 官方已不推荐使用 Memory 模块,本部分仅作为理解演进过程的参考。新项目请直接跳到第三部分学习现代方案。关于 Memory 的实践,读者可以参考本系列博文的 P4 部分内容。
2.1 为什么 LangChain 创建了 Memory?
手动管理消息历史虽然直观,但在实际应用中会遇到很多问题:
问题 1:重复代码
# 每次对话都要写这些代码
messages.append(HumanMessage(content=user_input))
response = llm.invoke(messages)
messages.append(AIMessage(content=response.content))
问题 2:历史管理复杂
- 对话太长怎么办?(超出模型上下文限制)
- 如何只保留最近几轮对话?
- 如何总结旧对话节省 token?
问题 3:多用户场景
- 如何隔离不同用户的对话?
- 如何持久化存储到数据库?
为了解决这些问题,LangChain 设计了 Memory 抽象层。
2.2 Memory 的设计思想
Memory 就像一个"智能助理",帮你自动完成以下工作:
┌─────────────────────────────────────┐
│ 你的应用代码 │
│ (只需调用 chain.invoke) │
└──────────────┬──────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ Memory 助理 │
│ • 自动添加用户消息 │
│ • 自动保存 AI 回复 │
│ • 自动管理历史长度 │
│ • 自动格式化上下文 │
└──────────────┬──────────────────────┘
│
▼
┌─────────────────────────────────────┐
│ 大语言模型 │
└─────────────────────────────────────┘
2.3 为什么 LangChain 弃用了 Memory?
尽管 Memory 提供了便利,但 LangChain 团队发现了以下问题:
- 灵活性不足
# 使用 Memory 时,很多细节被隐藏了 chain = ConversationChain(memory=memory) chain.invoke("你好") # 发生了什么?不清楚!
- 透明度差
- Memory 内部做了很多"自动化"操作
- 出问题时难以调试
- 行为不符合预期时难以定位原因
- 维护成本高
- LangChain 需要维护多种 Memory 类型
- 每次模型 API 更新都要适配
- 社区反馈说"太复杂了"
官方的新理念:
与其提供一个"黑盒",不如让开发者直接管理消息历史。代码虽然多了几行,但更清晰、可控、易于理解。
第三部分:现代方案 - 手动管理消息历史
3.1 新方案的核心优势
现代推荐方案的理念是:显式优于隐式,简单优于复杂。
对比维度 | 传统 Memory | 现代方案 |
---|---|---|
代码透明度 | 低(隐藏细节) | 高(一目了然) |
灵活性 | 受限于 Memory 类型 | 完全自定义 |
调试难度 | 困难 | 容易 |
学习曲线 | 需要理解 Memory 抽象 | 直接操作消息列表 |
维护成本 | 依赖 LangChain 更新 | 自己掌控 |
3.2 基础实现:构建对话管理器
让我们从零开始,构建一个简单但功能完整的对话管理器:
对话管理器方法
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
class ConversationManager:
"""对话管理器 - 现代推荐方案"""
def __init__(self, llm, system_prompt: str = None, max_history: int = 10):
"""
初始化对话管理器
参数说明:
- llm: 语言模型实例
- system_prompt: 系统提示词(定义 AI 的角色和行为)
- max_history: 最多保留几轮对话(一轮 = 用户消息 + AI回复)
"""
self.llm = llm
self.messages = [] # 存储所有消息的列表
# 如果有系统提示,添加到消息列表的开头
if system_prompt:
self.messages.append(SystemMessage(content=system_prompt))
self.max_history = max_history
def chat(self, user_input: str) -> str:
"""
发送消息并获取回复
这是用户的主要接口,就像和 AI 聊天一样简单
"""
# 步骤 1: 添加用户消息到历史
self.messages.append(HumanMessage(content=user_input))
# 步骤 2: 把整个历史传给模型,获取回复
response = self.llm.invoke(self.messages)
# 步骤 3: 把 AI 的回复也加入历史
self.messages.append(AIMessage(content=response.content))
# 步骤 4: 检查历史是否过长,需要清理
self._trim_history()
return response.content
def _trim_history(self):
"""
修剪历史消息,避免超出模型的上下文限制
策略:保留系统提示 + 最近 N 轮对话
"""
# 分离系统消息和对话消息
system_messages = [m for m in self.messages if isinstance(m, SystemMessage)]
conversation_messages = [m for m in self.messages if not isinstance(m, SystemMessage)]
# 如果对话消息太多,只保留最近的
# 注意:*2 是因为一轮对话包含用户消息和 AI 回复
if len(conversation_messages) > self.max_history * 2:
conversation_messages = conversation_messages[-(self.max_history * 2):]
# 重新组合:系统消息 + 保留的对话
self.messages = system_messages + conversation_messages
def get_history(self) -> list:
"""获取完整的对话历史"""
return self.messages
def clear_history(self):
"""清空对话历史(但保留系统提示)"""
system_messages = [m for m in self.messages if isinstance(m, SystemMessage)]
self.messages = system_messages
添加实例测试
# 创建对话管理器
manager = ConversationManager(
llm=chat_model,
system_prompt="你是一个友好的 AI 助手,名叫小智。你擅长回答问题并记住用户信息。",
max_history=5 # 只保留最近 5 轮对话
)
# 开始对话!
print("=== 第一轮 ===")
response1 = manager.chat("你好,我叫李明,是一名程序员")
print(f"小智: {response1}")
print("\n=== 第二轮 ===")
response2 = manager.chat("你还记得我的名字和职业吗?")
print(f"小智: {response2}")
print("\n=== 第三轮 ===")
response3 = manager.chat("帮我推荐一本适合程序员的书")
print(f"小智: {response3}")
# 查看完整历史
print("\n=== 对话历史 ===")
for i, msg in enumerate(manager.get_history()):
role = msg.__class__.__name__.replace("Message", "")
print(f"{i+1}. {role}: {msg.content[:50]}...")
运行效果
=== 第一轮 ===
小智: 你好李明!很高兴认识你这位程序员朋友!有什么可以帮助你的吗?
=== 第二轮 ===
小智: 当然记得!你叫李明,是一名程序员。有什么编程问题需要帮助吗?
=== 第三轮 ===
小智: 我推荐《代码大全》,这是程序员必读的经典书籍...
=== 对话历史 ===
1. System: 你是一个友好的 AI 助手,名叫小智...
2. Human: 你好,我叫李明,是一名程序员
3. AI: 你好李明!很高兴认识你这位程序员朋友...
4. Human: 你还记得我的名字和职业吗?
5. AI: 当然记得!你叫李明,是一名程序员...
6. Human: 帮我推荐一本适合程序员的书
7. AI: 我推荐《代码大全》...
功能模块详解
让我们深入理解每个功能模块的设计:
模块 1:消息类型
LangChain 定义了三种基本消息类型:
# 1. SystemMessage - 系统提示
# 用途:定义 AI 的角色、行为规则、回复风格
system_msg = SystemMessage(content="你是一个专业的医生")
# 2. HumanMessage - 用户消息
# 用途:用户的输入
user_msg = HumanMessage(content="我头疼怎么办?")
# 3. AIMessage - AI 回复
# 用途:模型的输出
ai_msg = AIMessage(content="建议您多休息,如果持续疼痛请就医")
为什么要区分消息类型?
- 模型需要知道谁在说话(用户 vs AI)
- 系统提示有特殊地位(通常放在最前面且不会被删除)
- 便于格式化显示和数据分析
模块 2:历史管理策略
def _trim_history(self):
"""
历史管理的核心逻辑
问题:为什么要限制历史长度?
- 模型有上下文窗口限制(如 GPT-3.5 是 4096 tokens)
- Token 越多,调用成本越高
- 历史太长可能引入噪音,影响回答质量
策略:
1. 永远保留系统提示(定义 AI 的身份)
2. 保留最近 N 轮对话(最相关的上下文)
3. 丢弃更早的对话(假设不再相关)
"""
system_messages = [m for m in self.messages if isinstance(m, SystemMessage)]
conversation_messages = [m for m in self.messages if not isinstance(m, SystemMessage)]
# 一轮对话 = 用户消息 + AI 回复,所以要 * 2
if len(conversation_messages) > self.max_history * 2:
conversation_messages = conversation_messages[-(self.max_history * 2):]
self.messages = system_messages + conversation_messages
图解历史管理:
对话历史增长过程:
第 1 轮:[System] [Human] [AI]
第 2 轮:[System] [Human] [AI] [Human] [AI]
第 3 轮:[System] [Human] [AI] [Human] [AI] [Human] [AI]
...
第 10 轮:[System] [H] [A] ... [H] [A] ← 达到 max_history 限制
第 11 轮:[System] [A] [H] [A] ... [H] [A] ← 删除最早的 [H]
↑ 保留最近 10 轮
3.3 进阶:带总结功能的对话管理器
当对话变得很长时,简单删除旧消息可能会丢失重要信息。更好的方法是总结旧对话:
class ConversationManagerWithSummary(ConversationManager):
"""带智能总结功能的对话管理器"""
def __init__(self, llm, system_prompt: str = None,
max_history: int = 10, summary_threshold: int = 20):
"""
新增参数:
- summary_threshold: 当对话超过多少轮时触发总结
"""
super().__init__(llm, system_prompt, max_history)
self.summary = None # 存储对话总结
self.summary_threshold = summary_threshold
def _trim_history(self):
"""
升级版历史管理:先总结,再删除
"""
system_messages = [m for m in self.messages if isinstance(m, SystemMessage)]
conversation_messages = [m for m in self.messages if not isinstance(m, SystemMessage)]
# 如果对话超过阈值,生成总结
if len(conversation_messages) > self.summary_threshold * 2:
# 把要删除的消息先总结一下
old_messages = conversation_messages[:self.summary_threshold * 2]
self._generate_summary(old_messages)
# 只保留最近的对话
conversation_messages = conversation_messages[self.summary_threshold * 2:]
self.messages = system_messages + conversation_messages
def _generate_summary(self, messages: list):
"""
调用 LLM 生成对话总结
总结的好处:
- 保留关键信息(如用户姓名、重要决策)
- 大幅减少 token 消耗
- 提供长期上下文
"""
# 格式化要总结的消息
history_text = "\n".join([
f"{'用户' if isinstance(m, HumanMessage) else 'AI'}: {m.content}"
for m in messages
])
# 构建总结提示
summary_prompt = f"""请总结以下对话的关键信息:
{history_text}
总结要点:
1. 用户的基本信息(姓名、需求等)
2. 讨论的主要话题
3. 达成的结论或决策
4. 其他重要细节
请用简洁的语言总结(不超过 200 字):"""
# 调用 LLM 生成总结
summary_response = self.llm.invoke([HumanMessage(content=summary_prompt)])
self.summary = summary_response.content
print(f"\n[系统] 已生成对话总结:{self.summary}\n")
def chat(self, user_input: str) -> str:
"""
聊天时,如果有总结,会自动加入上下文
"""
# 如果有总结,临时添加到消息开头(系统提示之后)
if self.summary:
summary_msg = SystemMessage(content=f"【之前对话的总结】\n{self.summary}")
self.messages.insert(1, summary_msg)
# 正常聊天流程
response = super().chat(user_input)
# 移除临时添加的总结消息(避免重复累积)
if self.summary:
self.messages = [m for m in self.messages
if not (isinstance(m, SystemMessage) and "之前对话的总结" in m.content)]
return response
总结功能的工作流程
对话进行中...
├── 第 1-20 轮:正常对话,全部保留
├── 第 21 轮:触发总结!
│ ├── 总结前 20 轮的关键信息 → 存为 summary
│ ├── 删除前 20 轮的原始消息
│ └── 保留 summary + 最近 10 轮
├── 第 22-40 轮:携带 summary 继续对话
└── 第 41 轮:再次触发总结...
上述部分为:对话记忆完全指南:从原理到实战(上)部分内容,下部内容(中)将对一个实例展开。(下)部分内容将并进一步讨论进阶处理方法,如持久化存储等。
请访问 【LangChain】系列博文,P6 文章。
2025.10.01 祝祖国母亲繁荣昌盛,我的家人一切顺利!
中国·吉林长春
更多推荐
所有评论(0)