05_记忆管理 - 让AI记住对话历史

发表于 2025-06-18 更新于 2025-10-28 分类于 rd ， python Changyan：本文字数： 6.5k 阅读时长 ≈ 24 分钟

本课程解决AI”健忘”的问题，教你使用LangChain的记忆系统让AI记住对话历史。你将学习四种记忆类型（缓冲、窗口、摘要、组合）的特点和应用场景，掌握如何在多用户场景下管理独立记忆，理解记忆与Token消耗的平衡。通过实战练习，构建具有连贯对话能力的聊天机器人。

🎯 学习目标

理解为什么AI需要记忆
掌握不同类型的记忆机制
学会在应用中集成记忆功能
构建有记忆的聊天机器人

📖 核心概念

为什么需要记忆？

问题：AI没有记忆

# 第一次对话
user: "我叫张三"
ai: "你好张三！"

# 第二次对话
user: "我叫什么名字？"
ai: "我不知道你的名字" ❌

# 问题：AI不记得之前说过什么

生活类比：失忆的朋友

想象你和一个朋友聊天，但他每次都忘记之前说的话：

你："我最喜欢吃pizza"
朋友："好的，我记住了"

5分钟后...
你："我喜欢吃什么？"
朋友："我不知道" 😅

这样的对话体验很糟糕！

AI的记忆系统

LangChain提供多种记忆类型：

1. 对话缓冲记忆 (ConversationBufferMemory)
   - 完整记录所有对话
   - 简单直接
   - 适合短对话

2. 对话窗口记忆 (ConversationBufferWindowMemory)
   - 只记录最近N轮对话
   - 节省tokens
   - 适合长对话

3. 对话摘要记忆 (ConversationSummaryMemory)
   - 总结历史对话
   - 节省更多tokens
   - 保留关键信息

4. 对话摘要缓冲记忆 (ConversationSummaryBufferMemory)
   - 结合缓冲和摘要
   - 最灵活
   - 适合生产环境

记忆的工作原理

用户输入
   ↓
[记忆] ← 读取历史对话
   ↓
提示词模板 + 历史 + 新输入
   ↓
LLM处理
   ↓
生成回复
   ↓
[记忆] ← 保存本轮对话
   ↓
返回结果

💻 代码示例

参见各个示例文件：

01_buffer_memory.py - 对话缓冲记忆

"""
对话缓冲记忆
演示：完整记录所有对话历史
"""

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

load_dotenv()


def example1_basic_memory() -> None:
    """示例1：基础记忆使用"""
    print("=== 示例1：基础对话记忆 ===\n")

    # 创建记忆
    memory = ConversationBufferMemory()

    # 手动保存对话
    memory.save_context(
        {"input": "我叫张三"},
        {"output": "你好张三！很高兴认识你。"}
    )

    memory.save_context(
        {"input": "我今年25岁"},
        {"output": "知道了，你今年25岁。"}
    )

    # 查看记忆
    print("💭 记忆内容：")
    print(memory.load_memory_variables({}))
    print()


