【LangChain】P4 LangChain 多轮对话与上下文记忆深度解析（待续）

作者注：本文所有代码均经过测试验证，可直接用于学习和开发。由于 LangChain 版本迭代较快，部分 API 可能会有变化，请以官方最新文档为准。

引言：大模型真的能记住我们的对话吗?

当我们与 ChatGPT、DeepSeek 等智能问答系统交互时，往往会产生一种错觉：这些 AI 似乎能够"记住"之前的对话内容，理解上下文，并根据历史信息给出连贯的回复。但事实上，大模型本身并不具备记忆能力。

那么，为什么我们在使用这些应用时，AI 能够基于上文进行回复呢？答案在于会话管理机制 —— 无论是基于缓存的临时存储，还是像 LangChain Memory 这样的专业框架，本质上都是通过外部系统将历史对话信息重新注入到大模型的输入中，从而营造出"记忆"的假象。

本文将通过实验验证大模型的无记忆特性，并深入探讨 LangChain 如何通过巧妙的设计实现多轮对话的上下文管理。

---

实验证明：大模型没有内在记忆

环境准备

首先，我们搭建一个基础的 LangChain 环境，连接到 OpenAI 兼容的 API：

import os
import dotenv
from langchain_openai import ChatOpenAI# 加载环境变量
dotenv.load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")
os.environ["OPENAI_BASE_URL"] = os.getenv("OPENAI_BASE_URL")# 初始化聊天模型
chat_model = ChatOpenAI(model=os.getenv("CHAT_MODEL")
)

---

实验一：单次调用无上下文

我们先定义一个系统提示（sys_message）和一个用户问题（human_message），但在实际调用时忽略这些消息，直接向模型提问：

from langchain_core.messages import SystemMessage, HumanMessagesys_message = SystemMessage(content="我是一个人工智能助手,我的名字叫小智"
)
human_message = HumanMessage(content="猫王是一只猫吗?")# 注意:这里直接传入新问题,完全忽略前面定义的消息
response = chat_model.invoke("你叫什么名字?")
print(response.content)

模型并未回复：“我叫小智”。是因为虽然我们在代码中定义了 sys_message，声明助手名叫"小智"，但由于调用 invoke() 时只传入了字符串 “你叫什么名字？”，模型根本没有看到系统消息，自然无法知道自己应该叫什么。

---

实验二：显式传入完整对话历史

现在，我们将所有消息（包括系统提示、历史问题和当前问题）打包成一个消息列表，一起传给模型：

from langchain_core.messages import SystemMessage, HumanMessagesys_message = SystemMessage(content="我是一个人工智能助手,我的名字叫小智"
)
human_message = HumanMessage(content="猫王是一只猫吗?")
human_message1 = HumanMessage(content="你叫什么名字?")# 将所有消息组成列表传入
messages = [sys_message, human_message, human_message1]
response = chat_model.invoke(messages)
print(response.content)

关键洞察：

1. 大模型是无状态的： 每次调用都是独立的，模型不会自动记住上一次交互的内容。
2. 上下文需要显式传递： 只有将历史消息作为输入的一部分传递给模型，它才能"看到"之前的对话。
3. 记忆是通过重新输入实现的： 所谓的"记忆能力"，本质上是每次调用时都把历史消息重新喂给模型。

---

理解大模型的工作原理

Transformer 的无状态特性

大语言模型基于 Transformer 架构，其核心机制是：

• 输入 → 编码 → 生成 → 输出
• 每次推理都是从零开始处理输入 tokens
• 模型参数固定，没有会话级别的动态状态更新

这意味着，对于模型而言，每一次 API 调用都是全新的。它不会"记得"5秒前、5分钟前或昨天与你的对话。

为什么 ChatGPT 看起来有记忆？

当我们在 ChatGPT 网页版中连续对话时，系统实际上在后台做了这些工作：

