前两篇我们实现了工具调用和多智能体协作。但有一个关键短板：每个对话都是独立的，智能体转眼就忘了你。
真正的智能体需要长期记忆——记住你的偏好、之前讨论过的方案、甚至跨会话的知识。今天我们就用向量数据库 + 对话摘要，给智能体装上持久记忆。
记忆的三层架构
· 短期记忆：当前会话的完整对话（上下文窗口）
· 工作记忆：本轮任务中的中间结果、计划步骤
· 长期记忆：跨会话存储，通过向量检索召回
我们重点实现长期记忆，并用一个简单的分层检索策略。
技术选型
· 向量数据库：Chroma（轻量、嵌入式）
· 嵌入模型：text-embedding-3-small（OpenAI）
· 记忆管理：自动存储 + 语义搜索
第一步：初始化记忆系统
```python
import chromadb
from chromadb.utils import embedding_functions
import hashlib
from datetime import datetime
from typing import List, Dict, Any
import json
class LongTermMemory:
def __init__(self, collection_name: str = "agent_memory"):
# 持久化到本地目录
self.client = chromadb.PersistentClient(path="./chroma_memory")
self.embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
model_name="text-embedding-3-small"
)
# 获取或创建集合
self.collection = self.client.get_or_create_collection(
name=collection_name,
embedding_function=self.embedding_fn
)
def add_memory(self, content: str, metadata: Dict[str, Any] = None):
"""存储一条记忆"""
if metadata is None:
metadata = {}
# 生成唯一ID（基于内容和时间戳）
timestamp = datetime.now().isoformat()
unique_id = hashlib.md5(f"{content}_{timestamp}".encode()).hexdigest()
# 自动添加时间戳
metadata["timestamp"] = timestamp
self.collection.add(
documents=[content],
metadatas=[metadata],
ids=[unique_id]
)
return unique_id
def recall(self, query: str, n_results: int = 5) -> List[Dict]:
"""根据查询召回最相关的记忆"""
results = self.collection.query(
query_texts=[query],
n_results=n_results
)
memories = []
if results['documents'] and results['documents'][0]:
for i, doc in enumerate(results['documents'][0]):
memories.append({
"content": doc,
"metadata": results['metadatas'][0][i] if results['metadatas'] else {},
"distance": results['distances'][0][i] if results['distances'] else None
})
return memories
def forget(self, memory_id: str):
"""删除特定记忆"""
self.collection.delete(ids=[memory_id])
def clear_all(self):
"""清空所有记忆（慎用）"""
all_ids = self.collection.get()['ids']
if all_ids:
self.collection.delete(ids=all_ids)
```
第二步：带记忆的智能体
我们将记忆整合到之前实现的Agent类中。关键策略：
1. 每次用户输入，先召回相关长期记忆，注入到系统提示中
2. 每次对话结束后，存储重要信息（可配置为只存储关键内容）
```python
import openai
from openai import AsyncOpenAI
from typing import List, Dict, Optional
import json
class MemorableAgent:
def __init__(self,
system_prompt: str,
memory: LongTermMemory,
memory_trigger_words: List[str] = None):
self.client = AsyncOpenAI()
self.system_prompt = system_prompt
self.memory = memory
self.messages = [{"role": "system", "content": system_prompt}]
# 当用户消息包含这些词时，强制召回记忆（可选）
self.memory_trigger_words = memory_trigger_words or ["记得", "之前", "上次"]
async def _retrieve_context(self, user_input: str) -> str:
"""根据用户输入召回长期记忆，格式化为上下文"""
relevant = self.memory.recall(user_input, n_results=3)
if not relevant:
return ""
memory_text = "【历史记忆】\\n"
for mem in relevant:
memory_text += f"- {mem['content']}\\n"
return memory_text
async def _store_important_info(self, user_input: str, assistant_response: str):
"""判断并存储重要信息（可自定义策略）"""
# 简单策略：用户表达了偏好/事实，或对话长度超过阈值
if any(word in user_input for word in ["我喜欢", "我的名字", "记住", "我的偏好"]):
self.memory.add_memory(
content=f"用户说: {user_input} | 我回答: {assistant_response[:200]}",
metadata={"type": "preference"}
)
# 也可以定期压缩存储整个对话摘要（见进阶部分）
async def run(self, user_input: str) -> str:
# 1. 召回长期记忆
memory_context = await self._retrieve_context(user_input)
# 2. 构造增强后的消息
enhanced_messages = self.messages.copy()
if memory_context:
# 将记忆作为系统消息临时插入
memory_msg = {"role": "system", "content": memory_context}
enhanced_messages.insert(1, memory_msg) # 紧跟主系统提示
enhanced_messages.append({"role": "user", "content": user_input})
# 3. 调用大模型
response = await self.client.chat.completions.create(
model="gpt-3.5-turbo",
messages=enhanced_messages,
temperature=0.7
)
assistant_content = response.choices[0].message.content
# 4. 存储对话到短期记忆（上下文窗口）
self.messages.append({"role": "user", "content": user_input})
self.messages.append({"role": "assistant", "content": assistant_content})
# 5. 选择性存储到长期记忆
await self._store_important_info(user_input, assistant_content)
# 限制短期记忆长度（保留最近20轮）
if len(self.messages) > 41:
self.messages = [self.messages[0]] + self.messages[-40:]
return assistant_content
```
第三步：记忆压缩与睡眠总结
长期记忆不能只是原始对话，否则会爆炸。我们需要定期压缩——用一个后台任务把最近的对话总结成摘要，存入向量库。
```python
class MemoryCompressor:
def __init__(self, memory: LongTermMemory, llm_client: AsyncOpenAI):
self.memory = memory
self.llm = llm_client
async def summarize_conversation(self, conversation: List[Dict]) -> str:
"""调用LLM把一段对话总结成简洁的段落"""
if not conversation:
return ""
# 构造对话文本
text = "\\n".join([f"{m['role']}: {m['content']}" for m in conversation])
prompt = f"请用两三句话总结以下对话的关键信息（用户偏好、事实、约定等）：\\n{text}"
response = await self.llm.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": prompt}],
temperature=0.3
)
return response.choices[0].message.content
async def compress_and_store(self, agent: 'MemorableAgent', session_id: str):
"""对当前短期记忆（除系统提示外）进行总结并存储"""
# 提取用户和助手的对话（跳过系统消息）
conversation_pairs = agent.messages[1:] # 第一条是system prompt
if len(conversation_pairs) > 10: # 只总结有意义的对话
summary = await self.summarize_conversation(conversation_pairs)
if summary:
self.memory.add_memory(
content=f"[会话{session_id}总结] {summary}",
metadata={"type": "summary", "session_id": session_id}
)
# 可选：清除短期记忆中的旧对话，只保留最近几轮
agent.messages = [agent.messages[0]] + agent.messages[-6:]
```
第四步：完整演示
```python
async def demo():
# 初始化记忆
memory = LongTermMemory("user_preferences")
# 创建带记忆的智能体
agent = MemorableAgent(
system_prompt="你是一个贴心的助手，能记住用户的偏好和历史对话。",
memory=memory
)
# 第一轮对话
print("=== 第一轮 ===")
res1 = await agent.