def example2_with_llm() -> None:
    """示例2：配合LLM使用记忆"""
    print("=== 示例2：带记忆的对话 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    # 创建记忆
    memory = ConversationBufferMemory()

    # 创建对话链
    conversation = ConversationChain(
        llm=llm,
        memory=memory,
        verbose=True  # 显示详细过程
    )

    # 多轮对话
    print("👤 用户：我叫李明")
    response1 = conversation.predict(input="我叫李明")
    print(f"🤖 AI：{response1}\n")

    print("👤 用户：我喜欢编程")
    response2 = conversation.predict(input="我喜欢编程")
    print(f"🤖 AI：{response2}\n")

    print("👤 用户：我叫什么名字？我有什么爱好？")
    response3 = conversation.predict(input="我叫什么名字？我有什么爱好？")
    print(f"🤖 AI：{response3}\n")

    # 查看完整历史
    print("📜 完整对话历史：")
    print(memory.load_memory_variables({}))


def example3_memory_inspection() -> None:
    """示例3：检查记忆内容"""
    print("\n=== 示例3：检查记忆 ===\n")

    memory = ConversationBufferMemory()

    # 添加几轮对话
    conversations = [
        ("Python是什么？", "Python是一种编程语言"),
        ("它有什么特点？", "Python简洁易读，功能强大"),
        ("适合初学者吗？", "非常适合，语法简单")
    ]

    for user_msg, ai_msg in conversations:
        memory.save_context({"input": user_msg}, {"output": ai_msg})

    # 查看记忆统计
    history = memory.load_memory_variables({})
    print(f"总对话轮数：{len(conversations)}")
    print(f"\n历史记录：\n{history['history']}\n")

    # 查看消息对象
    print("消息列表：")
    for msg in memory.chat_memory.messages:
        print(f"  {msg.__class__.__name__}: {msg.content}")


def example4_custom_keys() -> None:
    """示例4：自定义记忆键"""
    print("\n=== 示例4：自定义键名 ===\n")

    # 使用自定义键名
    memory = ConversationBufferMemory(
        memory_key="chat_history",     # 历史记录的键名
        input_key="user_input",        # 用户输入的键名
        output_key="ai_response",      # AI输出的键名
        return_messages=True           # 返回消息对象而不是字符串
    )

    memory.save_context(
        {"user_input": "你好"},
        {"ai_response": "你好！有什么可以帮你的？"}
    )

    # 查看
    variables = memory.load_memory_variables({})
    print(f"键名：{list(variables.keys())}")
    print(f"消息类型：{type(variables['chat_history'][0])}")
    print(f"内容：{variables}")


def main() -> None:
    """主函数"""
    example1_basic_memory()
    print("\n" + "="*60 + "\n")

    example2_with_llm()
    print("\n" + "="*60 + "\n")

    example3_memory_inspection()
    print("\n" + "="*60 + "\n")

    example4_custom_keys()


if __name__ == "__main__":
    main()

02_window_memory.py - 对话窗口记忆

"""
对话窗口记忆
演示：只保留最近N轮对话，控制Token消耗
"""

from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

load_dotenv()


def example1_basic_window_memory() -> None:
    """示例1：基础窗口记忆"""
    print("=== 示例1：窗口记忆基础用法 ===\n")

    # 创建窗口记忆，只保留最近3轮对话
    memory = ConversationBufferWindowMemory(k=3)

    # 手动添加对话
    conversations = [
        ("我叫张三", "你好张三！"),
        ("我今年25岁", "知道了，你今年25岁"),
        ("我是程序员", "明白，你是程序员"),
        ("我喜欢Python", "很好，Python是很棒的语言"),
        ("我住在北京", "了解，你住在北京"),
    ]

    for i, (user_msg, ai_msg) in enumerate(conversations, 1):
        memory.save_context(
            {"input": user_msg},
            {"output": ai_msg}
        )

        print(f"第{i}轮对话后的记忆：")
        history = memory.load_memory_variables({})
        print(history["history"])
        print("\n" + "-"*50 + "\n")


def example2_window_vs_buffer() -> None:
    """示例2：窗口记忆 vs 缓冲记忆对比"""
    print("=== 示例2：窗口记忆 vs 缓冲记忆 ===\n")

    from langchain.memory import ConversationBufferMemory

    # 创建两种记忆
    buffer_memory = ConversationBufferMemory()
    window_memory = ConversationBufferWindowMemory(k=2)

    # 添加5轮对话
    conversations = [
        ("第1个问题", "第1个回答"),
        ("第2个问题", "第2个回答"),
        ("第3个问题", "第3个回答"),
        ("第4个问题", "第4个回答"),
        ("第5个问题", "第5个回答"),
    ]

    for user_msg, ai_msg in conversations:
        buffer_memory.save_context(
            {"input": user_msg},
            {"output": ai_msg}
        )
        window_memory.save_context(
            {"input": user_msg},
            {"output": ai_msg}
        )

    # 对比结果
    print("📦 缓冲记忆（保留全部）：")
    buffer_history = buffer_memory.load_memory_variables({})
    print(buffer_history["history"])
    print()

    print("🪟 窗口记忆（只保留最近2轮）：")
    window_history = window_memory.load_memory_variables({})
    print(window_history["history"])
    print()

    # 统计消息数量
    buffer_count = len(buffer_memory.chat_memory.messages)
    window_count = len(window_memory.chat_memory.messages)

    print(f"📊 统计：")
    print(f"  缓冲记忆：{buffer_count}条消息")
    print(f"  窗口记忆：{window_count}条消息")
    print(f"  节省：{buffer_count - window_count}条消息")


def example3_with_chatbot() -> None:
    """示例3：窗口记忆配合聊天机器人"""
    print("\n=== 示例3：带窗口记忆的聊天机器人 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    # 创建窗口记忆，只保留最近3轮
    memory = ConversationBufferWindowMemory(k=3)

    # 创建对话链
    conversation = ConversationChain(
        llm=llm,
        memory=memory,
        verbose=False
    )

    # 模拟多轮对话
    questions = [
        "你好，我叫李明",
        "我今年30岁",
        "我是软件工程师",
        "我喜欢打篮球",
        "我住在上海",
        "我叫什么名字？",  # 测试：超出窗口范围
        "我多大了？",      # 测试：在窗口范围内
        "我喜欢什么运动？",  # 测试：在窗口范围内
    ]

    for i, question in enumerate(questions, 1):
        print(f"第{i}轮对话")
        print(f"👤 用户：{question}")

        response = conversation.predict(input=question)
        print(f"🤖 AI：{response}")

        # 显示当前记忆状态
        current_memory = memory.load_memory_variables({})
        message_count = len(memory.chat_memory.messages)
        print(f"📊 当前记忆：{message_count//2}轮对话")
        print()


def example4_dynamic_window_size() -> None:
    """示例4：动态调整窗口大小"""
    print("\n=== 示例4：动态窗口大小 ===\n")

    llm = ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )

    class DynamicWindowMemory:
        """动态窗口记忆管理器"""

        def __init__(self, min_k: int = 2, max_k: int = 10):
            self.min_k = min_k
            self.max_k = max_k
            self.current_k = min_k
            self.memory = ConversationBufferWindowMemory(k=self.current_k)

        def adjust_window_size(self, message_importance: str) -> None:
            """根据消息重要性调整窗口大小"""
            if message_importance == "high" and self.current_k < self.max_k:
                self.current_k += 1
                print(f"📈 窗口扩大到 {self.current_k} 轮")
            elif message_importance == "low" and self.current_k > self.min_k:
                self.current_k -= 1
                print(f"📉 窗口缩小到 {self.current_k} 轮")

            # 重新创建记忆（保留现有消息）
            old_messages = self.memory.chat_memory.messages
            self.memory = ConversationBufferWindowMemory(k=self.current_k)
            for msg in old_messages:
                self.memory.chat_memory.add_message(msg)

        def save_context(self, inputs: dict, outputs: dict, importance: str = "normal") -> None:
            """保存对话并调整窗口"""
            self.memory.save_context(inputs, outputs)
            self.adjust_window_size(importance)

    # 测试
    dynamic_memory = DynamicWindowMemory(min_k=2, max_k=5)

    conversations = [
        ("你好", "你好！", "normal"),
        ("今天天气怎么样", "今天天气不错", "low"),
        ("帮我分析一下这份重要合同", "好的，我来分析...", "high"),
        ("合同第三条款有什么风险", "第三条款...", "high"),
        ("谢谢", "不客气", "low"),
    ]

    for i, (user_msg, ai_msg, importance) in enumerate(conversations, 1):
        print(f"\n第{i}轮：{user_msg} (重要性: {importance})")
        dynamic_memory.save_context(
            {"input": user_msg},
            {"output": ai_msg},
            importance
        )


def example5_multi_user_window() -> None:
    """示例5：多用户窗口记忆管理"""
    print("\n\n=== 示例5：多用户窗口记忆管理 ===\n")

    from typing import Dict

    class MultiUserWindowMemory:
        """多用户窗口记忆管理器"""

        def __init__(self, default_k: int = 3):
            self.default_k = default_k
            self.memories: Dict[str, ConversationBufferWindowMemory] = {}

        def get_memory(self, user_id: str, k: int = None) -> ConversationBufferWindowMemory:
            """获取用户的记忆"""
            if user_id not in self.memories:
                window_size = k if k else self.default_k
                self.memories[user_id] = ConversationBufferWindowMemory(
                    k=window_size)
                print(f"✨ 为用户 {user_id} 创建新记忆（窗口大小：{window_size}）")
            return self.memories[user_id]

        def save_conversation(self, user_id: str, user_msg: str, ai_msg: str) -> None:
            """保存对话"""
            memory = self.get_memory(user_id)
            memory.save_context(
                {"input": user_msg},
                {"output": ai_msg}
            )

        def get_history(self, user_id: str) -> str:
            """获取对话历史"""
            if user_id not in self.memories:
                return "（无历史记录）"
            memory = self.memories[user_id]
            history = memory.load_memory_variables({})
            return history["history"]

        def get_stats(self) -> dict:
            """获取统计信息"""
            return {
                "total_users": len(self.memories),
                "users": list(self.memories.keys())
            }

    # 测试多用户场景
    manager = MultiUserWindowMemory(default_k=2)

    # 用户A的对话
    print("👤 用户A的对话：")
    manager.save_conversation("user_A", "我叫Alice", "你好Alice！")
    manager.save_conversation("user_A", "我是学生", "知道了")
    manager.save_conversation("user_A", "我学计算机", "很好的专业")

    print(f"用户A的历史：\n{manager.get_history('user_A')}\n")

    # 用户B的对话
    print("👤 用户B的对话：")
    manager.save_conversation("user_B", "我是Bob", "你好Bob！")
    manager.save_conversation("user_B", "我是工程师", "很高兴认识你")

    print(f"用户B的历史：\n{manager.get_history('user_B')}\n")

    # 统计信息
    stats = manager.get_stats()
    print(f"📊 系统统计：")
    print(f"  总用户数：{stats['total_users']}")
    print(f"  用户列表：{', '.join(stats['users'])}")


def example6_window_with_important_messages() -> None:
    """示例6：保留重要消息的窗口记忆"""
    print("\n\n=== 示例6：智能窗口记忆（保留重要信息）===\n")

    class SmartWindowMemory:
        """智能窗口记忆：自动保留重要信息"""

        def __init__(self, k: int = 3):
            self.k = k
            self.memory = ConversationBufferWindowMemory(k=k)
            self.important_messages = []  # 保留重要信息

        def save_context(self, inputs: dict, outputs: dict, is_important: bool = False) -> None:
            """保存对话"""
            # 保存到常规记忆
            self.memory.save_context(inputs, outputs)

            # 如果是重要信息，单独保存
            if is_important:
                self.important_messages.append({
                    "user": inputs["input"],
                    "ai": outputs["output"]
                })
                print(f"⭐ 标记为重要信息")

        def get_context(self) -> str:
            """获取完整上下文（包括重要信息）"""
            # 获取窗口记忆
            recent = self.memory.load_memory_variables({})["history"]

            # 添加重要信息
            if self.important_messages:
                important_context = "\n重要信息回顾：\n"
                for msg in self.important_messages[-3:]:  # 最多保留3条
                    important_context += f"- {msg['user']}: {msg['ai']}\n"
                return important_context + "\n最近对话：\n" + recent

            return recent

    # 测试
    smart_memory = SmartWindowMemory(k=2)

    print("对话序列：\n")

    # 对话1：重要 - 用户信息
    print("1. 👤 我叫张伟，是产品经理")
    smart_memory.save_context(
        {"input": "我叫张伟，是产品经理"},
        {"output": "你好张伟，很高兴认识你"},
        is_important=True
    )

    # 对话2：普通
    print("2. 👤 今天天气不错")
    smart_memory.save_context(
        {"input": "今天天气不错"},
        {"output": "是的，很好的天气"}
    )

    # 对话3：普通
    print("3. 👤 你喜欢什么颜色")
    smart_memory.save_context(
        {"input": "你喜欢什么颜色"},
        {"output": "我喜欢蓝色"}
    )

    # 对话4：重要 - 业务信息
    print("4. 👤 我们的产品订单号是ORD12345")
    smart_memory.save_context(
        {"input": "我们的产品订单号是ORD12345"},
        {"output": "好的，我记下了订单号"},
        is_important=True
    )

    # 对话5：普通
    print("5. 👤 现在几点了")
    smart_memory.save_context(
        {"input": "现在几点了"},
        {"output": "现在是下午3点"}
    )

    print("\n" + "="*60)
    print("\n📝 当前上下文（包含重要信息）：")
    print(smart_memory.get_context())


def main() -> None:
    """主函数"""
    example1_basic_window_memory()
    print("\n" + "="*70 + "\n")

    example2_window_vs_buffer()
    print("\n" + "="*70 + "\n")

    example3_with_chatbot()
    print("\n" + "="*70 + "\n")

    example4_dynamic_window_size()
    print("\n" + "="*70 + "\n")

    example5_multi_user_window()
    print("\n" + "="*70 + "\n")

    example6_window_with_important_messages()


if __name__ == "__main__":
    main()

03_summary_memory.py - 对话摘要记忆

"""
对话摘要记忆
演示：使用AI总结对话历史，节省Token
"""

from langchain.memory import ConversationSummaryMemory, ConversationSummaryBufferMemory
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os

load_dotenv()


def create_llm() -> ChatOpenAI:
    """创建LLM实例"""
    return ChatOpenAI(
        base_url=os.getenv("ALIBABA_BASE_URL"),
        api_key=os.getenv("ALIBABA_API_KEY"),
        model="qwen-plus",
        temperature=0.7,
    )


def example1_basic_summary_memory() -> None:
    """示例1：基础摘要记忆"""
    print("=== 示例1：摘要记忆基础 ===\n")

    llm = create_llm()

    # 创建摘要记忆
    memory = ConversationSummaryMemory(llm=llm)

    # 添加多轮对话
    conversations = [
        ("我叫张三", "你好张三！"),
        ("我今年28岁", "知道了，你28岁"),
        ("我在一家科技公司工作", "很好的工作"),
        ("我负责产品设计", "产品设计是很重要的岗位"),
        ("我喜欢旅行和摄影", "很棒的爱好"),
    ]

    print("添加对话中...\n")
    for user_msg, ai_msg in conversations:
        print(f"👤 {user_msg}")
        print(f"🤖 {ai_msg}")
        memory.save_context(
            {"input": user_msg},
            {"output": ai_msg}
        )

    print("\n" + "="*60)
    print("\n📝 对话摘要：")
    summary = memory.load_memory_variables({})
    print(summary["history"])
    print()


def example2_summary_vs_buffer() -> None:
    """示例2：摘要记忆 vs 缓冲记忆对比"""
    print("\n=== 示例2：摘要记忆 vs 缓冲记忆 ===\n")

    from langchain.memory import ConversationBufferMemory

    llm = create_llm()

    # 创建两种记忆
    buffer_memory = ConversationBufferMemory()
    summary_memory = ConversationSummaryMemory(llm=llm)

    # 添加长对话
    conversations = [
        ("早上好！", "早上好！"),
        ("我想咨询一下你们的产品", "当然，请问有什么可以帮您的？"),
        ("我对你们的旗舰产品感兴趣", "太好了，我为您介绍一下"),
        ("这个产品的主要功能是什么？", "主要功能包括A、B、C三个方面"),
        ("价格是多少？", "标准版价格是999元"),
        ("有优惠活动吗？", "目前有新用户8折优惠"),
        ("那我要购买标准版", "好的，我帮您下单"),
        ("什么时候能发货？", "我们会在24小时内发货"),
    ]

    for user_msg, ai_msg in conversations:
        buffer_memory.save_context(
            {"input": user_msg},
            {"output": ai_msg}
        )
        summary_memory.save_context(
            {"input": user_msg},
            {"output": ai_msg}
        )

    print("📦 缓冲记忆（完整对话）：")
    buffer_history = buffer_memory.load_memory_variables({})
    print(buffer_history["history"])
    buffer_length = len(buffer_history["history"])
    print(f"\n长度：{buffer_length}个字符\n")

    print("="*60 + "\n")

    print("📄 摘要记忆（总结版）：")
    summary_history = summary_memory.load_memory_variables({})
    print(summary_history["history"])
    summary_length = len(summary_history["history"])
    print(f"\n长度：{summary_length}个字符")
    print(
        f"节省：{buffer_length - summary_length}个字符 ({(1-summary_length/buffer_length)*100:.1f}%)")


def example3_summary_buffer_memory() -> None:
    """示例3：摘要缓冲记忆（最佳实践）"""
    print("\n\n=== 示例3：摘要缓冲记忆 ===\n")

    llm = create_llm()

    # 创建摘要缓冲记忆
    # 保留最近的消息作为缓冲，旧消息自动总结
    memory = ConversationSummaryBufferMemory(
        llm=llm,
        max_token_limit=100  # 超过100个token就开始总结
    )

    # 创建对话链
    conversation = ConversationChain(
        llm=llm,
        memory=memory,
        verbose=False
    )

    # 长对话
    questions = [
        "你好，我想了解一下编程",
        "Python适合初学者吗？",
        "Python有哪些主要应用领域？",
        "我应该先学习哪些基础知识？",
        "有什么好的学习资源推荐吗？",
        "学习Python大概需要多长时间？",
        "刚才你说Python适合初学者，能详细说说为什么吗？",  # 测试：能否记住早期内容
    ]

    for i, question in enumerate(questions, 1):
        print(f"\n第{i}轮对话")
        print(f"👤 用户：{question}")

        response = conversation.predict(input=question)
        print(f"🤖 AI：{response[:100]}...")

        # 显示记忆状态
        print(f"\n📊 当前记忆状态：")
        memory_vars = memory.load_memory_variables({})
        print(f"  记忆内容：{memory_vars['history'][:150]}...")


def example4_custom_summary_prompt() -> None:
    """示例4：自定义摘要提示词"""
    print("\n\n=== 示例4：自定义摘要风格 ===\n")

    from langchain.prompts import PromptTemplate

    llm = create_llm()

    # 自定义摘要提示词
    summary_prompt = PromptTemplate(
        input_variables=["summary", "new_lines"],
        template="""
请用简洁的要点形式总结对话。

已有摘要：
{summary}

新对话内容：
{new_lines}

更新后的摘要（使用要点形式）：
"""
    )

    # 创建带自定义提示词的摘要记忆
    memory = ConversationSummaryMemory(
        llm=llm,
        prompt=summary_prompt
    )

    # 添加对话
    conversations = [
        ("我是一名产品经理", "你好！"),
        ("我负责一款教育类APP", "听起来很有意义"),
        ("我们的用户主要是中小学生", "明白了"),
        ("最近在规划新功能", "有什么想法吗？"),
        ("想增加AI辅导功能", "很好的想法"),
    ]

    for user_msg, ai_msg in conversations:
        memory.save_context(
            {"input": user_msg},
            {"output": ai_msg}
        )

    print("📝 自定义风格的摘要：")
    summary = memory.load_memory_variables({})
    print(summary["history"])


def example5_incremental_summary() -> None:
    """示例5：增量摘要（观察摘要如何演化）"""
    print("\n\n=== 示例5：观察摘要的演化过程 ===\n")

    llm = create_llm()
    memory = ConversationSummaryMemory(llm=llm)

    conversations = [
        ("我是李华，来自北京", "你好李华！"),
        ("我在一家互联网公司工作", "很好"),
        ("我的职位是数据分析师", "了解了"),
        ("我主要负责用户行为分析", "这是重要的工作"),
        ("我们团队有10个人", "不小的团队"),
        ("我们最近在做一个新项目", "什么项目？"),
        ("是关于推荐系统优化的", "听起来很有挑战"),
        ("我们使用机器学习算法", "很前沿的技术"),
    ]

    for i, (user_msg, ai_msg) in enumerate(conversations, 1):
        print(f"\n第{i}轮对话：")
        print(f"👤 {user_msg}")
        print(f"🤖 {ai_msg}")

        memory.save_context(
            {"input": user_msg},
            {"output": ai_msg}
        )

        # 显示当前摘要
        if i % 3 == 0:  # 每3轮显示一次摘要
            summary = memory.load_memory_variables({})
            print(f"\n📄 摘要（第{i}轮后）：")
            print(summary["history"])
            print("="*60)


def example6_memory_management() -> None:
    """示例6：记忆管理策略"""
    print("\n\n=== 示例6：智能记忆管理 ===\n")

    llm = create_llm()

    class SmartMemoryManager:
        """智能记忆管理器"""

        def __init__(self, llm):
            self.llm = llm
            self.summary_memory = ConversationSummaryMemory(llm=llm)
            self.message_count = 0
            self.summary_interval = 5  # 每5轮总结一次

        def add_conversation(self, user_msg: str, ai_msg: str) -> None:
            """添加对话"""
            self.summary_memory.save_context(
                {"input": user_msg},
                {"output": ai_msg}
            )
            self.message_count += 1

            # 定期总结
            if self.message_count % self.summary_interval == 0:
                print(f"\n⏰ 已进行{self.message_count}轮对话，生成摘要")
                self.show_summary()

        def show_summary(self) -> None:
            """显示摘要"""
            summary = self.summary_memory.load_memory_variables({})
            print(f"📄 摘要：{summary['history']}")

        def get_stats(self) -> dict:
            """获取统计信息"""
            summary = self.summary_memory.load_memory_variables({})
            return {
                "message_count": self.message_count,
                "summary_length": len(summary["history"])
            }

    # 测试
    manager = SmartMemoryManager(llm)

    # 模拟长对话
    topics = [
        "Python基础",
        "Web开发",
        "数据库设计",
        "API开发",
        "测试",
        "部署",
        "性能优化",
    ]

    for i, topic in enumerate(topics, 1):
        print(f"\n第{i}轮：讨论{topic}")
        manager.add_conversation(
            f"我想学习{topic}",
            f"好的，{topic}的关键点是..."
        )

    print("\n" + "="*60)
    print("\n📊 最终统计：")
    stats = manager.get_stats()
    print(f"  总对话轮数：{stats['message_count']}")
    print(f"  摘要长度：{stats['summary_length']}字符")


def main() -> None:
    """主函数"""
    example1_basic_summary_memory()
    print("\n" + "="*70 + "\n")

    example2_summary_vs_buffer()
    print("\n" + "="*70 + "\n")

    example3_summary_buffer_memory()
    print("\n" + "="*70 + "\n")

    example4_custom_summary_prompt()
    print("\n" + "="*70 + "\n")

    example5_incremental_summary()
    print("\n" + "="*70 + "\n")

    example6_memory_management()


if __name__ == "__main__":
    main()

04_chatbot.py - 完整聊天机器人

"""
完整聊天机器人
演示：综合运用记忆功能构建实用的聊天机器人
"""

from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, AIMessage
from typing import Dict, Optional
from datetime import datetime
from dotenv import load_dotenv
import os

load_dotenv()


class SimpleChatBot:
    """简单聊天机器人"""

    def __init__(self):
        """初始化聊天机器人"""
        self.llm = ChatOpenAI(
            base_url=os.getenv("ALIBABA_BASE_URL"),
            api_key=os.getenv("ALIBABA_API_KEY"),
            model="qwen-plus",
            temperature=0.7,
        )

        self.memory = ConversationBufferWindowMemory(
            k=10,  # 保留最近10轮对话
            return_messages=True
        )

        self.conversation = ConversationChain(
            llm=self.llm,
            memory=self.memory,
            verbose=False
        )

    def chat(self, message: str) -> str:
        """发送消息并获取回复"""
        response = self.conversation.predict(input=message)
        return response

    def clear_history(self) -> None:
        """清除对话历史"""
        self.memory.clear()
        print("✨ 对话历史已清除")

    def show_history(self) -> None:
        """显示对话历史"""
        messages = self.memory.chat_memory.messages
        if not messages:
            print("（暂无对话历史）")
            return

        print("\n📜 对话历史：")
        for i, msg in enumerate(messages):
            if isinstance(msg, HumanMessage):
                print(f"  {i+1}. 👤 {msg.content}")
            elif isinstance(msg, AIMessage):
                print(f"     🤖 {msg.content}")


class PersonalAssistant:
    """个人助手（记住用户信息）"""

    def __init__(self):
        """初始化个人助手"""
        self.llm = ChatOpenAI(
            base_url=os.getenv("ALIBABA_BASE_URL"),
            api_key=os.getenv("ALIBABA_API_KEY"),
            model="qwen-plus",
            temperature=0.7,
        )

        self.memory = ConversationBufferWindowMemory(
            k=5,
            return_messages=True,
            memory_key="chat_history"
        )

        # 自定义提示词
        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """你是一个贴心的个人助手。
            
你的职责：
1. 记住用户告诉你的信息（名字、喜好、工作等）
2. 在回答时自然地使用这些信息
3. 主动关心用户
4. 提供有用的建议

请用友好、贴心的语气对话。
"""),
            MessagesPlaceholder(variable_name="chat_history"),
            ("human", "{input}")
        ])

    def chat(self, message: str) -> str:
        """聊天"""
        # 获取历史
        history = self.memory.load_memory_variables({})

        # 生成回复
        messages = self.prompt.format_messages(
            chat_history=history.get("chat_history", []),
            input=message
        )

        response = self.llm.invoke(messages)

        # 保存到记忆
        self.memory.save_context(
            {"input": message},
            {"output": response.content}
        )

        return response.content