1. 存储对话历史： 将用户的每一条消息和 AI 的每一条回复保存在数据库或缓存中。
2. 构建上下文窗口： 每次用户发送新消息时,系统会从历史记录中提取近期对话。
3. 拼接完整提示： 将系统提示 + 历史消息 + 当前消息拼接成一个完整的输入序列。
4. 调用模型推理： 将拼接后的完整上下文发送给大模型进行推理。
5. 返回并存储回复： 将模型的回复返回给用户，并追加到历史记录中。

这种设计让用户感觉 AI "记住了"对话，但实际上每次调用时都在重新"复习"历史。

---

LangChain Memory：优雅的上下文管理方案

为什么需要 LangChain Memory？

手动管理对话历史虽然可行，但存在诸多痛点:

• 代码冗余： 每次调用都需要手动拼接消息列表
• 窗口管理复杂： 需要处理 token 限制，决定保留哪些历史消息
• 多会话支持困难： 如果有多个用户或多个对话线程，管理起来非常麻烦
• 缺乏灵活性： 不同场景可能需要不同的记忆策略（如总结记忆、实体记忆等）
• …

重要提示：LangChain 的传统 Memory 组件（如 ConversationBufferMemory）已被标记为弃用。官方推荐使用 LangGraph 或手动管理消息历史的方式。本文将介绍两种方案：

1. 传统 Memory 方案（了解原理）
2. 现代推荐方案（实际使用）

---

传统 Memory 方案

ConversationBufferMemory

最基础的记忆类型，将所有对话历史完整保存：

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain# 初始化记忆组件
memory = ConversationBufferMemory()
# 创建对话链
conversation = ConversationChain(llm=chat_model,memory=memory,verbose=True
)# 第一轮对话
response1 = conversation.predict(input="我的名字叫李明")
print(response1)

输出：Human: 我的名字叫李明
AI:> Finished chain.
你好，李明！很高兴认识你。你的名字在中文里非常常见且富有历史感，“明”有光明、明亮之意，寓意智慧与希望。今天有什么想聊的话题吗？无论是生活、科技、文学，还是想探讨某个有趣的问题，我都很乐意与你交流！比如：1. **兴趣爱好**：你平时喜欢做什么？运动、阅读还是旅行？
2. **学习或工作**：最近有在专注某个领域吗？比如学习新技能或项目？
3. **奇思妙想**：如果想探讨科幻、哲学，甚至如何种好一盆植物，我也能凑个热闹！期待你的分享～ 😊

第二轮对话：

# 第二轮对话 - 模型会"记住"用户叫李明
response2 = conversation.predict(input="你还记得我的名字吗?")
print(response2)  # 输出:你的名字是李明

输出：> Entering new ConversationChain chain...
Prompt after formatting:
The following is a friendly conversation between a human and an AI. The AI is talkative and provides lots of specific details from its context. If the AI does not know the answer to a question, it truthfully says it does not know.Current conversation:
Human: 我的名字叫李明
AI: 你好，李明！很高兴认识你。你的名字在中文里非常常见且富有历史感，“明”有光明、明亮之意，寓意智慧与希望。今天有什么想聊的话题吗？无论是生活、科技、文学，还是想探讨某个有趣的问题，我都很乐意与你交流！比如：1. **兴趣爱好**：你平时喜欢做什么？运动、阅读还是旅行？
2. **学习或工作**：最近有在专注某个领域吗？比如学习新技能或项目？
3. **奇思妙想**：如果想探讨科幻、哲学，甚至如何种好一盆植物，我也能凑个热闹！期待你的分享～ 😊
Human: 你还记得我的名字吗?
> Entering new ConversationChain chain...
Prompt after formatting:
The following is a friendly conversation between a human and an AI. The AI is talkative and provides lots of specific details from its context. If the AI does not know the answer to a question, it truthfully says it does not know.Current conversation:
Human: 我的名字叫李明
AI: 你好，李明！很高兴认识你。你的名字在中文里非常常见且富有历史感，“明”有光明、明亮之意，寓意智慧与希望。今天有什么想聊的话题吗？无论是生活、科技、文学，还是想探讨某个有趣的问题，我都很乐意与你交流！比如：1. **兴趣爱好**：你平时喜欢做什么？运动、阅读还是旅行？
2. **学习或工作**：最近有在专注某个领域吗？比如学习新技能或项目？
3. **奇思妙想**：如果想探讨科幻、哲学，甚至如何种好一盆植物，我也能凑个热闹！期待你的分享～ 😊
Human: 你还记得我的名字吗?
AI:> Finished chain.
当然记得！你的名字是**李明**，刚刚在对话开始时你告诉我的。"明"这个字寓意光明与智慧，是个很棒的名字呢。作为AI，我会认真记住当前对话中的所有信息，不过需要注意的是，如果开启新的对话会话，记忆会重新开始哦～（所以趁现在多聊聊吧！）✨

