为什么Agent需要记忆?
用过ChatGPT的人都知道这个痛点:每次对话都是"失忆"的。你跟它聊了一个小时的项目方案,下次新开对话,它什么都不记得。
真正有用的Agent需要三种记忆:
| 记忆类型 |
类比 |
作用 |
示例 |
| 短期记忆 |
工作台上的便签 |
当前对话上下文 |
“刚才说的那个方案” |
| 长期记忆 |
笔记本 |
用户偏好和历史 |
“这个用户喜欢Python” |
| 工作记忆 |
临时计算器 |
当前任务的中间状态 |
“已经分析了3个文件” |
这篇文章用Mem0(开源记忆层)+ 向量数据库,为Agent构建完整的记忆系统。
技术选型
1
2
3
4
5
6
7
8
9
|
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Agent │────►│ Mem0 │────►│ 向量数据库 │
│ (推理引擎) │◄────│ (记忆层) │◄────│ (持久存储) │
└──────────────┘ └──────────────┘ └──────────────┘
│ │
│ ┌─────▼─────┐
│ │ LLM 提取 │
│ │ 关键记忆 │
└──────────────┘ │
|
- Mem0:自动从对话中提取关键信息(“用户是Python开发者”、“不喜欢抽象建议”)
- ChromaDB:轻量级向量数据库,本地运行无需外部服务
- OpenAI Embeddings:将记忆转为向量,支持语义搜索
安装
1
|
pip install mem0ai chromadb
|
基础版:5分钟搭建记忆系统
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
|
from mem0 import Memory
# 初始化Mem0
memory = Memory()
def chat_with_memory(user_id: str, message: str) -> str:
"""带记忆的对话"""
# 1. 搜索相关记忆
relevant_memories = memory.search(query=message, user_id=user_id, limit=3)
# 2. 构建记忆上下文
memory_context = ""
if relevant_memories.get("results"):
memory_items = [m["memory"] for m in relevant_memories["results"]]
memory_context = f"\n\n用户相关记忆:\n" + "\n".join(f"- {m}" for m in memory_items)
# 3. 调用LLM(带记忆上下文)
from openai import OpenAI
client = OpenAI()
messages = [
{"role": "system", "content": f"你是一个有记忆的AI助手。{memory_context}"},
{"role": "user", "content": message}
]
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
temperature=0.7
)
assistant_reply = response.choices[0].message.content
# 4. 自动提取并存储新记忆
memory.add(
[
{"role": "user", "content": message},
{"role": "assistant", "content": assistant_reply}
],
user_id=user_id
)
return assistant_reply
# 使用示例
if __name__ == "__main__":
user_id = "user_001"
# 第一次对话
print(chat_with_memory(user_id, "我是做Java后端的,最近在研究微服务架构"))
# Agent记住: 用户是Java后端开发者,对微服务感兴趣
# 第二次对话(新会话,但记忆还在)
print(chat_with_memory(user_id, "推荐一些适合我的技术书"))
# Agent会推荐Java和微服务相关的书
|
进阶:使用本地向量数据库
生产环境建议用ChromaDB本地存储,避免数据泄露:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
import chromadb
from mem0 import Memory
from mem0.configs.base import MemoryConfig
# ChromaDB持久化存储
chroma_client = chromadb.PersistentClient(path="./memory_db")
# 配置Mem0使用ChromaDB
config = {
"version": "v1.1",
"embedder": {
"provider": "openai",
"config": {
"model": "text-embedding-3-small",
"api_key": "your-key"
}
},
"vector_store": {
"provider": "chroma",
"config": {
"collection_name": "agent_memory",
"path": "./memory_db"
}
}
}
memory = Memory.from_config(config)
|
实战:个性化技术助手
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
|
class PersonalTechAssistant:
"""带持久记忆的个性化技术助手"""
def __init__(self, user_id: str):
self.user_id = user_id
self.memory = Memory()
self.client = OpenAI()
def _build_context(self) -> str:
"""从记忆中构建个性化上下文"""
# 搜索用户偏好
prefs = self.memory.search("用户偏好 技术栈 工作", user_id=self.user_id, limit=5)
# 搜索技术历史
tech_history = self.memory.search("技术问题 代码 错误", user_id=self.user_id, limit=5)
context_parts = []
if prefs.get("results"):
items = [m["memory"] for m in prefs["results"]]
context_parts.append("用户画像:\n" + "\n".join(f" - {i}" for i in items))
if tech_history.get("results"):
items = [m["memory"] for m in tech_history["results"]]
context_parts.append("技术历史:\n" + "\n".join(f" - {i}" for i in items))
return "\n\n".join(context_parts) if context_parts else "暂无用户记忆"
def chat(self, message: str) -> str:
"""带记忆的对话"""
context = self._build_context()
system_prompt = f"""你是一个资深技术顾问。基于以下用户信息提供个性化建议:
{context}
原则:
1. 基于用户的技术栈和经验给出针对性建议
2. 引用用户之前遇到的问题,避免重复建议
3. 如果用户提到新技术,记录到记忆中"""
# 获取最近对话(短期记忆)
recent = self.memory.search(
query=message,
user_id=self.user_id,
limit=5
)
messages = [{"role": "system", "content": system_prompt}]
# 注入相关记忆作为对话历史
if recent.get("results"):
for m in recent["results"][-3:]:
messages.append({"role": "user", "content": m["memory"]})
messages.append({"role": "user", "content": message})
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
temperature=0.7
)
reply = response.choices[0].message.content
# 存储新记忆
self.memory.add(
[
{"role": "user", "content": message},
{"role": "assistant", "content": reply}
],
user_id=self.user_id
)
return reply
def get_memory_stats(self) -> dict:
"""查看记忆统计"""
all_memories = self.memory.get_all(user_id=self.user_id)
count = len(all_memories.get("results", []))
categories = {}
for m in all_memories.get("results", []):
mem_text = m.get("memory", "")
# 简单分类
if any(kw in mem_text for kw in ["Python", "Java", "Go", "技术栈"]):
categories["技术偏好"] = categories.get("技术偏好", 0) + 1
elif any(kw in mem_text for kw in ["喜欢", "不喜欢", "偏好"]):
categories["个人偏好"] = categories.get("个人偏好", 0) + 1
elif any(kw in mem_text for kw in ["错误", "问题", "bug"]):
categories["问题历史"] = categories.get("问题历史", 0) + 1
else:
categories["其他"] = categories.get("其他", 0) + 1
return {"total_memories": count, "categories": categories}
# 使用
assistant = PersonalTechAssistant(user_id="dev_001")
# 多轮对话
print(assistant.chat("我是做Android开发的,用Kotlin,最近在搞Compose"))
print(assistant.chat("推荐一个好用的网络库"))
print(assistant.chat("我之前用Retrofit遇到过JSON解析的问题"))
print(assistant.chat("最近有什么值得关注的新技术?"))
# 查看记忆统计
print(assistant.get_memory_stats())
# {'total_memories': 8, 'categories': {'技术偏好': 3, '问题历史': 2, ...}}
|
记忆管理API
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
# 查看所有记忆
all_memories = memory.get_all(user_id="user_001")
for m in all_memories["results"]:
print(f"[{m['id']}] {m['memory']}")
# 更新记忆
memory.update(memory_id="mem_xxx", data="用户已从Java转到Go")
# 删除记忆
memory.delete(memory_id="mem_xxx")
# 删除用户所有记忆
memory.delete_all(user_id="user_001")
# 搜索(语义匹配)
results = memory.search(
query="数据库相关的问题",
user_id="user_001",
limit=5
)
|
踩坑记录
1. 记忆太多导致上下文爆炸
Mem0会自动提取记忆,但积累几个月后可能有上千条。解决方案:
- 限制搜索返回数量(
limit=3-5)
- 定期清理过期记忆(Mem0支持
created_at过滤)
- 用相关性排序,只注入最相关的
2. 敏感信息泄露
Agent可能把用户的密码、API Key等存入记忆。建议:
- 在
memory.add()前过滤敏感信息
- 定期审查记忆内容
- 生产环境用加密存储
3. 记忆冲突
用户说"我现在用Go了",但旧记忆"用户用Java"还在。Mem0会自动更新相关记忆,但偶尔会有遗漏。建议:
- 用
memory.update()主动覆盖
- 搜索时按时间排序,优先取最新的
4. Embedding成本
每次存储和搜索都会调用Embedding API。高频对话场景建议:
- 批量处理记忆(积累N条后一次性Embedding)
- 用本地Embedding模型(如sentence-transformers)替代OpenAI
总结
记忆是Agent从"工具"变成"助手"的关键。没有记忆的Agent每次对话都像初次见面;有记忆的Agent能真正理解你、记住你、帮助你。
三层记忆架构:
- 短期记忆:当前对话上下文(用对话历史实现)
- 长期记忆:用户偏好和历史(用Mem0+向量数据库实现)
- 工作记忆:任务执行的中间状态(用结构化数据实现)
记住:好的记忆系统不是记住一切,而是在正确的时间想起正确的信息。