LangChain Memory 详解：让 AI 记住对话历史

AI 记不住上下文？对话像失忆？用 LangChain Memory，让 AI 拥有长期和短期记忆！

---

为什么需要 Memory？

没有 Memory 的 AI：

👤 你：我叫小明
🤖 AI: 你好小明！

👤 你：我叫什么？
🤖 AI: 抱歉，我不知道您叫什么。

有 Memory 的 AI：

👤 你：我叫小明
🤖 AI: 你好小明！

👤 你：我叫什么？
🤖 AI: 你叫小明啊 😊

---

🛠️ 5 种 Memory 详解

1. ConversationBufferMemory（完整记忆）

from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory()

memory.save_context({"input": "我叫小明"}, {"output": "你好小明！"})
memory.save_context({"input": "我 25 岁"}, {"output": "25 岁正是好年华！"})

print(memory.chat_memory.messages)
# 输出所有对话历史

优点: 完整记录所有对话
缺点: 对话长了会超出 token 限制
适用: 短对话、测试环境

---

2. ConversationBufferWindowMemory（滑动窗口）

from langchain.memory import ConversationBufferWindowMemory

# 只保留最近 5 轮对话
memory = ConversationBufferWindowMemory(k=5, return_messages=True)

# 自动丢弃最早的对话
for i in range(10):
    memory.save_context({"input": f"问题{i}"}, {"output": f"回答{i}"})

# 只保留最后 5 轮
print(memory.load_memory_variables({}))

优点: 控制 token 使用
缺点: 会丢失早期重要信息
适用: 日常对话、客服场景

---

3. ConversationSummaryMemory（摘要记忆）

from langchain.memory import ConversationSummaryMemory
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo")
memory = ConversationSummaryMemory(llm=llm, return_messages=True)

# 自动总结对话
for i in range(20):
    memory.save_context(
        {"input": f"用户问了关于产品的问题{i}"},
        {"output": f"AI 回答了产品相关问题{i}"}
    )

# 加载的是摘要，不是原文
print(memory.load_memory_variables({}))

优点: 节省 token，保留核心信息
缺点: 丢失细节
适用: 长对话、会议记录

---

4. ConversationSummaryBufferMemory（摘要 + 缓冲）

from langchain.memory import ConversationSummaryBufferMemory
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo")

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,  # 超过 2000 token 开始摘要
    return_messages=True
)

# 前 10 轮完整记录，之后自动摘要
for i in range(30):
    memory.save_context(
        {"input": f"问题{i}"},
        {"output": f"回答{i}"}
    )

优点: 兼顾细节和长度
缺点: 实现复杂
适用: 生产环境、长周期对话

---

5. VectorStoreRetrieverMemory（向量记忆）

from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embeddings)

# 创建向量记忆
memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
)

# 保存对话
memory.save_context(
    {"input": "Python 怎么读取 Excel？"},
    {"output": "用 pandas 库，pd.read_excel()"}
)

memory.save_context(
    {"input": "PDF 怎么处理？"},
    {"output": "用 PyPDF2 或 pdfplumber"}
)

# 向量检索相关记忆
result = memory.load_memory_variables({"input": "Excel 和 PDF 怎么处理？"})
print(result['history'])

优点: 智能检索相关记忆
缺点: 需要向量数据库
适用: 知识库、专业问答

---

🎯 实战：企业级对话管理系统

"""
企业级对话管理系统
功能：
1. 短短期记忆结合
2. 对话摘要压缩
3. 重要信息永久存储
4. 支持多用户
"""

from langchain.memory import (
    ConversationSummaryBufferMemory,
    VectorStoreRetrieverMemory
)
from langchain_community.vectorstores import Chroma
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.chains import ConversationChain
from langchain.prompts import PromptTemplate
import json
import os

