
前言
欢迎来到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.挑战题:构建一个支持用户信息持久化的个人助手
夜雨聆风