人类的智能很大程度依赖于记忆。AI Agent 同样需要一个完整的记忆系统来保持上下文、积累知识、持续学习。
为什么 AI Agent 需要记忆？
大模型的上下文窗口是有限的（通常 128K-1M tokens）。当对话变长、任务变复杂时，AI 会面临：
- 上下文溢出：塞不进去
- 重要信息遗忘：早期关键信息被稀释
- 无法跨会话学习：每次对话都是从零开始
记忆系统的核心目标：
- 扩展有效上下文
- 选择性记住重要信息
- 跨会话积累知识
三层记忆架构
┌─────────────────────────────────────────────────────┐
│ 记忆系统分层 │
├─────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ 上下文窗口内的信息 │
│ │ 短期记忆 │ 当前对话、工作内存 │
│ │ Short-term │ 容量有限，随对话结束而消失 │
│ └─────────────┘ │
│ ↓ 摘要/压缩 │
│ ┌─────────────┐ 持久化存储，跨会话保持 │
│ │ 长期记忆 │ 用户偏好、长期目标、常识 │
│ │ Long-term │ 存储在外部数据库 │
│ └─────────────┘ │
│ ↑ 检索 │
│ ┌─────────────┐ 高效相似度检索 │
│ │ 向量记忆 │ 用 Embedding 存储和搜索 │
│ │ Vector Mem │ RAG 的核心基础设施 │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────┘
短期记忆：工作上下文
短期记忆是模型当前上下文窗口中的信息。
核心机制：滑动窗口 + 摘要
class ShortTermMemory:
def __init__(self, window_size=10):
self.messages = [] # 最近的消息列表
self.window_size = window_size # 保留最近N条
def add(self, role, content):
self.messages.append({"role": role, "content": content})
# 超过窗口大小时，移除最老的或压缩
if len(self.messages) > self.window_size:
self.messages.pop(0)
def get_context(self):
return self.messages[-self.window_size:]
def summarize_old(self, llm):
"""对旧消息做摘要，释放上下文空间"""
if len(self.messages) <= 3:
return
old_messages = self.messages[:-3]
summary = llm.invoke(
f"请简要总结以下对话要点：\n{old_messages}"
)
# 保留最近3条 + 摘要
self.messages = [{"role": "system", "content": f"之前摘要：{summary}"}] \
+ self.messages[-3:]
何时触发摘要？
- 消息数量超过阈值（如 20 条）
- token 数量接近上下文窗口 80%
- 对话主题发生转换时
长期记忆：持久化知识
长期记忆存储在外部数据库，跨会话保持。
存储结构
class LongTermMemory:
def __init__(self, vector_db):
self.vector_db = vector_db # Pinecone/Milvus/Chroma
self.structured_db = {} # 结构化记忆：用户偏好等
def store(self, content, memory_type, metadata):
"""
memory_type: 'preference' | 'fact' | 'goal' | 'skill'
metadata: {'user': 'xxx', 'timestamp': '...'}
"""
# 向量化存储
embedding = embed_model.encode(content)
self.vector_db.add(
vectors=[{"id": generate_id(), "values": embedding,
"metadata": {"content": content,
"type": memory_type,
**metadata}}]
)
# 结构化索引（快速查询）
if memory_type == 'preference':
self.structured_db[metadata['user']] = content
def retrieve(self, query, top_k=5):
"""基于语义检索记忆"""
query_embedding = embed_model.encode(query)
results = self.vector_db.search(
vector=query_embedding, top_k=top_k
)
return [r['metadata']['content'] for r in results]
def update_preference(self, user, key, value):
"""更新用户偏好"""
if user not in self.structured_db:
self.structured_db[user] = {}
self.structured_db[user][key] = value
记忆类型分类
| 类型 | 说明 | 示例 | 存储方式 |
|---|---|---|---|
| 偏好 | 用户喜欢什么 | 喜欢详细解释 | KV 结构 |
| 事实 | 用户的客观信息 | 工作是程序员 | 向量 + 结构 |
| 目标 | 用户想达成什么 | 备考软考高项 | 向量 |
| 技能 | 用户会什么/学过什么 | 会 Python | 向量 |
向量记忆：语义检索引擎
向量记忆是记忆系统的”搜索引擎”——用 Embedding 实现相似度检索。
核心原理
import chromadb
class VectorMemory:
def __init__(self, collection_name="agent_memory"):
self.client = chromadb.Client()
self.collection = self.client.create_collection(
collection_name,
metadata={"hnsw:space": "cosine"} # 余弦相似度
)
def add(self, text, memory_id=None, tags=None):
embedding = embed_model.encode(text)
self.collection.add(
embeddings=[embedding.tolist()],
documents=[text],
ids=[memory_id or generate_id()],
metadatas=[{"tags": tags or []}]
)
def search(self, query, top_k=5, filter_tags=None):
query_embedding = embed_model.encode(query)
result = self.collection.query(
query_embeddings=[query_embedding.tolist()],
n_results=top_k,
where={"tags": {"$in": filter_tags}} if filter_tags else None
)
return result['documents'], result['distances']
def delete(self, memory_id):
self.collection.delete(ids=[memory_id])
检索优化策略
1. 混合检索：语义相似度 + 关键词匹配
def hybrid_search(query, top_k=10):
# 语义检索
semantic_results = vector_mem.search(query, top_k=20)
# BM25 关键词检索
keyword_results = bm25_index.search(query, top_k=20)
# RRF 融合
fused = reciprocal_rank_fusion(
[semantic_results, keyword_results],
k=60
)
return fused[:top_k]
2. 时间衰减：新记忆权重更高
def time_weighted_score(distance, timestamp, half_life_days=30):
days_old = (now - timestamp).days
decay = 0.5 ** (days_old / half_life_days)
return distance * decay
3. 重要性加权：区分普通对话与关键决定
def importance_weighted检索(query, importance_hint=None):
base_results = vector_mem.search(query, top_k=20)
# 重要事件（如用户明确说"记住xxx"）权重翻倍
for r in base_results:
if r.metadata.get('importance') == 'high':
r.score *= 0.5 # 距离变小 = 更相关
return sorted(base_results, key=lambda x: x.score)
记忆系统的完整流程
┌─────────────────────────────────────────────────────────────┐
│ AI Agent 记忆流程 │
└─────────────────────────────────────────────────────────────┘
用户输入
│
▼
┌─────────────────┐
│ 检索长期记忆 │ ◄──── 跨会话积累的相关信息
└────────┬────────┘
│ 检索结果
▼
┌─────────────────┐
│ 构建上下文 │ ◄──── 短期记忆 + 长期记忆 + 用户输入
└────────┬────────┘
│ 完整上下文
▼
┌─────────────────┐
│ LLM 生成回复 │
└────────┬────────┘
│ 回复内容
┌────┴────┐
▼ ▼
┌───────┐ ┌───────────┐
│ 评估 │ │ 更新记忆 │
│ 重要？ │ │ │
└───┬───┘ └─────┬─────┘
│ yes │
▼ ▼
重要记忆 → 长期记忆存储 ──► 向量数据库
│
└──► 结构化存储（偏好/事实）
记忆更新策略
def should_remember(content, context) -> bool:
"""判断内容是否值得记忆"""
# 1. 显式标记
if "记住" in content or "重要" in content:
return True
# 2. 重复出现
if content_similarity(content, recent_memories) > 0.9:
return True # 已经有类似记忆，更新而非新增
# 3. 任务关键信息
if contains_goal(context) and contains_fact(content):
return True
# 4. 用户偏好
if contains_preference_signal(content):
return True
return False
实战：构建完整的 Agent 记忆系统
class MemFreeAgent:
"""MemGPT 风格的 Agent 记忆系统"""
def __init__(self):
# 三个层次的记忆
self.short_term = ShortTermMemory(window_size=10)
self.long_term = LongTermMemory(vector_db=chroma_client)
self.summary = "" # 对话摘要
def think(self, user_input):
# Step 1: 检索长期记忆
relevant_memories = self.long_term.retrieve(user_input, top_k=5)
# Step 2: 构建完整上下文
context = {
"recent": self.short_term.get_context(),
"memory": relevant_memories,
"summary": self.summary,
"preferences": self.long_term.structured_db.get('preferences', {})
}
# Step 3: LLM 生成
prompt = self.build_prompt(user_input, context)
response = llm.invoke(prompt)
# Step 4: 更新记忆
self.short_term.add("user", user_input)
self.short_term.add("assistant", response)
# Step 5: 定期总结
if len(self.short_term.messages) > 15:
self._summarize_and_store()
return response
def _summarize_and_store(self):
"""总结旧对话，存入长期记忆"""
old_messages = self.short_term.messages[:-5]
summary = llm.invoke(f"总结以下对话的关键信息：\n{old_messages}")
self.long_term.store(
content=summary,
memory_type="conversation_summary",
metadata={"timestamp": now()}
)
self.short_term.messages = self.short_term.messages[-5:]
💡 小结
AI Agent 的记忆系统是三层结构：
| 层次 | 容量 | 持久性 | 速度 | 用途 |
|---|---|---|---|---|
| 短期记忆 | 窗口大小 | 对话内 | 最快 | 当前工作上下文 |
| 长期记忆 | 无限制 | 永久 | 检索+解码 | 跨会话知识积累 |
| 向量记忆 | 无限制 | 永久 | 毫秒级检索 | 语义相似匹配 |
核心设计原则：
- 按需检索：不要把所有记忆都塞进上下文
- 重要性过滤：只记值得记的
- 定期整理：摘要、压缩、遗忘
- 混合检索：语义 + 关键词 + 结构化
相关阅读：AI Agent 原理与实践