class MultiUserChatBot:
    """多用户聊天机器人"""

    def __init__(self):
        """初始化多用户聊天机器人"""
        self.llm = ChatOpenAI(
            base_url=os.getenv("ALIBABA_BASE_URL"),
            api_key=os.getenv("ALIBABA_API_KEY"),
            model="qwen-plus",
            temperature=0.7,
        )

        self.user_memories: Dict[str, ConversationBufferWindowMemory] = {}
        self.user_info: Dict[str, dict] = {}

    def get_or_create_memory(self, user_id: str) -> ConversationBufferWindowMemory:
        """获取或创建用户记忆"""
        if user_id not in self.user_memories:
            self.user_memories[user_id] = ConversationBufferWindowMemory(
                k=5,
                return_messages=True
            )
            self.user_info[user_id] = {
                "created_at": datetime.now(),
                "message_count": 0
            }
            print(f"✨ 为用户 {user_id} 创建新会话")

        return self.user_memories[user_id]

    def chat(self, user_id: str, message: str) -> str:
        """用户聊天"""
        memory = self.get_or_create_memory(user_id)

        # 创建对话链
        conversation = ConversationChain(
            llm=self.llm,
            memory=memory,
            verbose=False
        )

        # 获取回复
        response = conversation.predict(input=message)

        # 更新统计
        self.user_info[user_id]["message_count"] += 1
        self.user_info[user_id]["last_active"] = datetime.now()

        return response

    def get_user_stats(self, user_id: str) -> Optional[dict]:
        """获取用户统计"""
        if user_id not in self.user_info:
            return None

        info = self.user_info[user_id]
        memory = self.user_memories[user_id]

        return {
            "user_id": user_id,
            "created_at": info["created_at"].strftime("%Y-%m-%d %H:%M:%S"),
            "message_count": info["message_count"],
            "last_active": info.get("last_active", info["created_at"]).strftime("%Y-%m-%d %H:%M:%S"),
            "memory_size": len(memory.chat_memory.messages)
        }

    def list_users(self) -> list:
        """列出所有用户"""
        return list(self.user_memories.keys())