工作原理：

• memory 内部维护一个消息列表
• 每次调用 predict() 时，自动将新消息追加到历史中
• 调用模型前，自动将完整历史注入到提示中

---

ConversationBufferWindowMemory

只保留最近 N 轮对话，避免上下文过长：

from langchain.memory import ConversationBufferWindowMemory# 只保留最近 3 轮对话
memory = ConversationBufferWindowMemory(k=3)conversation = ConversationChain(llm=chat_model,memory=memory
)

适用场景：长时间对话中，早期信息不再重要，且需要控制 token 消耗。

---

ConversationSummaryMemory

将历史对话总结为摘要，压缩上下文：

from langchain.memory import ConversationSummaryMemorymemory = ConversationSummaryMemory(llm=chat_model)conversation = ConversationChain(llm=chat_model,memory=memory
)

工作原理：

• 定期调用 LLM 对历史对话进行总结
• 用总结替换原始消息，大幅减少 token 数量
• 适合长期对话或知识积累场景

结合上述示例：

# 第一轮对话
response1 = conversation.predict(input="我的名字叫李明")
print(response1)

输出：你好，李明！很高兴认识你。这是一个很常见的中文名字，听起来充满力量与阳光。今天有什么想聊的吗？无论是关于学习、科技、生活趣事，还是想讨论某个特别的话题，我都很乐意和你交流哦！比如：- 你最近在读什么书或看什么电影吗？
- 对人工智能的发展有什么好奇的地方？
- 或者需要帮忙规划行程、解决某个问题？等着你分享更多呢 😊

第二轮对话：

# 第二轮对话 - 模型会"记住"用户叫李明
response2 = conversation.predict(input="你还记得我的名字吗?")
print(response2)  # 输出:你的名字是李明

输出：当然记得！你的名字是李明，一个非常经典且好听的中文名字，寓意着“明理睿智”。很高兴继续和你聊天！有什么想聊的话题吗？无论是文化、科技，还是日常琐事，我都很乐意交流～

---

ConversationEntityMemory

注意：传统的 ConversationEntityMemory 已被弃用。现代方案推荐使用 LangGraph 或手动实现实体提取逻辑。

---

现代推荐方案：手动管理消息历史（超纲、参看）

为什么要迁移？

LangChain 官方推荐弃用传统 Memory 的原因：

• 灵活性不足： Memory 抽象层限制了自定义能力
• 透明度差： 隐藏了太多实现细节，不利于调试
• 维护成本高： 随着 LangChain 演进，维护多个 Memory 类型变得困难

新方案的优势：

• 完全控制消息历史的管理方式
• 代码更简洁、易于理解和调试
• 更容易集成到现有系统中

基础实现：手动管理消息列表