class EnterpriseChatSystem:
    def __init__(self, user_id="default"):
        self.user_id = user_id
        self.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
        
        # 1. 短期记忆（滑动窗口）
        self.short_memory = ConversationSummaryBufferMemory(
            llm=self.llm,
            max_token_limit=2000,
            return_messages=True
        )
        
        # 2. 长期记忆（向量存储）
        embeddings = OpenAIEmbeddings()
        vectorstore = Chroma(
            persist_directory=f"./memory_db/{user_id}",
            embedding_function=embeddings
        )
        self.long_memory = VectorStoreRetrieverMemory(
            retriever=vectorstore.as_retriever(search_kwargs={"k": 3})
        )
        
        # 3. 创建对话链
        self.prompt = PromptTemplate(
            input_variables=["history", "context", "input"],
            template="""
你是一个专业的企业助手。

【长期记忆】
{context}

【对话历史】
{history}

【用户输入】
{input}

请结合记忆和上下文，用中文专业地回答：
"""
        )
        
        self.conversation = ConversationChain(
            llm=self.llm,
            memory=self.short_memory,
            prompt=self.prompt,
            verbose=False
        )
    
    def chat(self, user_input):
        """对话"""
        # 从长期记忆检索
        long_term = self.long_memory.load_memory_variables({"input": user_input})
        
        # 生成回答
        response = self.conversation.predict(
            input=user_input,
            context=long_term.get('history', '')
        )
        
        # 保存到长期记忆（重要信息）
        if self._is_important(user_input, response):
            self.long_memory.save_context(
                {"input": user_input},
                {"output": response}
            )
        
        return response
    
    def _is_important(self, input_text, output_text):
        """判断是否重要（简单规则）"""
        keywords = ['姓名', '电话', '邮箱', '地址', '账号', '密码', 'ID']
        return any(kw in input_text + output_text for kw in keywords)
    
    def clear_short_memory(self):
        """清空短期记忆"""
        self.short_memory.clear()
    
    def get_memory_summary(self):
        """获取记忆摘要"""
        return self.short_memory.load_memory_variables({})

def main():
    print("=" * 60)
    print("🤖 企业级对话管理系统")
    print("=" * 60)
    
    user_id = input("请输入用户 ID: ") or "default"
    system = EnterpriseChatSystem(user_id)
    
    print("\n💬 开始对话（输入 quit 退出，clear 清空历史）\n")
    
    while True:
        user_input = input("👤 你：")
        
        if user_input.lower() == 'quit':
            print("👋 再见！")
            break
        elif user_input.lower() == 'clear':
            system.clear_short_memory()
            print("✅ 短期记忆已清空")
            continue
        
        response = system.chat(user_input)
        print(f"🤖 AI: {response}\n")

if __name__ == '__main__':
    main()

---

⚠️ 血泪教训

1. 别用完整记忆，10 轮对话就超出 token 限制
2. 重要信息单独存，别依赖自动摘要
3. 多用户要隔离，用 user_id 分开存储
4. 定期清理记忆，不然向量库越来越大

---

💡 进阶玩法

1. 自定义摘要提示词

from langchain.memory import ConversationSummaryMemory
from langchain.prompts import PromptTemplate

custom_prompt = PromptTemplate(
    input_variables=["lines"],
    template="""
请摘要以下对话，保留关键信息（人名、时间、地点、数字）：

{lines}

摘要（100 字以内）：
"""
)

memory = ConversationSummaryMemory(
    llm=llm,
    prompt=custom_prompt
)

2. 记忆导出导入

# 导出
import json
messages = memory.chat_memory.messages
with open('memory.json', 'w', encoding='utf-8') as f:
    json.dump([{'role': m.type, 'content': m.content} for m in messages], f)

# 导入
with open('memory.json', 'r', encoding='utf-8') as f:
    data = json.load(f)
    for item in data:
        memory.chat_memory.add_user_message(item['content'])

3. 记忆分析

def analyze_memory(memory):
    """分析记忆使用情况"""
    messages = memory.chat_memory.messages
    print(f"总对话轮数：{len(messages) // 2}")
    
    # 统计 token
    total_tokens = sum(len(str(m.content)) for m in messages)
    print(f"总 Token 数：{total_tokens}")
    
    # 找出最频繁的话题
    # ...（用 NLP 分析）

🌟 下篇预告

下一篇：《LangChain Chain 实战：串联多个 AI 任务》

• SequentialChain 顺序执行
• TransformChain 数据转换
• 多步骤任务自动化
• 文章生成器实战

觉得有用？点赞 + 关注不迷路！ 🎉

---

标签: #LangChain #Memory #AI #对话系统 #Python