乐于分享
好东西不私藏

AI Agent零基础入门第六节:记忆系统与上下文管理

AI Agent零基础入门第六节:记忆系统与上下文管理

前言

欢迎来到AI Agent零基础入门教程的第六课!在之前的学习中,我们已经掌握了如何构建能够执行任务的AI Agent。然而,一个真正智能的Agent不仅需要能够完成任务,还需要能够”记住”之前的交互内容,理解对话的上下文。

本节课我们将深入学习Agent的记忆系统,这是实现真正智能对话的关键技术。我们将学习不同类型的记忆系统、如何实现短期记忆和长期记忆、以及如何管理对话上下文。

通过本节课的学习,你将能够构建具有持续记忆能力的智能Agent,为用户提供更加连贯和个性化的服务。

为什么Agent需要记忆系统

没有记忆的问题

想象一下,如果一个人在和你对话时,每次你说完话他就忘记了你之前说了什么,你会作何感想?这正是许多早期AI系统的现状——它们是”无记忆”的,每次对话都是独立的。

无记忆Agent的问题: 1. 无法理解代词(如”它”、“那个”指代什么) 2. 无法保持话题连贯性 3. 重复询问相同的信息 4. 无法记住用户的偏好和习惯 5. 无法进行真正的多轮对话

记忆系统的价值

一个好的记忆系统能让Agent: 1. 理解上下文:知道对话的背景和之前的内容 2. 保持连贯:在多轮对话中保持话题一致性 3. 个性化服务:记住用户的偏好,提供定制化服务 4. 积累知识:从历史交互中学习,不断改进服务 5. 情感连接:建立与用户的长期关系

┌─────────────────────────────────────────────────────────────┐│                    Agent 记忆系统架构                        │├─────────────────────────────────────────────────────────────┤│                                                             ││  ┌─────────────────────────────────────────────────────┐  ││  │                    长期记忆                            │  ││  │  (Long-term Memory)                                   │  ││  │  • 用户配置信息                                       │  ││  │  • 历史交互摘要                                       │  ││  │  • 偏好和习惯                                         │  ││  │  • 持久化存储                                         │  ││  └─────────────────────────────────────────────────────┘  ││                            │                                ││                            ▼                                ││  ┌─────────────────────────────────────────────────────┐  ││  │                    短期记忆                            │  ││  │  (Short-term Memory)                                  │  ││  │  • 当前对话内容                                       │  ││  │  • 最近的消息                                         │  ││  │  • 上下文窗口                                         │  ││  └─────────────────────────────────────────────────────┘  ││                            │                                ││                            ▼                                ││  ┌─────────────────────────────────────────────────────┐  ││  │                    工作记忆                            │  ││  │  (Working Memory)                                     │  ││  │  • 当前任务状态                                       │  ││  │  • 临时变量                                           │  ││  │  • 会话状态                                           │  ││  └─────────────────────────────────────────────────────┘  ││                                                             │└─────────────────────────────────────────────────────────────┘

记忆的类型

1. 短期记忆(Short-term Memory)

短期记忆存储当前会话中的信息,类似人类的”工作记忆”。当会话结束时,这些信息通常会被清除。

特点: - 仅在当前会话有效 - 容量有限(通常几千到几万token) - 响应速度快 - 通常存储在内存中

2. 长期记忆(Long-term Memory)

长期记忆存储跨会话的信息,类似人类的”永久记忆”。这些信息会持久保存,即使关闭应用也不会丢失。

特点: - 持久存储 - 容量大 - 访问速度相对较慢 - 需要索引和检索机制

3. 语义记忆(Semantic Memory)

语义记忆以结构化方式存储知识,允许Agent根据语义相似性检索信息。这通常通过向量数据库实现。

特点: - 基于语义相似性检索 - 支持模糊匹配 - 需要Embedding模型 - 适合知识库应用

LangChain记忆组件详解

ConversationBufferMemory

最简单的记忆类型,存储完整的对话历史:

from langchain.memory import ConversationBufferMemorymemory = ConversationBufferMemory(    memory_key="chat_history",    return_messages=True,    output_key="output")

ConversationBufferWindowMemory