from langchain_core.messages import SystemMessage, HumanMessage, AIMessageclass ConversationManager:"""对话管理器 - 手动管理消息历史"""def __init__(self, llm, system_prompt: str = None, max_history: int = 10):self.llm = llmself.messages = []# 添加系统提示if system_prompt:self.messages.append(SystemMessage(content=system_prompt))self.max_history = max_historydef chat(self, user_input: str) -> str:"""发送消息并获取回复"""# 添加用户消息self.messages.append(HumanMessage(content=user_input))# 调用模型response = self.llm.invoke(self.messages)# 添加AI回复self.messages.append(AIMessage(content=response.content))# 管理历史长度(保留系统提示 + 最近N轮对话)self._trim_history()return response.contentdef _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)]# 只保留最近的N条对话消息if len(conversation_messages) > self.max_history * 2:  # *2 因为包含用户和AI消息conversation_messages = conversation_messages[-(self.max_history * 2):]self.messages = system_messages + conversation_messagesdef get_history(self) -> list:"""获取对话历史"""return self.messagesdef 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轮对话
)# 第一轮对话
response1 = manager.chat("你好,我叫李明")
print(f"AI: {response1}")# 第二轮对话
response2 = manager.chat("你还记得我叫什么吗?")
print(f"AI: {response2}")# 查看完整历史
print("\n=== 对话历史 ===")
for msg in manager.get_history():role = msg.__class__.__name__.replace("Message", "")print(f"{role}: {msg.content}")

输出：AI: 你好，李明！很高兴认识你！我是小智，你的AI助手。有什么我可以帮你的吗？无论是回答问题、提供建议，还是陪你聊天，我都很乐意为你服务！😊
AI: 当然记得！你刚刚告诉我你叫**李明**。很高兴能继续为你提供帮助！有什么需要我做的吗？😊=== 对话历史 ===
System: 你是一个友好的AI助手,名叫小智
Human: 你好,我叫李明
AI: 你好，李明！很高兴认识你！我是小智，你的AI助手。有什么我可以帮你的吗？无论是回答问题、提供建议，还是陪你聊天，我都很乐意为你服务！😊
Human: 你还记得我叫什么吗?
AI: 当然记得！你刚刚告诉我你叫**李明**。很高兴能继续为你提供帮助！有什么需要我做的吗？😊

进阶实现：带总结功能的对话管理器

class ConversationManagerWithSummary(ConversationManager):"""带总结功能的对话管理器"""def __init__(self, llm, system_prompt: str = None, max_history: int = 10, summary_threshold: int = 20):super().__init__(llm, system_prompt, max_history)self.summary = Noneself.summary_threshold = summary_thresholddef _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:self._generate_summary(conversation_messages[:self.summary_threshold * 2])# 保留总结 + 最近的对话conversation_messages = conversation_messages[self.summary_threshold * 2:]self.messages = system_messages + conversation_messagesdef _generate_summary(self, messages: list):"""生成对话总结"""# 格式化历史消息history_text = "\n".join([f"{'用户' if isinstance(m, HumanMessage) else 'AI'}: {m.content}"for m in messages])# 调用LLM生成总结summary_prompt = f"""请总结以下对话的关键信息:{history_text}请用简洁的语言总结对话中的重要信息、决策和上下文。"""summary_response = self.llm.invoke([HumanMessage(content=summary_prompt)])self.summary = summary_response.contentdef 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# 使用示例
advanced_manager = ConversationManagerWithSummary(llm=chat_model,system_prompt="你是一个AI助手",max_history=3,summary_threshold=5  # 超过5轮对话就生成总结
)

持久化存储方案

import json
from pathlib import Path
from datetime import datetimeclass PersistentConversationManager(ConversationManager):"""支持持久化的对话管理器"""def __init__(self, llm, session_id: str, storage_path: str = "./conversations",system_prompt: str = None, max_history: int = 10):super().__init__(llm, system_prompt, max_history)self.session_id = session_idself.storage_path = Path(storage_path)self.storage_path.mkdir(exist_ok=True)# 加载历史对话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 file_path.exists():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))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:continuemessages_data.append({'type': msg_type,'content': msg.content,'timestamp': datetime.now().isoformat()})with open(file_path, 'w', encoding='utf-8') as f:json.dump({'messages': messages_data}, f, ensure_ascii=False, indent=2)def chat(self, user_input: str) -> str:"""发送消息并自动保存"""response = super().chat(user_input)self._save_history()  # 每次对话后保存return response# 使用示例
persistent_manager = PersistentConversationManager(llm=chat_model,session_id="user_123",  # 用户唯一标识system_prompt="你是客服助手小智"
)# 即使程序重启,对话历史也会保留
response = persistent_manager.chat("你好")