Menu
Article mem0

mem0学习笔记4：Mem0 架构深度解析

Gloary Lei

• 8 min read

在之前的文章中，我们专注于 使用 Mem0。现在，让我们看看它是 如何工作 的。

作为工程师，我们知道"记忆"不仅仅是一个魔法盒子。它是一个具有特定权衡的分布式系统。Mem0 做出了一些非常有主见的架构选择，使其有别于标准的 RAG 管道。

架构背后的"为什么"

在看图表之前，让我们先理解问题所在。

向量数据库的局限性

向量数据库（如 Pinecone 或 Qdrant）在 相似性搜索 方面非常出色。

  • 查询："我想喝热饮。"
  • 结果："咖啡"、"茶"、"热巧克力"。

它们的工作原理是将文本转换为高维向量并找到"最近的邻居"。

但它们在结构化方面表现糟糕。

  • 查询："谁是 Alice 的兄弟？"
  • 向量结果：它可能会返回"Alice"、"兄弟"或"家庭"，但它本质上并不理解这种 关系。它只是匹配关键词或语义。

图数据库的优势

图数据库（如 Neo4j）则相反。它们在模糊搜索方面表现糟糕，但在 结构化 方面非常出色。

  • 数据：(Alice)-[:SISTER_OF]->(Bob)
  • 查询："谁是 Alice 的兄弟？"
  • 结果："Bob"。精确、确定且无幻觉。

解决方案：混合存储策略

Mem0 的"核心秘诀"在于它 同时使用两者。

当你向 Mem0 发送消息时，它不仅仅是将其转储到向量存储中。它将数据分为三层：

  1. 向量存储 (Vector Store)
    • 角色：捕获"氛围"和语义含义。
    • 支持的提供商：Qdrant, Chroma, PGVector, Milvus, Pinecone, Weaviate, FAISS, MongoDB, Redis, Elasticsearch 等 18+ 种。
    • 用例："用户通常喜欢什么？"
  2. 图存储 (Graph Store)
    • 角色：捕获实体和关系。
    • 支持的提供商：Neo4j (默认), Memgraph, Kuzu, Neptune。
    • 用例："用户公司的名称是什么？"
  3. SQL / SQLite
    • 角色：历史记录和元数据的真实来源。
    • 用例："显示昨天创建的所有记忆。"

系统架构

Mem0 基于模块化的 工厂模式 (Factory Pattern) 构建。让我们看看实际的代码实现。

组件初始化流程

当你创建 Memory 实例时，发生了什么？

# mem0/memory/main.py (简化版)
class Memory(MemoryBase):
    def __init__(self, config: MemoryConfig = MemoryConfig()):
        self.config = config
        
        # 1. 初始化 Embedder
        self.embedding_model = EmbedderFactory.create(
            self.config.embedder.provider,
            self.config.embedder.config,
            self.config.vector_store.config,
        )
        
        # 2. 初始化 Vector Store
        self.vector_store = VectorStoreFactory.create(
            self.config.vector_store.provider, 
            self.config.vector_store.config
        )
        
        # 3. 初始化 LLM
        self.llm = LlmFactory.create(
            self.config.llm.provider, 
            self.config.llm.config
        )
        
        # 4. 初始化历史数据库
        self.db = SQLiteManager(self.config.history_db_path)
        
        # 5. (可选) 初始化 Reranker
        self.reranker = None
        if config.reranker:
            self.reranker = RerankerFactory.create(
                config.reranker.provider, 
                config.reranker.config
            )
        
        # 6. (可选) 初始化 Graph Store
        self.enable_graph = False
        if hasattr(config, "graph_store") and config.graph_store:
            self.graph = GraphStoreFactory.create(
                config.graph_store.provider,
                config.graph_store.config
            )
            self.enable_graph = True

工厂模式的实际实现

让我们深入看看 LlmFactory 是如何实现的。

# mem0/utils/factory.py
class LlmFactory:
    # 提供商映射表
    provider_to_class = {
        "openai": ("mem0.llms.openai.OpenAILLM", OpenAIConfig),
        "anthropic": ("mem0.llms.anthropic.AnthropicLLM", AnthropicConfig),
        "azure_openai": ("mem0.llms.azure_openai.AzureOpenAILLM", AzureOpenAIConfig),
        "ollama": ("mem0.llms.ollama.OllamaLLM", OllamaConfig),
        "groq": ("mem0.llms.groq.GroqLLM", BaseLlmConfig),
        "gemini": ("mem0.llms.gemini.GeminiLLM", BaseLlmConfig),
        # ... 还有 13+ 种 LLM 提供商
    }
    
    @classmethod
    def create(cls, provider_name: str, config: Optional[Union[BaseLlmConfig, Dict]] = None, **kwargs):
        if provider_name not in cls.provider_to_class:
            raise ValueError(f"Unsupported Llm provider: {provider_name}")
        
        class_type, config_class = cls.provider_to_class[provider_name]
        llm_class = load_class(class_type)  # 动态加载类
        
        # 配置处理逻辑
        if config is None:
            config = config_class(**kwargs)
        elif isinstance(config, dict):
            config = config_class(**config, **kwargs)
        
        return llm_class(config)