def example1_simple_chatbot() -> None:
    """示例1：简单聊天机器人"""
    print("=== 示例1：简单聊天机器人 ===\n")

    bot = SimpleChatBot()

    # 对话序列
    messages = [
        "你好！",
        "我叫小明",
        "我今年20岁",
        "我是大学生",
        "我叫什么名字？",
        "我多大了？",
    ]

    for msg in messages:
        print(f"👤 用户：{msg}")
        response = bot.chat(msg)
        print(f"🤖 机器人：{response}\n")

    # 显示历史
    bot.show_history()


def example2_personal_assistant() -> None:
    """示例2：个人助手"""
    print("\n\n=== 示例2：个人助手 ===\n")

    assistant = PersonalAssistant()

    conversations = [
        "你好，我是张伟",
        "我是一名软件工程师",
        "我喜欢跑步和阅读",
        "我最近在学习AI",
        "能根据我的情况给些学习建议吗？",
    ]

    for msg in conversations:
        print(f"👤 用户：{msg}")
        response = assistant.chat(msg)
        print(f"🤖 助手：{response}\n")


def example3_multi_user_bot() -> None:
    """示例3：多用户聊天机器人"""
    print("\n\n=== 示例3：多用户聊天机器人 ===\n")

    bot = MultiUserChatBot()

    # 用户A的对话
    print("👤 用户A 的对话：")
    print("-" * 50)

    userA_messages = [
        "你好，我是Alice",
        "我是学生",
        "我学的是计算机",
    ]

    for msg in userA_messages:
        print(f"Alice：{msg}")
        response = bot.chat("userA", msg)
        print(f"Bot：{response}\n")

    # 用户B的对话
    print("\n👤 用户B 的对话：")
    print("-" * 50)

    userB_messages = [
        "Hi，我是Bob",
        "我是工程师",
        "我在做Web开发",
    ]

    for msg in userB_messages:
        print(f"Bob：{msg}")
        response = bot.chat("userB", msg)
        print(f"Bot：{response}\n")

    # 继续用户A的对话
    print("\n👤 用户A 继续对话：")
    print("-" * 50)
    print("Alice：我是做什么的？")
    response = bot.chat("userA", "我是做什么的？")
    print(f"Bot：{response}\n")

    # 显示统计
    print("\n📊 系统统计：")
    print("-" * 50)
    for user_id in bot.list_users():
        stats = bot.get_user_stats(user_id)
        print(f"\n用户：{stats['user_id']}")
        print(f"  创建时间：{stats['created_at']}")
        print(f"  消息数量：{stats['message_count']}")
        print(f"  最后活跃：{stats['last_active']}")
        print(f"  记忆大小：{stats['memory_size']}条消息")


