Memory 系统设计

记忆（Memory）是 Agent 从「无状态工具」进化为「智能助手」的关键。

记忆类型

短期记忆：对话管理

"""
短期记忆 - 滑动窗口 + 摘要
"""
from openai import OpenAI
from dataclasses import dataclass, field
@dataclass
class ConversationMemory:
"""对话记忆管理"""
max_messages: int = 20
max_tokens: int = 4000
messages: list[dict] = field(default_factory=list)
summary: str = ""
def add(self, role: str, content: str) -> None:
"""添加消息"""
self.messages.append({"role": role, "content": content})
# 超出限制时压缩
if len(self.messages) > self.max_messages:
self._compress()
def get_context(self) -> list[dict]:
"""获取当前上下文"""
context = []
# 加入摘要（如果有）
if self.summary:
context.append({
"role": "system",
"content": f"之前对话摘要: {self.summary}",
})
context.extend(self.messages)
return context
def _compress(self) -> None:
"""压缩历史消息为摘要"""
client = OpenAI()
# 取前半部分消息进行摘要
half = len(self.messages) // 2
to_summarize = self.messages[:half]
old_text = "\n".join(
f"{m['role']}: {m['content'][:200]}"
for m in to_summarize
)
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": "请用 2-3 句话概括以下对话的关键信息:",
},
{"role": "user", "content": old_text},
],
max_tokens=200,
)
new_summary = response.choices[0].message.content
if self.summary:
self.summary = f"{self.summary}\n{new_summary}"
else:
self.summary = new_summary
# 保留后半部分
self.messages = self.messages[half:]
print(f"记忆已压缩: {half} 条消息 → 摘要")
# 使用
memory = ConversationMemory(max_messages=10)
memory.add("user", "我想学习 Python")
memory.add("assistant", "Python 是一门很好的语言，建议从基础语法开始...")
memory.add("user", "有什么好的教程推荐吗？")
memory.add("assistant", "推荐《Python编程入门》和官方教程...")
context = memory.get_context()

长期记忆：向量存储

"""
长期记忆 - 基于向量的持久化记忆
"""
import json
import time
from pathlib import Path
class LongTermMemory:
"""长期记忆系统"""
def __init__(self, db_path: str = "./memory_db"):
self.db_path = Path(db_path)
self.db_path.mkdir(parents=True, exist_ok=True)
# 使用 ChromaDB 作为向量存储
import chromadb
self.chroma = chromadb.PersistentClient(path=str(self.db_path))
self.collection = self.chroma.get_or_create_collection("agent_memory")
self.client = OpenAI()
def remember(self, content: str, metadata: dict = None) -> str:
"""存储记忆"""
memory_id = f"mem_{int(time.time() * 1000)}"
# 获取 embedding
embedding = self._embed(content)
# 存储
self.collection.add(
ids=[memory_id],
embeddings=[embedding],
documents=[content],
metadatas=[{
"timestamp": time.time(),
"type": metadata.get("type", "general") if metadata else "general",
**(metadata or {}),
}],
)
return memory_id
def recall(self, query: str, top_k: int = 5) -> list[dict]:
"""检索相关记忆"""
query_embedding = self._embed(query)
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k,
)
memories = []
for i in range(len(results["ids"][0])):
memories.append({
"id": results["ids"][0][i],
"content": results["documents"][0][i],
"relevance": 1 - results["distances"][0][i],
"metadata": results["metadatas"][0][i],
})
return memories
def forget(self, memory_id: str) -> None:
"""删除记忆"""
self.collection.delete(ids=[memory_id])
def _embed(self, text: str) -> list[float]:
"""获取文本向量"""
response = self.client.embeddings.create(
model="text-embedding-3-small",
input=text,
)
return response.data[0].embedding
@property
def count(self) -> int:
return self.collection.count()
# 使用
ltm = LongTermMemory()
# 存储学习到的信息
ltm.remember(
"用户喜欢使用 Python 和 FastAPI 开发后端服务",
metadata={"type": "user_preference"},
)
ltm.remember(
"用户的项目是一个电商平台，使用 PostgreSQL 数据库",
metadata={"type": "project_info"},
)
# 后续对话中检索
memories = ltm.recall("用户用什么技术栈？")
for m in memories:
print(f"[{m['relevance']:.2f}] {m['content']}")

完整记忆系统

将短期和长期记忆整合到 Agent 中：

"""
完整记忆系统
"""
class MemorySystem:
"""统一记忆管理"""
def __init__(self):
self.short_term = ConversationMemory(max_messages=20)
self.long_term = LongTermMemory()
self.working: dict = {}  # 工作记忆
def process_interaction(self, user_msg: str, assistant_msg: str) -> None:
"""处理一轮交互"""
# 1. 更新短期记忆
self.short_term.add("user", user_msg)
self.short_term.add("assistant", assistant_msg)
# 2. 判断是否需要存入长期记忆
if self._should_remember(user_msg, assistant_msg):
self.long_term.remember(
f"用户: {user_msg}\n助手: {assistant_msg}",
metadata={"type": "conversation"},
)
def get_context(self, current_query: str) -> str:
"""获取与当前查询相关的完整上下文"""
parts = []
# 长期记忆中检索相关信息
memories = self.long_term.recall(current_query, top_k=3)
if memories:
parts.append("## 相关历史记忆")
for m in memories:
if m["relevance"] > 0.7:
parts.append(f"- {m['content']}")
# 工作记忆
if self.working:
parts.append("\n## 当前工作状态")
for key, value in self.working.items():
parts.append(f"- {key}: {value}")
return "\n".join(parts)
def _should_remember(self, user_msg: str, assistant_msg: str) -> bool:
"""判断是否值得长期记忆"""
# 简单规则：较长的有实质内容的交互
total_len = len(user_msg) + len(assistant_msg)
return total_len > 200
# ==================
# 集成到 Agent
# ==================
class MemoryAgent:
"""带记忆的 Agent"""
def __init__(self):
self.client = OpenAI()
self.memory = MemorySystem()
def chat(self, user_message: str) -> str:
"""对话"""
# 获取记忆上下文
memory_context = self.memory.get_context(user_message)
# 构建消息
messages = [
{
"role": "system",
"content": f"""你是一个有长期记忆的 AI 助手。
以下是你记住的信息：
{memory_context}
请利用这些记忆来更好地回答用户。""",
},
]
# 加入短期对话历史
messages.extend(self.memory.short_term.get_context())
messages.append({"role": "user", "content": user_message})
response = self.client.chat.completions.create(
model="gpt-4o",
messages=messages,
)
assistant_msg = response.choices[0].message.content
# 更新记忆
self.memory.process_interaction(user_message, assistant_msg)
return assistant_msg
# 使用
agent = MemoryAgent()
# 第一轮
print(agent.chat("我正在用 FastAPI 开发一个电商后端"))
# 第二轮 - Agent 记住了之前的内容
print(agent.chat("帮我设计一下订单模块的 API"))

记忆设计原则

本章小结

• 短期记忆使用滑动窗口 + 摘要压缩管理对话上下文
• 长期记忆使用向量数据库存储，按语义检索
• 工作记忆保存当前任务的中间状态
• 记忆系统要有选择性存储和遗忘机制
• 隐私敏感信息不应存入长期记忆

下一章：学习多 Agent 系统的架构设计。