为什么这很重要？

  1. 解耦：添加新的 LLM 提供商不需要修改核心代码，只需注册到工厂。
  2. 可测试性：可以轻松 mock 任何组件进行单元测试。
  3. 灵活性：用户可以混搭组件（OpenAI embedding + Anthropic LLM + Qdrant vector store）。

支持的组件生态

Mem0 当前支持：

LLM 提供商 (13+)

  • OpenAI, Anthropic, Azure OpenAI
  • Ollama, LMStudio (本地部署)
  • Groq, Together (推理服务)
  • Gemini, AWS Bedrock
  • DeepSeek, xAI, LiteLLM

向量存储 (18+)

  • 云服务：Pinecone, Weaviate, Qdrant Cloud
  • 开源：Chroma, Milvus, FAISS
  • 数据库集成：PGVector, MongoDB, Redis, Elasticsearch
  • 云厂商：Azure AI Search, Vertex AI

Embedder (11+)

  • OpenAI, Azure OpenAI
  • Hugging Face, FastEmbed
  • Google (Gemini, Vertex AI)
  • Together, Ollama

重排序器 (5+)

  • Cohere Reranker
  • Sentence Transformer
  • LLM-based Reranker
  • Zero Entropy Reranker
  • Hugging Face Reranker

新特性：Reranker 集成

Mem0 最近加入了 Reranker 支持，这是一个重要的性能优化。

Reranker 的实际效果：对比示例

让我们通过一个实际例子来理解 Reranker 的价值。

场景：智能体存储了以下记忆：

memories = [
    "用户喜欢打篮球",
    "用户最喜欢的篮球队是湖人队",
    "用户喜欢看篮球比赛",
    "用户喜欢在周末踢足球",
    "用户的篮球鞋是 Nike 品牌",
]

查询："用户最喜欢哪支篮球队？"

仅使用向量搜索（无 Reranker）

向量搜索基于语义相似度返回前 5 个结果：

results_without_reranker = [
    {"memory": "用户喜欢打篮球", "score": 0.82},
    {"memory": "用户喜欢看篮球比赛", "score": 0.79},
    {"memory": "用户最喜欢的篮球队是湖人队", "score": 0.77},  # 最相关的结果排在第3
    {"memory": "用户的篮球鞋是 Nike 品牌", "score": 0.74},
    {"memory": "用户喜欢在周末踢足球", "score": 0.68},
]

问题：虽然"用户最喜欢的篮球队是湖人队"是最直接的答案，但它在向量搜索中只排第 3。这是因为向量搜索关注的是语义相似度，而"喜欢打篮球"和查询的重叠词更多。

使用 Reranker（Cohere Rerank）

Reranker 会重新评估结果的相关性：

# 配置 Reranker
config = MemoryConfig(
    reranker={
        "provider": "cohere",
        "config": {
            "model": "rerank-english-v3.0",
            "top_n": 3  # 只返回最相关的 3 条
        }
    }
)

memory = Memory(config)
results = memory.search("用户最喜欢哪支篮球队？", user_id="alice")

Reranked 结果：

reranked_results = [
    {"memory": "用户最喜欢的篮球队是湖人队", "relevance_score": 0.95},  # ✅ 最相关的排在第1
    {"memory": "用户喜欢看篮球比赛", "relevance_score": 0.72},
    {"memory": "用户喜欢打篮球", "relevance_score": 0.68},
]

改进：

  • 最直接的答案"湖人队"被提升到第 1 位
  • 不相关的"Nike 篮球鞋"和"踢足球"被过滤掉（因为 top_n=3）
  • Relevance score 更准确地反映了答案的直接程度

代码实现对比

不使用 Reranker：

from mem0 import Memory

memory = Memory()  # 默认配置

# 搜索返回向量相似度排序的结果
results = memory.search("用户最喜欢哪支篮球队？", user_id="alice", limit=5)

# 需要 LLM 从 5 个结果中挑选最相关的
for r in results:
    print(f"{r['memory']} (score: {r.get('score', 'N/A')})")

使用 Reranker：

from mem0 import Memory
from mem0.configs.base import MemoryConfig

config = MemoryConfig(
    reranker={
        "provider": "cohere",  # 或 "sentence_transformer", "llm_reranker"
        "config": {
            "api_key": "your_cohere_key",
            "model": "rerank-english-v3.0",
            "top_n": 3
        }
    }
)