def example4_interactive_chatbot() -> None:
    """示例4：交互式聊天机器人（命令行版）"""
    print("\n\n=== 示例4：交互式聊天机器人 ===\n")

    bot = SimpleChatBot()

    print("💬 聊天机器人已启动！")
    print("命令：")
    print("  /history - 查看对话历史")
    print("  /clear   - 清除对话历史")
    print("  /quit    - 退出")
    print("="*60 + "\n")

    # 模拟交互（实际使用时可以用input()）
    demo_messages = [
        "你好",
        "我叫测试用户",
        "/history",
        "我叫什么名字？",
        "/clear",
        "我叫什么名字？",
        "/quit"
    ]

    for message in demo_messages:
        print(f"👤 {message}")

        if message == "/quit":
            print("👋 再见！")
            break
        elif message == "/history":
            bot.show_history()
        elif message == "/clear":
            bot.clear_history()
        else:
            response = bot.chat(message)
            print(f"🤖 {response}")

        print()


def example5_contextual_chatbot() -> None:
    """示例5：上下文感知的聊天机器人"""
    print("\n\n=== 示例5：上下文感知聊天机器人 ===\n")

    class ContextAwareChatBot:
        """上下文感知的聊天机器人"""

        def __init__(self):
            self.llm = ChatOpenAI(
                base_url=os.getenv("ALIBABA_BASE_URL"),
                api_key=os.getenv("ALIBABA_API_KEY"),
                model="qwen-plus",
                temperature=0.7,
            )

            self.memory = ConversationBufferWindowMemory(
                k=5, return_messages=True)
            self.context_info = {}  # 存储提取的上下文信息

        def extract_info(self, message: str, response: str) -> None:
            """从对话中提取关键信息"""
            # 简化版：检测关键词
            if "我叫" in message or "我是" in message:
                # 提取名字（实际应该用NER）
                if "我叫" in message:
                    name = message.split("我叫")[1].split()[0].strip("，。！")
                    self.context_info["name"] = name

            if "岁" in message:
                # 提取年龄
                import re
                age_match = re.search(r'(\d+)岁', message)
                if age_match:
                    self.context_info["age"] = age_match.group(1)

        def chat(self, message: str) -> str:
            """聊天"""
            conversation = ConversationChain(
                llm=self.llm,
                memory=self.memory,
                verbose=False
            )

            response = conversation.predict(input=message)

            # 提取信息
            self.extract_info(message, response)

            return response

        def show_context(self) -> None:
            """显示提取的上下文信息"""
            print("\n📋 提取的上下文信息：")
            if not self.context_info:
                print("  （暂无）")
            else:
                for key, value in self.context_info.items():
                    print(f"  {key}: {value}")

    # 测试
    bot = ContextAwareChatBot()

    messages = [
        "你好，我叫李华",
        "我今年25岁",
        "我是程序员",
    ]

    for msg in messages:
        print(f"👤 {msg}")
        response = bot.chat(msg)
        print(f"🤖 {response}\n")

    bot.show_context()


