为什么记忆是 Agent 的第一道坎
一个没有记忆的 Agent 就像金鱼——每次对话都从零开始。你告诉它"我不吃辣"，下一轮它依然推荐重庆火锅。
Agent 的记忆需求远比聊天机器人复杂：
- 上下文连续性：多轮对话中记住用户偏好和历史决策
- 知识持久化：用户的饮食习惯、预算偏好不该每次重复
- 信息检索：从海量历史交互中快速找到相关片段
- 自我进化：从错误中学习，逐步优化行为
本文将系统性地拆解 Agent 记忆的完整架构。
三层记忆模型
借鉴认知心理学的人类记忆模型，AI Agent 的记忆同样可以分为三层：
code
┌─────────────────────────────────────────────────────┐
│ 工作记忆 (Working Memory) │
│ · 当前对话上下文 │
│ · LLM 的 context window 直接承载 │
│ · 容量: 4K-128K tokens │
│ · 生命周期: 单次推理 │
├─────────────────────────────────────────────────────┤
│ 短期记忆 (Short-term Memory) │
│ · 当前会话的完整历史 │
│ · Redis / 内存缓存 │
│ · 容量: 无硬限制，但需裁剪 │
│ · 生命周期: 单次会话 │
├─────────────────────────────────────────────────────┤
│ 长期记忆 (Long-term Memory) │
│ · 跨会话的用户画像 + 知识库 │
│ · PostgreSQL + pgvector / Chroma / Milvus │
│ · 容量: 无限（向量检索保证效率） │
│ · 生命周期: 永久 │
└─────────────────────────────────────────────────────┘
短期记忆：滑动窗口 vs 摘要压缩
LLM 的 context window 有限，不能无限制塞入历史对话。管理短期记忆有两种主流方案：
方案一：滑动窗口
最简单直接——只保留最近 N 轮对话。
python
class SlidingWindowMemory:
def __init__(self, max_turns: int = 10):
self.messages = []
self.max_turns = max_turns
def add(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
# 超过上限时丢掉最早的对话对
while len(self.messages) > self.max_turns * 2:
self.messages.pop(0)
def get_context(self) -> list:
return self.messages
优点：实现简单，延迟低。 缺点：超过窗口的对话永远丢失。用户说了"我不吃辣"但那是第 11 轮之前的事——Agent 就忘了。
方案二：摘要压缩
定期将早期对话压缩为摘要，保留关键信息。
python
class SummaryMemory:
def __init__(self, llm, max_tokens: int = 4000):
self.llm = llm
self.messages = []
self.summary = ""
self.max_tokens = max_tokens
def add(self, role: str, content: str):
self.messages.append({"role": role, "content": content,
"timestamp": time.time()})
self._maybe_compress()
def _maybe_compress(self):
total = self._count_tokens()
if total > self.max_tokens:
# 摘最早的 4 轮对话
old = self.messages[:8]
new_summary = self.llm.summarize(old)
self.summary = self._merge_summaries(self.summary, new_summary)
self.messages = self.messages[8:]
def get_context(self):
ctx = []
if self.summary:
ctx.append({"role": "system",
"content": f"历史对话摘要: {self.summary}"})
return ctx + self.messages
方案三（推荐）：滑动窗口 + 摘要混合
实际生产中最常用的是混合方案：保留最近 N 轮的完整对话 + 更早对话的摘要。这样既有短期的精确上下文，又不丢失长期信息。
python
class HybridMemory:
def __init__(self, llm, recent_turns: int = 5):
self.llm = llm
self.recent = [] # 最近 N 轮完整对话
self.archived = [] # 更早的对话
self.summary = "" # 归档对话的摘要
self.recent_turns = recent_turns
def add(self, role, content):
self.recent.append({"role": role, "content": content})
if len(self.recent) > self.recent_turns * 2:
# 移出最早的对话对到归档区
overflow = self.recent[:2]
self.archived.extend(overflow)
self.recent = self.recent[2:]
self._update_summary()
def _update_summary(self):
if len(self.archived) >= 4:
new = self.llm.summarize(self.archived[-4:])
self.summary = self._merge(self.summary, new)
def get_context(self):
ctx = []
if self.summary:
ctx.append({"role": "system",
"content": f"历史摘要: {self.summary}"})
return ctx + self.recent
长期记忆：向量语义检索
短期记忆解决"这次对话记住什么"，长期记忆解决"下次对话还能想起什么"。
核心组件
python
from pgvector.sqlalchemy import Vector
import sqlalchemy as sa
# 记忆表结构
class Memory(sa.orm.DeclarativeBase):
__tablename__ = "agent_memories"
id: int
user_id: str # 用户标识
content: str # 记忆内容（自然语言）
embedding: Vector(1024) # 语义向量
memory_type: str # "preference" | "fact" | "event"
importance: float # 重要性权重 0-1
created_at: datetime
last_accessed: datetime
记忆写入策略
不是所有对话都值得记住。需要智能筛选：
python
class MemoryWriter:
def __init__(self, llm, embedder, db):
self.llm = llm
self.embedder = embedder
self.db = db
async def extract_memories(self, conversation: list) -> list:
"""从对话中提取值得持久化的记忆"""
prompt = f"""
从以下对话中提取值得长期记住的信息。
只提取事实和偏好，不要提取寒暄。
每条记忆用一句话描述。
对话: {self._format(conversation)}
输出 JSON:
[
{{"content": "用户偏好高铁出行", "type": "preference", "importance": 0.8}},
{{"content": "用户计划6月去西安", "type": "event", "importance": 0.6}}
]
"""
memories = self.llm.extract(prompt)
return memories
async def upsert(self, user_id: str, memory: dict):
embedding = self.embedder.encode(memory["content"])
# 检查是否已存在相似记忆
similar = await self.db.search_similar(
user_id, embedding, threshold=0.85
)
if similar:
# 更新已有记忆
await self.db.update(similar[0].id, {
"content": memory["content"],
"importance": max(similar[0].importance, memory["importance"]),
"last_accessed": datetime.now()
})
else:
# 写入新记忆
await self.db.insert({
"user_id": user_id,
"content": memory["content"],
"embedding": embedding,
"memory_type": memory["type"],
"importance": memory["importance"]
})
记忆检索
关键：不是返回所有记忆，而是返回当前对话最相关的几条。
python
class MemoryRetriever:
def __init__(self, embedder, db):
self.embedder = embedder
self.db = db
async def retrieve(
self, user_id: str, query: str,
top_k: int = 5, freshness_days: int = 30
) -> list:
"""语义检索 + 时效性 + 重要性加权排序"""
query_vec = self.embedder.encode(query)
results = await self.db.search(
user_id, query_vec, top_k * 2 # 多取一些做重排序
)
# 加权排序
scored = []
for m in results:
# 时效性衰减
days_old = (datetime.now() - m.last_accessed).days
freshness = max(0, 1 - days_old / freshness_days)
# 最终得分 = 语义相似度 × 0.4 + 时效性 × 0.3 + 重要性 × 0.3
score = (
m.similarity * 0.4 +
freshness * 0.3 +
m.importance * 0.3
)
scored.append((m, score))
scored.sort(key=lambda x: x[1], reverse=True)
return [m for m, _ in scored[:top_k]]
记忆遗忘与冲突解决
遗忘机制
长期记忆不能无限增长。需要被动和主动遗忘：
python
class MemoryPruner:
async def prune(self, user_id: str):
# 1. 被动遗忘：低重要性 + 长时间未访问
await self.db.delete(
"importance < 0.3 AND last_accessed < ?",
datetime.now() - timedelta(days=90)
)
# 2. 主动去重：合并高度相似的记忆
duplicates = await self.db.find_duplicates(
user_id, threshold=0.95
)
for group in duplicates:
# 保留最完整的一条，合并重要性
best = max(group, key=lambda m: len(m.content))
best.importance = max(m.importance for m in group)
await self.db.update(best.id, best)
for m in group:
if m.id != best.id:
await self.db.delete(m.id)
冲突解决
当新旧记忆矛盾时（如"用户爱吃辣" vs "用户最近不吃辣了"），需要冲突检测：
python
async def resolve_conflict(self, user_id, new_memory):
# 检索语义相反的记忆
opposite = await self.search_opposite(user_id, new_memory)
if opposite:
# 用 LLM 判断是真矛盾还是场景变化
judgment = await self.llm.judge(
f"旧记忆: {opposite.content}\n新记忆: {new_memory.content}\n"
f"这是偏好改变还是场景不同？"
)
if judgment == "preference_change":
# 偏好改变了，删除旧记忆
await self.db.delete(opposite.id)
else:
# 场景不同，两条都保留但添加场景标签
await self.db.add_context_tag(opposite.id, "past")
await self.db.add_context_tag(new_memory.id, "recent")
主流框架对比
| 框架 | 短期记忆 | 长期记忆 | 亮点 |
|---|---|---|---|
| LangChain Memory | ConversationBuffer/Summary | VectorStore-backed | 生态最全，插件多 |
| Mem0 | 自动管理 | pgvector/Qdrant | 记忆冲突解决 + 去重 |
| MemGPT/Letta | 操作系统式管理 | 虚拟上下文 | 自编辑记忆，灵感来自 OS |
| CrewAI Memory | 对话级 | SQLite + Embedding | 多 Agent 共享记忆 |
| Zep | 自动摘要 | 知识图谱 + 向量 | 时序感知 + 实体提取 |
选型建议
code
快速原型 → LangChain Memory（最简）
生产级应用 → Mem0（去重 + 冲突解决）
学术研究 → MemGPT（虚拟上下文超长记忆）
多 Agent → CrewAI（共享记忆总线）
总结：记忆设计的 5 条金律
- 三层分离：工作/短期/长期各司其职，不要用 context window 硬抗
- 摘要优先：早期对话应被摘要，而非丢弃
- 语义检索：长期记忆必须支持向量检索，关键词匹配远远不够
- 遗忘机制：不重要 + 不活跃的记忆应自动淘汰
- 冲突解决：偏好变化应被识别，旧记忆应被更新而非共存
Agent 的记忆系统是决定用户体验上限的核心——对话越久越聪明，而不是越久越健忘。