限制存储最近N轮对话,节省token:

from langchain.memory import ConversationBufferWindowMemorymemory = ConversationBufferWindowMemory(    memory_key="chat_history",    k=5,  # 只保留最近5轮对话    return_messages=True)

ConversationTokenBufferMemory

基于token数量限制记忆大小:

from langchain.memory import ConversationTokenBufferMemorymemory = ConversationTokenBufferMemory(    llm=llm,    max_token_limit=1000,  # 最多保留1000个token    memory_key="chat_history")

ConversationSummaryMemory

自动生成对话摘要,节省token:

from langchain.memory import ConversationSummaryMemorymemory = ConversationSummaryMemory(    llm=llm,    memory_key="chat_history",    return_messages=True)

实战:构建具有记忆的Agent

项目1:基本对话记忆

创建01_basic_memory.py:

"""LangChain 记忆系统实战:基本对话记忆本程序展示如何使用ConversationBufferMemory实现对话记忆"""import osfrom dotenv import load_dotenvfrom langchain.agents import AgentType, initialize_agent, load_toolsfrom langchain_openai import OpenAIfrom langchain.memory import ConversationBufferMemoryfrom langchain_core.messages import HumanMessage, AIMessage# 加载环境变量load_dotenv()def create_memory_agent():"""创建具有记忆功能的Agent    Returns:        tuple: (AgentExecutor, ConversationBufferMemory)    """print("�� 正在创建带记忆的Agent...")# 1. 初始化语言模型    llm = OpenAI(        temperature=0.7,        model="gpt-3.5-turbo-instruct"    )# 2. 加载工具    tools = load_tools(["serpapi", "llm-math"], llm=llm)# 3. 创建记忆组件    memory = ConversationBufferMemory(        memory_key="chat_history",        return_messages=True,        output_key="output"    )print(f"  ✅ 记忆组件创建完成")# 4. 创建Agent    agent = initialize_agent(        tools,        llm,        agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,        memory=memory,        verbose=True,        handle_parsing_errors=True    )print("✅ 带记忆的Agent创建成功!\n")return agent, memorydef demonstrate_memory(agent, memory):"""演示记忆功能    """print("\n"+"="*60)print("�� 记忆功能演示")print("="*60)# 第一轮对话print("\n--- 对话 1 ---")    q1 ="我喜欢摄影,尤其是风景摄影。"print(f"�� 用户:{q1}")    a1 = agent.run(q1)print(f"�� Agent:{a1}")# 第二轮对话print("\n--- 对话 2 ---")    q2 ="你能推荐一些适合摄影的地方吗?"print(f"�� 用户:{q2}")    a2 = agent.run(q2)print(f"�� Agent:{a2}")# 第三轮对话(测试Agent是否能记住之前的爱好)print("\n--- 对话 3 ---")    q3 ="根据我的爱好,你推荐的地方有什么共同特点?"print(f"�� 用户:{q3}")    a3 = agent.run(q3)print(f"�� Agent:{a3}")# 查看记忆内容print("\n--- 记忆内容检查 ---")    chat_history = memory.load_memory_variables({})print(f"记忆中的对话数:{len(chat_history.get('chat_history', []))}")print(f"记忆长度:{len(str(chat_history))}字符")# 打印记忆内容print("\n��记忆内容详情:")for i, msg inenumerate(chat_history.get('chat_history', []), 1):        msg_type ="用户"ifisinstance(msg, HumanMessage) else"Agent"        content = msg.content[:100] +"..."iflen(msg.content) >100else msg.contentprint(f"  {i}. [{msg_type}]: {content}")def interactive_conversation(agent, memory):"""交互式对话(带记忆)    """print("""╔════════════════════════════════════════════════════════════╗║                                                            ║║            �� 带记忆的Agent - 交互模式                     ║║                                                            ║║    Agent会记住之前的对话内容                                ║║    输入 'history' 查看对话历史                              ║║    输入 'clear' 清空记忆                                   ║║    输入 'exit' 退出                                        ║║                                                            ║╚════════════════════════════════════════════════════════════╝    """)whileTrue:try:            user_input =input("\n��请输入(输入exit退出):\n> ").strip()if user_input.lower() in ['exit', 'quit']:print("\n��感谢使用!再见!")breakif user_input.lower() =='clear':                memory.clear()print("��️ 记忆已清空")continueif user_input.lower() =='history':                chat_history = memory.load_memory_variables({})print("\n��对话历史:")for i, msg inenumerate(chat_history.get('chat_history', []), 1):                    msg_type ="用户"ifisinstance(msg, HumanMessage) else"Agent"print(f"  {i}. [{msg_type}]: {msg.content[:200]}...")continueifnot user_input:continueprint("\n�� Agent正在思考...\n")            response = agent.run(user_input)print(f"\n�� Agent:{response}")exceptKeyboardInterrupt:print("\n\n��再见!")breakexceptExceptionas e:print(f"\n❌ 错误:{e}")def main():"""主函数"""print("""╔════════════════════════════════════════════════════════════╗║                                                            ║║          �� LangChain 记忆系统实战:基本对话记忆           ║║                                                            ║║          学习如何使用ConversationBufferMemory             ║║                                                            ║╚════════════════════════════════════════════════════════════╝    """)    agent, memory = create_memory_agent()    demonstrate_memory(agent, memory)    interactive_conversation(agent, memory)if__name__=="__main__":    main()

项目2:自动摘要记忆

创建02_summary_memory.py:

"""LangChain 记忆系统实战:自动摘要记忆本程序展示如何使用ConversationSummaryMemory节省token"""import osfrom dotenv import load_dotenvfrom langchain.agents import AgentType, initialize_agent, load_toolsfrom langchain_openai import OpenAIfrom langchain.memory import ConversationSummaryMemory# 加载环境变量load_dotenv()def create_summary_agent():"""创建使用摘要记忆的Agent    """print("�� 正在创建摘要记忆Agent...")# 1. 初始化语言模型    llm = OpenAI(        temperature=0.7,        model="gpt-3.5-turbo-instruct"    )# 2. 加载工具    tools = load_tools(["serpapi"], llm=llm)# 3. 创建摘要记忆# ConversationSummaryMemory会自动将长对话压缩成摘要    memory = ConversationSummaryMemory(        llm=llm,        memory_key="chat_history",        return_messages=True    )print("  ✅ 摘要记忆组件创建完成")print("     特点:长对话会被自动压缩成摘要,节省token")# 4. 创建Agent    agent = initialize_agent(        tools,        llm,        agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,        memory=memory,        verbose=True,        handle_parsing_errors=True    )print("✅ 摘要记忆Agent创建成功!\n")return agent, memorydef demonstrate_summary_memory(agent, memory):"""演示摘要记忆功能    """print("\n"+"="*60)print("�� 摘要记忆功能演示")print("="*60)# 进行多轮对话    conversations = ["我叫张三,是一名软件工程师,在北京工作。","我最近在学习人工智能相关的技术。","Python是我最常用的编程语言。","我每天工作8小时,周末会休息。","我的目标是成为一名AI技术专家。"    ]for i, msg inenumerate(conversations, 1):print(f"\n--- 对话 {i} ---")print(f"�� 用户:{msg}")        response = agent.run(msg)print(f"�� Agent:{response}")# 显示记忆状态        memory_variables = memory.load_memory_variables({})        summary = memory_variables.get('chat_history', '')print(f"\n��记忆摘要长度:{len(summary)}字符")# 如果是摘要(而不是完整对话),显示摘要if summary and'摘要'in summary orlen(summary) <100:print(f"  �� 记忆内容:{summary[:200]}...")def compare_memory_types():"""对比不同记忆类型的效率    """print("\n"+"="*60)print("�� 不同记忆类型对比")print("="*60)# 模拟长对话    sample_messages = ["我叫张三,是一名软件工程师。","我每天写代码8小时,主要使用Python。","Python是一门高级编程语言。","我最近在学习机器学习。","机器学习是人工智能的一个分支。","深度学习是机器学习的一个子领域。","我使用PyTorch框架进行深度学习。","PyTorch由Facebook开发。","TensorFlow由Google开发。","两个框架都很受欢迎。"    ]from langchain.memory import (        ConversationBufferMemory,        ConversationBufferWindowMemory,        ConversationSummaryMemory    )# 创建不同类型的记忆    buffer_memory = ConversationBufferMemory(memory_key="chat")    window_memory = ConversationBufferWindowMemory(k=3, memory_key="chat")    summary_memory = ConversationSummaryMemory(        llm=OpenAI(temperature=0),        memory_key="chat"    )print("\n��模拟10轮对话后,各记忆类型的存储方式:\n")# 添加消息for msg in sample_messages:        buffer_memory.chat_memory.add_user_message(msg)        window_memory.chat_memory.add_user_message(msg)        summary_memory.chat_memory.add_user_message(msg)# 获取各记忆的输出    buffer_output = buffer_memory.load_memory_variables({})['chat']    window_output = window_memory.load_memory_variables({})['chat']    summary_output = summary_memory.load_memory_variables({})['chat']print(f"1️⃣ BufferMemory(完整记忆):")print(f"   存储了全部 {len(sample_messages)}条消息")print(f"   总字符数:{len(buffer_output)}")print(f"   ⚠️ 问题:对话越长,token消耗越大\n")print(f"2️⃣ WindowMemory(窗口记忆):")print(f"   仅保留最近 3 条消息")print(f"   总字符数:{len(window_output)}")print(f"   ⚠️ 问题:可能丢失重要历史信息\n")print(f"3️⃣ SummaryMemory(摘要记忆):")print(f"   将对话压缩成摘要")print(f"   总字符数:{len(summary_output)}")print(f"   ✅ 优势:保持信息的同时节省token\n")def main():"""主函数"""print("""╔════════════════════════════════════════════════════════════╗║                                                            ║║       �� LangChain 记忆系统实战:自动摘要记忆               ║║                                                            ║║       学习如何使用ConversationSummaryMemory                ║║                                                            ║╚════════════════════════════════════════════════════════════╝    """)print("�� 首先对比不同记忆类型...\n")    compare_memory_types()print("\n"+"="*60)print("�� 接下来演示摘要记忆的实际效果")print("="*60)    agent, memory = create_summary_agent()    demonstrate_summary_memory(agent, memory)if__name__=="__main__":    main()

项目3:向量存储记忆

创建03_vector_memory.py:

"""LangChain 记忆系统实战:向量存储记忆本程序展示如何使用向量数据库实现语义记忆"""import osfrom dotenv import load_dotenvfrom langchain.agents import AgentType, initialize_agent, load_toolsfrom langchain_openai import OpenAI, OpenAIEmbeddingsfrom langchain.memory import VectorStoreRetrieverMemoryfrom langchain.docstore import InMemoryDocstorefrom langchain.vectorstores import Chromaimport faiss# 加载环境变量load_dotenv()def create_vector_memory_agent():"""创建使用向量存储记忆的Agent    """print("�� 正在创建向量记忆Agent...")# 1. 初始化语言模型和嵌入模型    llm = OpenAI(temperature=0.7)    embeddings = OpenAIEmbeddings()# 2. 加载工具    tools = load_tools(["serpapi"], llm=llm)# 3. 创建向量存储# 使用FAISS向量数据库(支持本地存储)print("  �� 创建向量存储...")# 方法1:使用内存向量存储(适合测试)# 在生产环境中建议使用持久化存储    embedding_dim =1536# OpenAI embeddings的维度    index = faiss.IndexFlatL2(embedding_dim)    vectorstore = Chroma(        embedding_function=embeddings,        index=index,        docstore=InMemoryDocstore()    )# 4. 创建向量记忆    memory = VectorStoreRetrieverMemory(        retriever=vectorstore.as_retriever(            search_kwargs={"k": 3}  # 检索最相关的3条记忆        ),        memory_key="chat_history",        input_key="input",  # 使用用户输入作为检索键        return_messages=True    )# 添加一些初始记忆print("  �� 添加初始记忆...")    memory.save_context(        {"input": "用户叫张三,是一名Python程序员"},        {"output": "好的,我记住了,张三是一位Python程序员。"}    )    memory.save_context(        {"input": "张三最喜欢的编程语言是Python"},        {"output": "明白了,Python是张三最喜欢的编程语言。"}    )    memory.save_context(        {"input": "张三目前在学习机器学习"},        {"output": "好的,我会记住张三正在学习机器学习。"}    )print("  ✅ 初始记忆添加完成")print(f"     记忆数量:3条\n")# 5. 创建Agent    agent = initialize_agent(        tools,        llm,        agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,        memory=memory,        verbose=True,        handle_parsing_errors=True    )print("✅ 向量记忆Agent创建成功!\n")return agent, memorydef demonstrate_vector_memory(agent, memory):"""演示向量记忆功能    """print("\n"+"="*60)print("�� 向量记忆功能演示")print("="*60)# 测试1:检索相关记忆print("\n--- 测试1:检索相关记忆 ---")    query ="用户的学习方向是什么?"# 手动检索看看能找到什么    retrieved = memory.load_memory_variables({"input": query})print(f"问题:{query}")print(f"检索到的记忆:{retrieved['chat_history']}")# 测试2:对话时自动使用记忆print("\n--- 测试2:利用记忆回答 ---")    q ="张三正在学习什么技术?"print(f"�� 用户:{q}")    a = agent.run(q)print(f"�� Agent:{a}")# 测试3:模糊匹配print("\n--- 测试3:模糊匹配 ---")    q ="用户的职业是什么?"print(f"�� 用户:{q}")    a = agent.run(q)print(f"�� Agent:{a}")# 测试4:语义相关但不包含关键词print("\n--- 测试4:语义相关检索 ---")    q ="他平时用什么语言写代码?"print(f"�� 用户:{q}")    a = agent.run(q)print(f"�� Agent:{a}")def add_new_memories(agent, memory):"""添加新记忆    """print("\n"+"="*60)print("�� 添加新记忆演示")print("="*60)# 通过对话添加记忆    new_info = ["我最近在研究深度学习中的Transformer架构","Transformer是GPT和BERT等模型的基础","我计划在下个月参加机器学习比赛"    ]for info in new_info:print(f"\n��用户:{info}")        response = agent.run(info)print(f"�� Agent:{response}")# 测试新记忆是否可用print("\n--- 测试新记忆 ---")    q ="张三提到过什么技术或计划?"print(f"�� 用户:{q}")    a = agent.run(q)print(f"�� Agent:{a}")def main():"""主函数"""print("""╔════════════════════════════════════════════════════════════╗║                                                            ║║       �� LangChain 记忆系统实战:向量存储记忆              ║║                                                            ║║       学习如何使用VectorStoreRetrieverMemory              ║║                                                            ║╚════════════════════════════════════════════════════════════╝    """)    agent, memory = create_vector_memory_agent()    demonstrate_vector_memory(agent, memory)    add_new_memories(agent, memory)if__name__=="__main__":    main()

项目4:组合多种记忆

创建04_combined_memory.py:

"""LangChain 记忆系统实战:组合多种记忆本程序展示如何组合使用多种记忆类型"""import osfrom dotenv import load_dotenvfrom langchain.agents import AgentType, initialize_agent, load_toolsfrom langchain_openai import OpenAIfrom langchain.memory import (    CombinedMemory,    ConversationBufferMemory,    ConversationSummaryMemory)# 加载环境变量load_dotenv()class CombinedMemoryManager:"""组合记忆管理器管理多种类型的记忆    """def__init__(self, llm):"""初始化组合记忆管理器        Args:            llm: 语言模型实例        """self.llm = llm# 1. 对话缓冲记忆(短期记忆)self.buffer_memory = ConversationBufferMemory(            memory_key="recent_conversation",            return_messages=True,            output_key="output"        )# 2. 摘要记忆(长期记忆)self.summary_memory = ConversationSummaryMemory(            llm=llm,            memory_key="conversation_summary",            return_messages=True        )# 3. 用户信息记忆(持久信息)self.user_memory = ConversationBufferMemory(            memory_key="user_info",            return_messages=True,            output_key="output"        )# 4. 组合所有记忆self.combined_memory = CombinedMemory(            memories=[self.buffer_memory,self.summary_memory,self.user_memory            ]        )def load_memory_variables(self, inputs):"""加载所有记忆变量"""returnself.combined_memory.load_memory_variables(inputs)def save_context(self, inputs, outputs):"""保存上下文到所有记忆"""self.combined_memory.save_context(inputs, outputs)def clear_all(self):"""清空所有记忆"""self.buffer_memory.clear()self.summary_memory.clear()self.user_memory.clear()def create_combined_memory_agent():"""创建使用组合记忆的Agent    """print("�� 正在创建组合记忆Agent...")# 1. 初始化语言模型    llm = OpenAI(        temperature=0.7,        model="gpt-3.5-turbo-instruct"    )# 2. 加载工具    tools = load_tools(["serpapi"], llm=llm)# 3. 创建组合记忆管理器    memory_manager = CombinedMemoryManager(llm)# 预设用户信息print("  �� 预设用户信息...")    memory_manager.user_memory.save_context(        {"input": "用户叫李明,是一名大学生"},        {"output": "好的,已记录用户信息。"}    )    memory_manager.user_memory.save_context(        {"input": "李明在清华大学学习计算机科学"},        {"output": "好的,已记住李明的学校和专业。"}    )print("  ✅ 组合记忆创建完成")print("     包含:短期记忆、摘要记忆、用户信息记忆\n")# 4. 创建自定义提示模板,让Agent理解组合记忆    prompt ="""你是一个智能助手,正在与用户进行对话。【用户信息】(长期记住){user_info}【最近对话】(当前会话){recent_conversation}【对话摘要】(重要内容){conversation_summary}当前对话:Human: {human_input}AI: """# 5. 创建Agent    agent = initialize_agent(        tools,        llm,        agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,        memory=memory_manager.combined_memory,        verbose=True,        prompt=prompt,        handle_parsing_errors=True    )print("✅ 组合记忆Agent创建成功!\n")return agent, memory_managerdef demonstrate_combined_memory(agent, memory_manager):"""演示组合记忆功能    """print("\n"+"="*60)print("�� 组合记忆功能演示")print("="*60)# 查看用户信息记忆print("\n��用户信息记忆:")    user_info = memory_manager.user_memory.load_memory_variables({})print(user_info.get('user_info', '无'))# 进行对话    conversations = ["我今天心情很好!","刚才收到了一封邮件,你知道怎么写正式的商务邮件吗?","谢谢!你总是这么有帮助。","对了,我明天有个考试,要复习计算机网络。"    ]for i, msg inenumerate(conversations, 1):print(f"\n--- 对话 {i} ---")print(f"�� 用户:{msg}")        response = agent.run(msg)print(f"�� Agent:{response}")# 显示记忆状态        all_memory = memory_manager.load_memory_variables({})print(f"\n��记忆状态:")print(f"     用户信息:{len(all_memory.get('user_info', ''))}字符")print(f"     最近对话:{len(all_memory.get('recent_conversation', ''))}字符")print(f"     对话摘要:{len(all_memory.get('conversation_summary', ''))}字符")# 测试综合检索print("\n--- 测试综合记忆理解 ---")    q ="根据我们的对话,你能总结一下关于我的哪些信息?"print(f"�� 用户:{q}")    a = agent.run(q)print(f"�� Agent:{a}")def main():"""主函数"""print("""╔════════════════════════════════════════════════════════════╗║                                                            ║║       �� LangChain 记忆系统实战:组合多种记忆              ║║                                                            ║║       学习如何组合使用多种记忆类型                         ║║                                                            ║╚════════════════════════════════════════════════════════════╝    """)    agent, memory_manager = create_combined_memory_agent()    demonstrate_combined_memory(agent, memory_manager)if__name__=="__main__":    main()

上下文窗口管理

什么是上下文窗口

语言模型在处理文本时有最大token限制,这被称为”上下文窗口”。以GPT-3.5为例,上下文窗口约为16K tokens。这意味着: - 输入提示 - 对话历史 - Agent的思考过程 - 输出内容

所有这些加起来不能超过这个限制。

管理策略

策略1:截断旧消息

from langchain.memory import ConversationBufferWindowMemory# 只保留最近5轮对话memory = ConversationBufferWindowMemory(    k=5,    memory_key="chat_history",    return_messages=True)

策略2:压缩为摘要

from langchain.memory import ConversationSummaryMemory# 自动将长对话压缩成摘要memory = ConversationSummaryMemory(    llm=llm,    memory_key="chat_history")

策略3:基于重要性保留

from langchain.memory import BaseChatMessageHistoryfrom langchain_core.messages import BaseMessageclass SelectiveMemory(BaseChatMessageHistory):"""选择性记忆:只保留重要的消息    """def__init__(self):self.messages = []self.important_keywords = ["重要", "记住", "提醒", "计划", "预约","名字", "生日", "地址", "电话", "邮箱"        ]def add_message(self, message: BaseMessage):"""添加消息,自动过滤不重要消息"""        content = message.content.lower()# 检查是否包含重要关键词        is_important =any(kw in content for kw inself.important_keywords)# 或者是人名、日期等实体ifany(c.isupper() for c in message.content[:10]):  # 开头大写可能是新话题            is_important =Trueif is_important orlen(self.messages) <10:self.messages.append(message)

记忆系统的持久化

保存记忆到文件

import jsonfrom pathlib import Pathdef save_memory(memory, filepath: str):"""保存记忆到文件    """    messages = []for msg in memory.chat_memory.messages:        messages.append({"type": type(msg).__name__,"content": msg.content        })withopen(filepath, 'w', encoding='utf-8') as f:        json.dump(messages, f, ensure_ascii=False, indent=2)print(f"记忆已保存到:{filepath}")def load_memory(memory, filepath: str):"""从文件加载记忆    """ifnot Path(filepath).exists():print(f"文件不存在:{filepath}")returnwithopen(filepath, 'r', encoding='utf-8') as f:        messages = json.load(f)from langchain_core.messages import HumanMessage, AIMessagefor msg in messages:if msg['type'] =='HumanMessage':            memory.chat_memory.add_user_message(msg['content'])else:            memory.chat_memory.add_ai_message(msg['content'])print(f"已从文件加载 {len(messages)}条记忆")

保存向量记忆

def save_vector_memory(vectorstore, persist_directory: str):"""保存向量记忆到磁盘    """    vectorstore.save(persist_directory)print(f"向量记忆已保存到:{persist_directory}")def load_vector_memory(persist_directory: str, embeddings):"""从磁盘加载向量记忆    """return Chroma.load_local(persist_directory, embeddings)

常见问题与解决方案

问题1:记忆丢失

问题描述:重新启动后记忆丢失。

解决方案: 1. 实现记忆持久化(保存到文件或数据库) 2. 使用向量数据库存储长期记忆 3. 在启动时加载历史记忆

问题2:Token超限

问题描述:对话历史太长导致Token超限。

解决方案: 1. 使用窗口记忆限制长度 2. 使用摘要记忆压缩历史 3. 实现智能截断策略 4. 分层管理记忆(短期+长期)

问题3:检索不准确

问题描述:向量检索返回不相关的结果。

解决方案: 1. 调整检索的k值 2. 使用更好的嵌入模型 3. 实现重排序机制 4. 结合关键词检索

总结与下节预告

本节课,我们深入学习了Agent的记忆系统:

1.为什么需要记忆:理解了记忆系统的重要性

2.记忆类型:学习了短期、长期、语义三种记忆

3.记忆组件:掌握了各种记忆组件的使用方法

4.实战应用:创建了多个具有记忆功能的Agent

5.上下文管理:学会了如何管理上下文窗口

6.持久化:学习了如何保存和加载记忆

下一节课,我们将学习多Agent协作,这是构建复杂AI系统的关键技术。我们将学习: - 什么是多Agent架构 - 如何让多个Agent协作工作 - Agent网络的设计模式 - 构建一个多Agent协作系统

敬请期待!

练习题

1.实践题:修改示例代码,将记忆保存到本地文件

2.思考题:分析不同记忆类型的适用场景

3.拓展题:研究如何实现记忆的自动过期和清理

4.挑战题:构建一个支持用户信息持久化的个人助手

相关学习资料