def main() -> None:
    """主函数"""
    example1_simple_chatbot()
    print("\n" + "="*70 + "\n")

    example2_personal_assistant()
    print("\n" + "="*70 + "\n")

    example3_multi_user_bot()
    print("\n" + "="*70 + "\n")

    example4_interactive_chatbot()
    print("\n" + "="*70 + "\n")

    example5_contextual_chatbot()


if __name__ == "__main__":
    main()

🎨 记忆使用模式

模式1：简单对话机器人

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

memory = ConversationBufferMemory()
conversation = ConversationChain(
    llm=llm,
    memory=memory
)

# 多轮对话
conversation.predict(input="你好")
conversation.predict(input="我叫张三")
conversation.predict(input="我叫什么？")  # 会记得

模式2：自定义记忆键

memory = ConversationBufferMemory(
    memory_key="chat_history",      # 自定义键名
    input_key="user_input",         # 输入键名
    output_key="ai_response",       # 输出键名
    return_messages=True            # 返回消息对象
)

模式3：持久化记忆

# 保存到文件
import json

def save_memory(memory, filename):
    """保存记忆到文件"""
    with open(filename, 'w') as f:
        json.dump(memory.chat_memory.messages, f)

def load_memory(filename):
    """从文件加载记忆"""
    with open(filename, 'r') as f:
        messages = json.load(f)
    # 恢复记忆...

