Contents

为你的AI Agent添加持久记忆:让智能体真正记住用户

为什么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+向量数据库实现)
  • 工作记忆:任务执行的中间状态(用结构化数据实现)

记住:好的记忆系统不是记住一切,而是在正确的时间想起正确的信息