AI Agent 30天速成｜Day9 笔记

在Day8中，我们学习了如何构建一个基础的ReAct Agent。今天，我们将深入探讨Agent的核心机制之一：记忆与上下文管理。没有记忆的Agent就像金鱼，每次对话都会忘记之前说过什么。通过实现短期记忆（对话历史）和长期记忆（向量数据库），我们可以让Agent真正“记住”用户。### 为什么记忆如此重要？大多数Agent在无状态模式下运行，每次请求都是独立的。但现实世界的任务（如客服对话、代码调试、多轮问答）需要Agent理解上下文。例如：- 用户：“帮我查找昨天的销售数据。”- Agent：执行查询。- 用户：“再按区域分类显示。” - 如果Agent没有记忆，它需要重新理解“再”指的是什么。记忆机制分为两种：1. 短期记忆：当前对话轮次内的上下文（通常存储在内存中）。2. 长期记忆：跨会话的知识（存储在向量数据库或文件系统中）。### 实战1：实现短期记忆我们将使用Python的collections.deque来管理一个固定长度的对话历史，并让Agent在每次回答时参考最近3轮对话。python# agent_with_short_term_memory.pyfrom collections import dequeimport openai # 假设已安装openai库class ShortTermMemoryAgent: def __init__(self, system_prompt: str, max_history: int = 3): self.system_prompt = system_prompt self.history = deque(maxlen=max_history * 2) # 存储用户和助手消息 self.max_history = max_history def add_user_message(self, message: str): """添加用户消息到历史""" self.history.append({"role": "user", "content": message}) def add_assistant_message(self, message: str): """添加助手消息到历史""" self.history.append({"role": "assistant", "content": message}) def get_context_messages(self) -> list: """构建包含系统提示和历史的消息列表""" messages = [{"role": "system", "content": self.system_prompt}] # 只取最近max_history轮对话（每轮包含用户+助手两条消息） recent = list(self.history)[-self.max_history * 2:] messages.extend(recent) return messages def chat(self, user_input: str) -> str: """处理用户输入并返回响应""" self.add_user_message(user_input) context = self.get_context_messages() # 调用OpenAI API（此处使用模拟响应进行演示） # response = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=context) # simulated_response = response.choices[0].message.content simulated_response = f"根据上下文，我理解你说的是：{user_input}" self.add_assistant_message(simulated_response) return simulated_response# 使用示例if __name__ == "__main__": agent = ShortTermMemoryAgent("你是一个友好的助手。") # 第一轮对话 response1 = agent.chat("今天天气怎么样？") print(f"用户: 今天天气怎么样？\n助手: {response1}\n") # 第二轮对话（引用前一轮） response2 = agent.chat("那明天呢？") print(f"用户: 那明天呢？\n助手: {response2}\n") # 查看当前历史 print("当前对话历史:", list(agent.history))关键点：- 使用deque的maxlen自动丢弃最旧的消息。- get_context_messages确保只传递最近N轮对话，避免超出模型token限制。- 实际部署中，建议将max_history设置为3-5轮，根据模型上下文长度调整。### 实战2：引入长期记忆（向量数据库）短期记忆在会话结束后就丢失了。为了让Agent记住跨会话的信息（如用户偏好、历史事实），我们需要引入长期记忆。这里使用chromadb作为轻量级向量数据库。python# agent_with_long_term_memory.pyimport chromadbfrom chromadb.utils import embedding_functionsimport uuidclass LongTermMemoryAgent: def __init__(self, collection_name: str = "agent_memory"): # 初始化ChromaDB客户端 self.client = chromadb.Client() self.collection = self.client.get_or_create_collection( name=collection_name, embedding_function=embedding_functions.DefaultEmbeddingFunction() # 使用默认嵌入 ) self.short_term = [] # 短期对话历史 def remember(self, fact: str, metadata: dict = None): """存储一条事实到长期记忆""" unique_id = str(uuid.uuid4()) self.collection.add( documents=[fact], ids=[unique_id], metadatas=[metadata or {"source": "user_input"}] ) print(f"已记住: {fact}") def recall(self, query: str, top_k: int = 3) -> list: """根据查询检索相关记忆""" results = self.collection.query( query_texts=[query], n_results=top_k ) # 返回匹配的文档内容 return results['documents'][0] if results['documents'] else [] def chat_with_memory(self, user_input: str) -> str: """结合长期记忆进行对话""" # 1. 检索相关长期记忆 relevant_memories = self.recall(user_input) memory_context = "\n".join([f"相关记忆: {mem}" for mem in relevant_memories]) # 2. 构建短期上下文 short_context = "\n".join(self.short_term[-5:]) # 最近5条消息 # 3. 生成响应（模拟） response = f"根据长期记忆:\n{memory_context}\n\n短期上下文:\n{short_context}\n\n我的回答: 基于以上信息，我..." # 4. 更新短期记忆 self.short_term.append(f"用户: {user_input}") self.short_term.append(f"助手: {response}") return response# 使用示例if __name__ == "__main__": agent = LongTermMemoryAgent() # 第一阶段：存储长期记忆 agent.remember("用户喜欢编程语言Python") agent.remember("用户上次提问了关于AI Agent的问题") print("--- 新会话开始 ---") # 第二阶段：新会话中回忆 response = agent.chat_with_memory("我上次问了什么？") print(response) print("\n--- 再次提问 ---") response2 = agent.chat_with_memory("我喜欢什么编程语言？") print(response2)关键点：- 使用chromadb的query方法进行语义搜索，而非关键词匹配。- 长期记忆存储为向量，可以跨会话检索。- 实际项目中，建议将用户ID或会话ID作为metadata的一部分，实现多用户隔离。### 记忆管理的进阶技巧1. 记忆过期机制：给每条记忆添加时间戳，定期清理超过一定时间的记忆。2. 重要性评分：对记忆进行优先级排序，只保留重要信息（如通过用户反馈评分）。3. 记忆压缩：将多条相关记忆合并为摘要，减少存储空间。示例代码片段（记忆压缩）：pythondef compress_memories(memories: list, max_tokens: int = 100) -> str: """将多条记忆压缩为摘要""" combined = "; ".join(memories) # 实际项目中使用LLM生成摘要 # response = openai.ChatCompletion.create( # model="gpt-3.5-turbo", # messages=[{"role": "user", "content": f"请将以下记忆压缩为100字摘要: {combined}"}] # ) return combined[:max_tokens] # 简化版### 总结Day9我们实现了Agent记忆系统的两个核心组件：- 短期记忆：通过deque管理对话历史，确保Agent能理解多轮对话的上下文。- 长期记忆：使用向量数据库存储和检索跨会话信息，实现“记住”能力。关键收获：1. 记忆是Agent从“弱智”走向“智能”的关键一步。2. 记忆管理需要平衡：太多记忆会浪费token，太少则失去上下文。3. 生产环境中，建议结合Redis实现分布式记忆存储，并添加过期策略。明天（Day10），我们将学习如何让Agent具备“规划”能力——不仅回答问题，还能自主分解复杂任务。记住，没有记忆的Agent只是工具，有记忆的Agent才是伙伴。