memory = Memory(config)

# 搜索返回重新排序并过滤后的结果
results = memory.search("用户最喜欢哪支篮球队？", user_id="alice")

# 第一个结果就是最相关的答案
print(f"答案: {results[0]['memory']}")  # "用户最喜欢的篮球队是湖人队"

何时使用 Reranker？

推荐场景：

  • 需要高精度检索（问答系统、知识库查询）
  • 向量搜索返回了太多弱相关结果
  • 需要减少发送给 LLM 的上下文长度

可选场景：

  • 对延迟非常敏感（Reranker 会增加 100-300ms）
  • 向量搜索已经足够准确
  • 预算受限（Cohere Reranker 需要额外 API 调用）

Reranker Factory 实现

# mem0/utils/factory.py
class RerankerFactory:
    provider_to_class = {
        "cohere": ("mem0.reranker.cohere_reranker.CohereReranker", CohereRerankerConfig),
        "sentence_transformer": ("mem0.reranker.sentence_transformer_reranker.SentenceTransformerReranker", SentenceTransformerRerankerConfig),
        "llm_reranker": ("mem0.reranker.llm_reranker.LLMReranker", LLMRerankerConfig),
        # ...
    }
    
    @classmethod
    def create(cls, provider_name: str, config: Optional[Union[BaseRerankerConfig, Dict]] = None, **kwargs):
        if provider_name not in cls.provider_to_class:
            raise ValueError(f"Unsupported reranker provider: {provider_name}")
        
        class_path, config_class = cls.provider_to_class[provider_name]
        reranker_class = load_class(class_path)
        
        if config is None:
            config = config_class(**kwargs)
        elif isinstance(config, dict):
            config = config_class(**config, **kwargs)
        
        return reranker_class(config)

图存储架构

图存储不是必需的，但强烈推荐用于需要精确关系建模的场景。

图工具 (Graph Tools)

Mem0 使用 LLM 的 Function Calling 来决定如何操作图。

# mem0/graphs/tools.py
ADD_MEMORY_TOOL_GRAPH = {
    "type": "function",
    "function": {
        "name": "add_graph_memory",
        "description": "Add a new graph memory to the knowledge graph.",
        "parameters": {
            "type": "object",
            "properties": {
                "source": {"type": "string"},
                "destination": {"type": "string"},
                "relationship": {"type": "string"},
                "source_type": {"type": "string"},
                "destination_type": {"type": "string"},
            },
            "required": ["source", "destination", "relationship", "source_type", "destination_type"],
        },
    },
}

UPDATE_MEMORY_TOOL_GRAPH = {
    "type": "function",
    "function": {
        "name": "update_graph_memory",
        "description": "Update the relationship of an existing graph memory.",
        # ...
    },
}

DELETE_MEMORY_TOOL_GRAPH = {
    "type": "function",
    "function": {
        "name": "delete_graph_memory",
        "description": "Delete a relationship between two nodes.",
        # ...
    },
}

LLM 会分析对话内容，决定调用哪个工具：

  • "Alice 是 Bob 的姐姐" → add_graph_memory(source="Alice", relationship="SISTER_OF", destination="Bob")
  • "Alice 现在是 Bob 的同事" → update_graph_memory(source="Alice", relationship="COLLEAGUE_OF", destination="Bob")

配置灵活性

Mem0 的配置系统非常灵活：

from mem0 import Memory
from mem0.configs.base import MemoryConfig

# 最小配置（使用默认值）
memory = Memory()

# 自定义配置
config = MemoryConfig(
    llm={
        "provider": "anthropic",
        "config": {
            "model": "claude-3-5-sonnet-20241022",
            "temperature": 0,
            "api_key": "your_key"
        }
    },
    vector_store={
        "provider": "qdrant",
        "config": {
            "collection_name": "my_memories",
            "host": "localhost",
            "port": 6333
        }
    },
    embedder={
        "provider": "openai",
        "config": {
            "model": "text-embedding-3-large"
        }
    },
    graph_store={
        "provider": "neo4j",
        "config": {
            "url": "bolt://localhost:7687",
            "username": "neo4j",
            "password": "password"
        }
    },
    reranker={
        "provider": "cohere",
        "config": {
            "model": "rerank-english-v3.0",
            "top_n": 5
        }
    }
)

memory = Memory(config)

下一步是什么？

我们已经看到了高层架构和工厂模式。现在让我们放大到微观层面。

在 下一篇文章 中，我们将追踪单个 add() 调用的生命周期。我们将分析 记忆处理管道——LLM 如何决定保留什么、丢弃什么，以及它如何防止重复和幻觉。

Share