模式4：多会话记忆

# 为不同用户维护独立记忆
user_memories = {}

def get_memory(user_id):
    if user_id not in user_memories:
        user_memories[user_id] = ConversationBufferMemory()
    return user_memories[user_id]

# 使用
memory = get_memory("user_123")

📊 记忆类型对比

类型	优点	缺点	使用场景
Buffer	完整保留实现简单	Token消耗大有长度限制	短对话需要完整上下文
Window	Token可控性能稳定	丢失旧对话可能断上下文	长对话只需近期上下文
Summary	Token最省可处理长对话	可能丢失细节需要额外调用	非常长的对话成本敏感
SummaryBuffer	灵活平衡保留细节	实现复杂需要调优	生产环境企业应用

🔍 常见问题

Q1: 记忆会一直增长吗？

是的，需要管理：

# 方案1：使用窗口记忆
memory = ConversationBufferWindowMemory(k=10)  # 只保留10轮

# 方案2：定期清理
memory.clear()

# 方案3：使用摘要记忆
memory = ConversationSummaryMemory(llm=llm)

Q2: 如何在重启后保留记忆？

# 使用数据库或文件存储
from langchain.memory import ChatMessageHistory
from langchain.memory import ConversationBufferMemory

# 1. 创建持久化存储
message_history = ChatMessageHistory()

# 2. 加载历史消息
# ... 从数据库加载 ...

# 3. 创建带历史的记忆
memory = ConversationBufferMemory(
    chat_memory=message_history
)

Q3: 记忆消耗太多Token怎么办？

优化策略：

# 1. 使用窗口记忆限制轮数
memory = ConversationBufferWindowMemory(k=5)

# 2. 使用摘要记忆
memory = ConversationSummaryMemory(llm=llm)

# 3. 只保存重要信息
def filter_important(message):
    # 自定义逻辑判断是否重要
    if is_important(message):
        memory.save_context(...)

# 4. 定期总结并清理
if len(memory.messages) > 20:
    summary = summarize_conversation(memory)
    memory.clear()
    memory.save_context({"summary": summary}, {})

Q4: 多用户场景如何管理记忆？

from langchain.memory import ConversationBufferMemory
from typing import Dict

class MemoryManager:
    """记忆管理器"""
    
    def __init__(self):
        self.memories: Dict[str, ConversationBufferMemory] = {}
    
    def get_memory(self, session_id: str) -> ConversationBufferMemory:
        """获取或创建会话记忆"""
        if session_id not in self.memories:
            self.memories[session_id] = ConversationBufferMemory()
        return self.memories[session_id]
    
    def clear_memory(self, session_id: str) -> None:
        """清除会话记忆"""
        if session_id in self.memories:
            del self.memories[session_id]
    
    def cleanup_old_sessions(self, max_age: int) -> None:
        """清理过期会话"""
        # 实现清理逻辑
        pass

# 使用
manager = MemoryManager()
memory = manager.get_memory("user_123")

📝 练习任务

练习1：个人信息助手

创建一个记住用户信息的助手：

记住用户的名字、年龄、职业等
在后续对话中使用这些信息
能够更新信息

练习2：智能客服系统

创建带记忆的客服：

记住订单号、问题类型
跨多轮对话解决问题
可以回顾之前讨论的内容

练习3：学习伙伴

创建一个编程学习伙伴：

记住学习进度
记住已经解释过的概念
根据历史调整教学方式

🎯 检查清单

完成本课后，你应该能够：

理解AI记忆的重要性
使用不同类型的记忆
选择合适的记忆类型
管理记忆的增长
实现多用户记忆隔离
持久化和恢复记忆

💡 核心要点

记忆让AI更智能
- 记住上下文
- 提供连贯的对话体验
选择合适的记忆类型
- 短对话：Buffer
- 长对话：Window或Summary
- 生产环境：SummaryBuffer
注意Token消耗
- 记忆会消耗tokens
- 需要定期清理或总结
多用户场景
- 隔离不同用户的记忆
- 考虑持久化方案

📚 下一步

现在你已经学会了让AI记住对话，接下来学习如何让AI自主使用工具：

继续学习： 第6课：Agent和工具 →

类型	优点	缺点	使用场景
Buffer	完整保留实现简单	Token消耗大有长度限制	短对话需要完整上下文
Window	Token可控性能稳定	丢失旧对话可能断上下文	长对话只需近期上下文
Summary	Token最省可处理长对话	可能丢失细节需要额外调用	非常长的对话成本敏感
SummaryBuffer	灵活平衡保留细节	实现复杂需要调优	生产环境企业应用