AI Agent 记忆系统架构深度拆解：从 Mem0 四层编排到 LoCoMo 基准的 10 种方案横评——手写一个生产级长期记忆引擎

2026 年，AI Agent 的核心战场已经从「工具调用」转移到「长期记忆」。一个能跨周跨月持续记住你的 Agent，才是真正的护城河。本文从第一性原理出发，逐字节拆解 Mem0 / Mem0g / MemGPT / A-Mem / MemoryBank 等主流方案的架构设计，并附完整 Python 实现——手写一个能跑通 LoCoMo 评测的生产级长期记忆引擎。

一、为什么 Agent 记忆是 2026 的核心战场

1.1 没有记忆的 Agent 是什么体验

想象一下你和一个同事共事三个月，每次开会他都问你同一个问题——「我们的产品是什么？」「上次那个 bug 修了吗？」「你叫什么名字来着？」。你会觉得这人脑子有毛病。

但这就是当前大部分 AI Agent 的状态。每次对话都从零开始，没有跨会话的持续记忆，用户的偏好、历史决策、项目上下文全部丢失。

1.2 记忆的四个层次

从认知科学的角度，人类记忆分为四个层次，AI Agent 的记忆系统设计也遵循类似的层次结构：

┌──────────────────────────────────────────────────────┐
│  感知记忆 (Sensory Memory)                           │
│  - 实时上下文，当前对话的即时输入                      │
│  - 生命周期：单轮对话                                 │
│  - 实现：LLM 上下文窗口                              │
├──────────────────────────────────────────────────────┤
│  工作记忆 (Working Memory)                           │
│  - 任务相关数据，当前任务的中间状态                    │
│  - 生命周期：单个任务                                 │
│  - 实现：状态变量、临时缓存                           │
├──────────────────────────────────────────────────────┤
│  情景记忆 (Episodic Memory)                          │
│  - 历史交互记录，「我们上次聊过什么」                  │
│  - 生命周期：跨会话                                   │
│  - 实现：对话历史、检索增强                           │
├──────────────────────────────────────────────────────┤
│  语义记忆 (Semantic Memory)                          │
│  - 结构化知识库，「这个用户的偏好是什么」              │
│  - 生命周期：永久                                     │
│  - 实现：知识图谱、向量数据库                         │
└──────────────────────────────────────────────────────┘

1.3 三种朴素方案及其致命缺陷

方案一：无记忆（No Memory）
每次对话都是白板。用户体验崩溃，Agent 无法积累任何上下文。

方案二：全上下文塞入（Full-context）
把所有历史对话塞进 prompt。理论最准（J Score 72.9），但：

• Token 消耗：~28,000/轮
• p95 延迟：17.12 秒
• 成本系数：1.0×（基准线）

方案三：简单 RAG
把对话切片存向量库，每轮检索 top-K。便宜但：

• 时序推理能力缺失（「上个月说过什么？」无法回答）
• 多跳关联断裂（「那家公司的 CEO 是谁？」检索不到）
• J Score 仅 61.0

1.4 结构化长期记忆：2026 的工业级答案

mem0.ai 在 2026 年 5 月发布的《State of AI Agent Memory 2026》报告，第一次给出了三维 Pareto 分析——准确率 × 成本 × 延迟。结论很明确：

核心洞察：用 6 分准确率换 14 倍成本压缩 + 12 倍延迟降低，是任何理性产品决策都会做的交易。

二、记忆系统的三代架构演进

2.1 第一代：RAG-based 记忆（2023-2024）

用户输入 → 向量化 → 向量检索 top-K → 拼接上下文 → LLM 生成

问题：检索结果是片段级的，缺乏结构化关联。用户说「我上次提过的那个项目」，RAG 只能检索语义相似的片段，无法理解「那个项目」指代的是哪个具体实体。

2.2 第二代：结构化 Fact 记忆（2025）

用户输入 → LLM 抽取事实 → 存入结构化存储 → 检索时按类型路由

代表方案：Mem0。核心创新是让 LLM 从对话中自动抽取事实（而非存储原始对话），每条记忆是一句自然语言的短事实。

优势：Token 消耗极低，检索精准。
劣势：单跳事实问答强，但时序推理和多跳关联弱。

2.3 第三代：Memory Graph（2026）

用户输入 → LLM 抽取事实 + 实体 → 构建实体-记忆二部图 → 图查询 + 向量检索融合

代表方案：Mem0g。在 Mem0 基础上增加实体关系图，通过图查询增强时序推理和多跳关联能力。

LoCoMo 基准对比：

工程启示：图谱的真正价值集中在时序推理（+3.5）和多跳关联（+2.6）。如果你的 Agent 主要做单轮问答 + 偏好记忆，纯向量 Mem0 就够了。

三、Mem0 四层架构深度拆解

3.1 总体架构：四层同构

Mem0 的源码结构可以切成四层——上层的 30 个向量库、24 个 LLM、15 个 Embedding 的排列组合，全部在 Memory 这一层归一：

┌───────────────────────────────────────────────────────┐
│  接入层 (Entry/Protocol)                              │
│  REST: FastAPI + X-API-Key                           │
│  MCP: openmemory/api/, mem0-plugin/                  │
│  SDK: mem0(Py) / mem0-ts(TS)                        │
│  集成: vercel-ai-sdk / openclaw / cookbooks          │
├───────────────────────────────────────────────────────┤
│  核心编排层 (Memory)                                  │
│  Memory / AsyncMemory                                │
│  _add_to_vector_store — V3 分阶段批处理写入流水线    │
│  _search_vector_store — 三路混合检索                 │
│  storage.py SQLiteManager (history / messages)       │
│  utils.py 消息解析 / 视觉消息 / JSON 提取            │
│  configs/prompts.py 事实抽取 / 记忆更新提示词        │
├───────────────────────────────────────────────────────┤
│  Provider 可插拔层                                   │
│  LlmFactory(24) / EmbedderFactory(15)               │
│  VectorStoreFactory(30) / RerankerFactory(5)        │
├───────────────────────────────────────────────────────┤
│  底层存储 (Storage)                                   │
│  Vector: pgvector / Qdrant / Pinecone / Milvus / …  │
│  History: SQLite (~/.mem0/history.db)                │
│  LLM API: OpenAI / Anthropic / Bedrock / Gemini     │
└───────────────────────────────────────────────────────┘

关键观察：真正独特的代码 90% 在第二层（核心编排层）。向量库、LLM、Embedding 只是把各家 SDK 包一层 base.py 的抽象——Mem0 的价值几乎完全来自那两条流水线的设计。

3.2 记忆的最小形式化

Mem0 的形式化非常克制：维护一个可以被 add / search / update / delete 的事实集合 Ω，每次新来一对消息 (m_{t-1}, m_t) 就跑一次 extraction → update 两阶段流水线。

from pydantic import BaseModel
from typing import Optional, Dict

class MemoryItem(BaseModel):
    id: str
    memory: str       # 抽取出的事实文本（自然语言短事实）
    hash: Optional[str]  # md5(memory)，用于幂等
    metadata: Optional[Dict]
    score: Optional[float]
    created_at: Optional[str]
    updated_at: Optional[str]

三个设计选择值得特别关注：

对比 Zep 等方案把全量摘要缓存在每个图节点上的做法：Zep 的图构建出来大约要 60 万 Token 每会话（原始对话才 26k），Mem0 只要 7k，Mem0g 也只要 14k。差距不是来自更好的压缩算法，而是来自「什么不该存」的判断。

3.3 写入链路：add() 的九个 Phase

Mem0 的写入链路是整个系统最精妙的部分。当 infer=True（默认）时，add() 方法会执行一个完整的九阶段流水线：

# 简化的写入链路伪代码
async def add(self, messages, *, user_id=None, agent_id=None, 
              run_id=None, infer=True, **kwargs):
    """写入链路：九个 Phase"""
    
    # Phase 1: 消息预处理
    parsed_messages = parse_messages(messages)
    
    # Phase 2: 构建 filter 和 metadata
    filters, metadata = build_filters(user_id, agent_id, run_id)
    
    if infer:
        # Phase 3: LLM 事实抽取
        # 输入：最近 N 轮对话
        # 输出：自然语言事实列表
        facts = await self.llm.extract_facts(
            messages=parsed_messages,
            prompt=CUSTOM_INSTRUCTIONS or DEFAULT_FACT_EXTRACTION_PROMPT
        )
        
        # Phase 4: 事实去重与合并
        # 与已有记忆比对，合并相似事实，更新过时事实
        existing = await self.search(filters=filters)
        facts = deduplicate_and_merge(facts, existing)
        
        # Phase 5: Embedding 向量化
        embeddings = await self.embedder.embed([f.memory for f in facts])
        
        # Phase 6: 向量写入（批量）
        await self.vector_store.insert(
            vectors=embeddings,
            ids=[f.id for f in facts],
            payloads=[f.to_dict() for f in facts]
        )
        
        # Phase 7: 图操作（Mem0g 专属）
        # 实体抽取 → 二部图更新
        if self.graph_store:
            entities = extract_entities(facts)
            await self.graph_store.update_edges(entities, facts)
        
        # Phase 8: 历史记录
        await self.storage.add_history(facts, operation="ADD")
        
        # Phase 9: 返回写入的记忆
        return facts
    
    else:
        # 快速通道：直接写入，不走 LLM 抽取
        # 适用于批量导入等场景
        ...

Phase 3 的核心 Prompt 设计是关键。Mem0 使用的默认提示词会指导 LLM 执行以下操作：

1. 事实抽取：从对话中识别所有可持久化的事实
2. 更新检测：对比已有记忆，识别矛盾和过时信息
3. 冲突消解：当新事实与旧事实冲突时，保留最新的
4. 置信度评估：对抽取结果给出置信度评分

# configs/prompts.py 中的默认事实抽取提示词（简化版）
FACT_EXTRACTION_PROMPT = """
Given the following conversation and a list of existing memories, 
extract new facts from the conversation.

Rules:
1. Extract only factual information (preferences, names, dates, etc.)
2. Do not extract opinions or temporary states
3. If a fact contradicts an existing memory, update the existing one
4. If a fact is already covered by existing memories, skip it
5. Each fact should be a single, self-contained sentence

Existing memories:
{existing_memories}

Conversation:
{conversation}

Return a JSON array of new facts. Each fact should have:
- "fact": the extracted fact text
- "confidence": float between 0 and 1
"""

3.4 读取链路：search() 的三路融合

Mem0 的读取链路同样是精妙的设计。当调用 search() 时，系统会执行三路混合检索：

async def search(self, query, *, top_k=20, filters=None, 
                 threshold=0.1, rerank=False, **kwargs):
    """读取链路：三路混合检索"""
    
    # Step 1: 语义检索（向量相似度）
    query_embedding = await self.embedder.embed([query])
    semantic_results = await self.vector_store.search(
        query=query_embedding[0],
        vectors=None,  # 语义模式
        top_k=top_k * 2,  # 多取一些用于融合
        filters=filters
    )
    
    # Step 2: BM25 关键词检索
    keyword_results = await self.vector_store.keyword_search(
        query=query,
        top_k=top_k * 2,
        filters=filters
    )
    
    # Step 3: 图查询增强（Mem0g 专属）
    if self.graph_store:
        graph_results = await self.graph_store.search(
            query=query,
            top_k=top_k * 2,
            filters=filters
        )
    else:
        graph_results = []
    
    # Step 4: 三路融合排序
    # 使用 RRF (Reciprocal Rank Fusion) 或加权融合
    fused_results = reciprocal_rank_fusion([
        semantic_results,
        keyword_results,
        graph_results
    ], weights=[0.6, 0.2, 0.2])
    
    # Step 5: 阈值过滤
    filtered = [r for r in fused_results if r.score >= threshold]
    
    # Step 6: 可选重排
    if rerank and self.reranker:
        filtered = await self.reranker.rerank(query, filtered)
    
    return filtered[:top_k]

为什么三路融合比单路检索好？

• 语义检索擅长理解同义词和概念相似性（「喜欢动作电影」≈「偏好动作片」）
• BM25 检索擅长精确匹配关键词（「Python 3.11」精确匹配）
• 图查询擅长多跳关联（「那家公司的 CEO」→ 公司实体 → CEO 关系 → 人物实体）

3.5 实体-记忆二部图：Mem0g 的图到底是什么

Mem0g 的图不是传统意义上的知识图谱（Neo4j 那种三元组），而是一种轻量级的实体-记忆二部图：

┌─────────────┐          ┌─────────────┐
│  实体节点    │          │  记忆节点    │
│  (Entity)   │◄────────►│  (Memory)   │
│             │  关系边   │             │
│  - Person   │          │  - 事实文本  │
│  - Company  │          │  - 时间戳   │
│  - Concept  │          │  - 置信度   │
└─────────────┘          └─────────────┘

实体抽取的四类模式：

# 实体抽取规则
ENTITY_PATTERNS = {
    "person": {
        "patterns": [
            r"(?:我|你|他|她)\s*(?:叫|是|的名字)\s*(\w+)",
            r"(\w+)\s*(?:说|认为|觉得)",
        ],
        "type": "Person"
    },
    "company": {
        "patterns": [
            r"(?:在|去|从|到)\s*(\w+(?:公司|科技|集团|Inc|Corp))",
        ],
        "type": "Company"
    },
    "concept": {
        "patterns": [
            r"(?:学习|使用|喜欢|讨厌)\s*(\w+)",
            r"(\w+)\s*(?:框架|语言|工具|库)",
        ],
        "type": "Concept"
    },
    "temporal": {
        "patterns": [
            r"(?:上|这|下)\s*(?:周|月|年|星期)",
            r"(\d{4})\s*年\s*(\d{1,2})\s*月",
        ],
        "type": "Temporal"
    }
}

# 屏蔽词表（不抽取为实体）
STOP_ENTITIES = {
    "这个", "那个", "什么", "怎么", "为什么", "可以", "应该",
    "今天", "明天", "昨天", "现在", "然后", "所以", "但是"
}

写入链路上的图操作（Phase 7 展开）：

async def update_graph(self, facts, entities):
    """更新实体-记忆二部图"""
    for fact in facts:
        fact_id = fact.id
        
        # 对每个事实，找到关联的实体
        related_entities = entities.get(fact_id, [])
        
        for entity in related_entities:
            # 检查实体节点是否已存在
            existing_entity = await self.get_entity(entity.name, entity.type)
            
            if not existing_entity:
                # 创建新实体节点
                existing_entity = await self.create_entity(
                    name=entity.name,
                    entity_type=entity.type
                )
            
            # 创建或更新「实体 → 记忆」边
            await self.create_or_update_edge(
                source_id=existing_entity.id,
                target_id=fact_id,
                edge_type="MENTIONS",
                metadata={
                    "created_at": datetime.now().isoformat(),
                    "confidence": entity.confidence
                }
            )
    
    # 清理：删除已过期记忆对应的边
    await self.cleanup_orphan_edges()

读取链路上的图查询：

async def graph_search(self, query, top_k=20, filters=None):
    """图增强检索"""
    
    # Step 1: 从查询中提取实体
    query_entities = self.extract_entities_from_query(query)
    
    # Step 2: 在图中查找匹配的实体
    matched_entities = []
    for entity in query_entities:
        results = await self.search_entities(
            name=entity.name,
            entity_type=entity.type,
            limit=5
        )
        matched_entities.extend(results)
    
    # Step 3: 从实体出发，沿边找到关联的记忆
    related_memories = []
    for entity in matched_entities:
        # 一跳检索
        memories = await self.get_neighbors(
            entity_id=entity.id,
            edge_type="MENTIONS",
            limit=top_k
        )
        related_memories.extend(memories)
        
        # 二跳检索（多跳关联）
        for memory in memories:
            # 找到这条记忆关联的其他实体
            other_entities = await self.get_neighbors(
                entity_id=memory.id,
                edge_type="MENTIONS",
                reverse=True
            )
            # 从其他实体出发再找记忆
            for other in other_entities:
                second_hop = await self.get_neighbors(
                    entity_id=other.id,
                    edge_type="MENTIONS",
                    limit=3
                )
                related_memories.extend(second_hop)
    
    # Step 4: 去重并排序
    seen_ids = set()
    unique_memories = []
    for m in related_memories:
        if m.id not in seen_ids:
            seen_ids.add(m.id)
            unique_memories.append(m)
    
    # Step 5: 与原始 filter 结合
    if filters:
        unique_memories = [m for m in unique_memories 
                          if self.match_filters(m, filters)]
    
    return unique_memories[:top_k]

为什么不用 Neo4j？

Mem0 选择自建图存储而非使用 Neo4j，原因有三：

1. 依赖轻量化：Mem0 的目标是 pip install mem0ai 即可运行，Neo4j 需要额外的服务器
2. 写入性能：二部图的边操作可以用简单的键值对存储实现，比 Cypher 查询快一个数量级
3. 查询简化：Mem0 的图只有一层实体-记忆关系，不需要 Neo4j 的多标签、多关系类型能力

四、手写一个生产级长期记忆引擎

接下来我们从零实现一个简化版但可跑通的长期记忆引擎，包含 Mem0 的核心设计思想。

4.1 项目结构

agent_memory/
├── __init__.py
├── core/
│   ├── __init__.py
│   ├── memory.py          # 核心 Memory 类
│   ├── types.py           # 数据类型定义
│   └── prompts.py         # LLM 提示词
├── stores/
│   ├── __init__.py
│   ├── vector_store.py    # 向量存储抽象
│   ├── sqlite_store.py    # SQLite 实现
│   └── graph_store.py     # 图存储（简化版）
├── extractors/
│   ├── __init__.py
│   ├── entity_extractor.py  # 实体抽取
│   └── fact_extractor.py    # 事实抽取
└── examples/
    └── demo.py            # 演示用例

4.2 数据类型定义

# core/types.py
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
from datetime import datetime
from enum import Enum
import uuid
import hashlib


class EntityType(str, Enum):
    PERSON = "person"
    COMPANY = "company"
    CONCEPT = "concept"
    TOOL = "tool"
    TEMPORAL = "temporal"
    LOCATION = "location"


class Entity(BaseModel):
    """实体节点"""
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    name: str
    entity_type: EntityType
    metadata: Dict[str, Any] = Field(default_factory=dict)
    created_at: str = Field(default_factory=lambda: datetime.now().isoformat())


class MemoryFact(BaseModel):
    """记忆事实"""
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    memory: str                    # 自然语言事实
    hash: str = ""                 # md5(memory)，幂等
    user_id: Optional[str] = None
    agent_id: Optional[str] = None
    run_id: Optional[str] = None
    metadata: Dict[str, Any] = Field(default_factory=dict)
    confidence: float = 0.8
    score: Optional[float] = None
    created_at: str = Field(default_factory=lambda: datetime.now().isoformat())
    updated_at: str = Field(default_factory=lambda: datetime.now().isoformat())

    def model_post_init(self, __context) -> None:
        if not self.hash:
            self.hash = hashlib.md5(self.memory.encode()).hexdigest()


class MemoryHistory(BaseModel):
    """记忆变更历史"""
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    memory_id: str
    old_memory: Optional[str] = None
    new_memory: Optional[str] = None
    operation: str  # ADD / UPDATE / DELETE
    created_at: str = Field(default_factory=lambda: datetime.now().isoformat())


class SearchResult(BaseModel):
    """检索结果"""
    memory: MemoryFact
    score: float
    source: str  # semantic / keyword / graph


class MemoryConfig(BaseModel):
    """记忆系统配置"""
    vector_store_type: str = "sqlite"  # sqlite / qdrant / pgvector
    llm_provider: str = "openai"
    llm_model: str = "gpt-4o-mini"
    embedder_provider: str = "openai"
    embedder_model: str = "text-embedding-3-small"
    embedding_dim: int = 1536
    history_db_path: str = "~/.agent_memory/history.db"
    enable_graph: bool = False
    custom_instructions: Optional[str] = None
    max_facts_per_turn: int = 10
    dedup_threshold: float = 0.85  # 事实去重阈值

4.3 向量存储抽象层

# stores/vector_store.py
from abc import ABC, abstractmethod
from typing import List, Optional, Dict, Any
import numpy as np


class VectorStoreBase(ABC):
    """向量存储抽象基类"""
    
    @abstractmethod
    async def insert(self, vectors: List[List[float]], 
                     ids: List[str], 
                     payloads: List[Dict[str, Any]]) -> None:
        """批量写入"""
        pass
    
    @abstractmethod
    async def search(self, query_vector: List[float], 
                     top_k: int = 20,
                     filters: Optional[Dict] = None) -> List[Dict]:
        """语义检索"""
        pass
    
    @abstractmethod
    async def delete(self, vector_id: str) -> None:
        """删除单条"""
        pass
    
    @abstractmethod
    async def update(self, vector_id: str, 
                     vector: List[float],
                     payload: Dict[str, Any]) -> None:
        """更新单条"""
        pass
    
    @abstractmethod
    async def get(self, vector_id: str) -> Optional[Dict]:
        """按 ID 取回"""
        pass
    
    @abstractmethod
    async def keyword_search(self, query: str,
                             top_k: int = 20,
                             filters: Optional[Dict] = None) -> List[Dict]:
        """关键词检索（BM25）"""
        pass


class SQLiteVectorStore(VectorStoreBase):
    """基于 SQLite + numpy 的向量存储（适合开发和小规模场景）"""
    
    def __init__(self, db_path: str = "~/.agent_memory/vectors.db"):
        import sqlite3
        import os
        
        self.db_path = os.path.expanduser(db_path)
        os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
        
        self.conn = sqlite3.connect(self.db_path)
        self._init_db()
        
        # 内存中的向量索引（生产环境用 FAISS）
        self.vectors: Dict[str, np.ndarray] = {}
        self.payloads: Dict[str, Dict] = {}
    
    def _init_db(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS memories (
                id TEXT PRIMARY KEY,
                memory TEXT NOT NULL,
                hash TEXT,
                user_id TEXT,
                agent_id TEXT,
                run_id TEXT,
                metadata TEXT,
                confidence REAL DEFAULT 0.8,
                created_at TEXT,
                updated_at TEXT
            )
        """)
        self.conn.execute("""
            CREATE INDEX IF NOT EXISTS idx_user_id ON memories(user_id)
        """)
        self.conn.execute("""
            CREATE INDEX IF NOT EXISTS idx_agent_id ON memories(agent_id)
        """)
        self.conn.commit()
    
    async def insert(self, vectors, ids, payloads):
        import json
        
        for i, (vector, mid, payload) in enumerate(zip(vectors, ids, payloads)):
            # 存储向量（内存）
            self.vectors[mid] = np.array(vector, dtype=np.float32)
            self.payloads[mid] = payload
            
            # 存储元数据（SQLite）
            self.conn.execute("""
                INSERT OR REPLACE INTO memories 
                (id, memory, hash, user_id, agent_id, run_id, metadata, 
                 confidence, created_at, updated_at)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                mid,
                payload.get("memory", ""),
                payload.get("hash", ""),
                payload.get("user_id"),
                payload.get("agent_id"),
                payload.get("run_id"),
                json.dumps(payload.get("metadata", {})),
                payload.get("confidence", 0.8),
                payload.get("created_at"),
                payload.get("updated_at")
            ))
        self.conn.commit()
    
    async def search(self, query_vector, top_k=20, filters=None):
        """余弦相似度检索"""
        if not self.vectors:
            return []
        
        query = np.array(query_vector, dtype=np.float32)
        
        results = []
        for mid, vec in self.vectors.items():
            # 应用 filter
            if filters and not self._match_filters(self.payloads[mid], filters):
                continue
            
            # 余弦相似度
            similarity = np.dot(query, vec) / (
                np.linalg.norm(query) * np.linalg.norm(vec)
            )
            results.append({
                "id": mid,
                "score": float(similarity),
                "payload": self.payloads[mid]
            })
        
        # 排序并返回 top_k
        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:top_k]
    
    async def keyword_search(self, query, top_k=20, filters=None):
        """简单的关键词匹配（生产环境用 BM25）"""
        results = []
        query_lower = query.lower()
        
        for mid, payload in self.payloads.items():
            if filters and not self._match_filters(payload, filters):
                continue
            
            memory_text = payload.get("memory", "").lower()
            # 简单的关键词匹配分数
            score = sum(1 for word in query_lower.split() 
                       if word in memory_text)
            if score > 0:
                results.append({
                    "id": mid,
                    "score": score / len(query_lower.split()),
                    "payload": payload
                })
        
        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:top_k]
    
    async def delete(self, vector_id):
        self.vectors.pop(vector_id, None)
        self.payloads.pop(vector_id, None)
        self.conn.execute("DELETE FROM memories WHERE id = ?", (vector_id,))
        self.conn.commit()
    
    async def update(self, vector_id, vector, payload):
        await self.delete(vector_id)
        await self.insert([vector], [vector_id], [payload])
    
    async def get(self, vector_id):
        if vector_id in self.payloads:
            return {"id": vector_id, "payload": self.payloads[vector_id]}
        return None
    
    def _match_filters(self, payload, filters):
        """简单的 filter 匹配"""
        for key, value in filters.items():
            if key in ("AND", "OR", "NOT"):
                continue  # 简化版不支持复杂逻辑
            if payload.get(key) != value:
                return False
        return True

4.4 事实抽取器

# extractors/fact_extractor.py
import json
import hashlib
from typing import List, Dict, Any, Optional
from ..core.types import MemoryFact, MemoryConfig


class FactExtractor:
    """基于 LLM 的事实抽取器"""
    
    SYSTEM_PROMPT = """你是一个专业的信息抽取助手。你的任务是从对话中提取可持久化的事实信息。

规则：
1. 只提取事实性信息（偏好、姓名、日期、事件、决策等）
2. 不提取意见或临时状态
3. 如果新事实与已有事实矛盾，返回更新版本
4. 如果新事实已被已有事实覆盖，跳过
5. 每条事实应该是独立的、完整的句子
6. 返回 JSON 数组格式

输出格式：
[
    {
        "fact": "提取的事实文本",
        "confidence": 0.9,
        "operation": "add"  // add / update / skip
    }
]"""
    
    def __init__(self, config: MemoryConfig):
        self.config = config
        self.llm_client = None  # 延迟初始化
    
    async def _get_llm_client(self):
        if self.llm_client is None:
            # 根据配置初始化 LLM 客户端
            if self.config.llm_provider == "openai":
                from openai import AsyncOpenAI
                self.llm_client = AsyncOpenAI()
            # 可扩展其他 provider
        return self.llm_client
    
    async def extract_facts(
        self,
        messages: List[Dict[str, str]],
        existing_memories: List[MemoryFact] = None
    ) -> List[MemoryFact]:
        """从对话中抽取事实"""
        
        # 构建对话文本
        conversation = "\n".join([
            f"{m.get('role', 'user')}: {m.get('content', '')}"
            for m in messages
        ])
        
        # 构建已有记忆文本
        existing_text = ""
        if existing_memories:
            existing_text = "\n".join([
                f"- [{m.id}] {m.memory}"
                for m in existing_memories
            ])
        
        # 构建 prompt
        user_prompt = f"""现有记忆：
{existing_text if existing_text else "无"}

对话内容：
{conversation}

请提取新的事实信息，返回 JSON 数组。"""
        
        # 调用 LLM
        client = await self._get_llm_client()
        response = await client.chat.completions.create(
            model=self.config.llm_model,
            messages=[
                {"role": "system", "content": self.SYSTEM_PROMPT},
                {"role": "user", "content": user_prompt}
            ],
            temperature=0.1,
            response_format={"type": "json_object"}
        )
        
        # 解析结果
        try:
            result = json.loads(response.choices[0].message.content)
            facts_data = result if isinstance(result, list) else result.get("facts", [])
        except json.JSONDecodeError:
            facts_data = []
        
        # 转换为 MemoryFact 对象
        facts = []
        for item in facts_data:
            if item.get("operation") == "skip":
                continue
            
            fact = MemoryFact(
                memory=item["fact"],
                confidence=item.get("confidence", 0.8),
                metadata={
                    "source": "conversation",
                    "operation": item.get("operation", "add")
                }
            )
            facts.append(fact)
        
        return facts[:self.config.max_facts_per_turn]
    
    def compute_hash(self, text: str) -> str:
        """计算文本的 md5 哈希"""
        return hashlib.md5(text.encode()).hexdigest()

4.5 核心 Memory 类

# core/memory.py
import json
from typing import List, Optional, Dict, Any
from datetime import datetime
from .types import (
    MemoryFact, MemoryConfig, SearchResult, 
    MemoryHistory, EntityType
)
from ..stores.vector_store import SQLiteVectorStore
from ..extractors.fact_extractor import FactExtractor
from ..stores.graph_store import GraphStore


class Memory:
    """核心记忆管理器"""
    
    def __init__(self, config: MemoryConfig = None):
        self.config = config or MemoryConfig()
        
        # 初始化组件
        self.vector_store = SQLiteVectorStore(
            db_path=self.config.history_db_path
        )
        self.fact_extractor = FactExtractor(self.config)
        
        if self.config.enable_graph:
            self.graph_store = GraphStore()
        else:
            self.graph_store = None
    
    async def add(
        self,
        messages: List[Dict[str, str]],
        *,
        user_id: Optional[str] = None,
        agent_id: Optional[str] = None,
        run_id: Optional[str] = None,
        infer: bool = True,
        metadata: Optional[Dict] = None
    ) -> List[MemoryFact]:
        """
        写入记忆（九阶段流水线）
        
        Args:
            messages: 对话消息列表
            user_id: 用户 ID（与 agent_id/run_id 至少一个）
            agent_id: Agent ID
            run_id: 运行 ID
            infer: 是否通过 LLM 抽取事实
            metadata: 附加元数据
        
        Returns:
            写入的记忆列表
        """
        # 校验：至少提供一个 scope ID
        if not any([user_id, agent_id, run_id]):
            raise ValueError(
                "At least one of user_id, agent_id, run_id must be provided"
            )
        
        # Phase 1: 消息预处理
        parsed_messages = self._parse_messages(messages)
        
        # Phase 2: 构建 filter
        filters = self._build_filters(user_id, agent_id, run_id)
        
        if infer:
            # Phase 3: LLM 事实抽取
            existing = await self.get_all(filters=filters)
            new_facts = await self.fact_extractor.extract_facts(
                parsed_messages, existing
            )
            
            # Phase 4: 事实去重与合并
            merged_facts = await self._deduplicate_and_merge(
                new_facts, existing
            )
            
            # Phase 5: Embedding 向量化
            embeddings = await self._embed([f.memory for f in merged_facts])
            
            # Phase 6: 向量写入
            for fact, embedding in zip(merged_facts, embeddings):
                fact.user_id = user_id
                fact.agent_id = agent_id
                fact.run_id = run_id
                if metadata:
                    fact.metadata.update(metadata)
                
                await self.vector_store.insert(
                    vectors=[embedding],
                    ids=[fact.id],
                    payloads=[fact.model_dump()]
                )
            
            # Phase 7: 图操作（如果启用）
            if self.graph_store:
                await self._update_graph(merged_facts)
            
            # Phase 8: 记录历史
            for fact in merged_facts:
                await self._add_history(fact, "ADD")
            
            # Phase 9: 返回
            return merged_facts
        
        else:
            # 快速通道：直接写入
            facts = []
            for msg in messages:
                fact = MemoryFact(
                    memory=msg.get("content", ""),
                    user_id=user_id,
                    agent_id=agent_id,
                    run_id=run_id,
                    metadata=metadata or {}
                )
                embedding = (await self._embed([fact.memory]))[0]
                
                await self.vector_store.insert(
                    vectors=[embedding],
                    ids=[fact.id],
                    payloads=[fact.model_dump()]
                )
                facts.append(fact)
            
            return facts
    
    async def search(
        self,
        query: str,
        *,
        top_k: int = 20,
        filters: Optional[Dict] = None,
        threshold: float = 0.1
    ) -> List[SearchResult]:
        """
        三路混合检索
        
        Args:
            query: 查询文本
            top_k: 返回数量
            filters: 过滤条件
            threshold: 最低分数阈值
        
        Returns:
            检索结果列表
        """
        # Step 1: 语义检索
        query_embedding = (await self._embed([query]))[0]
        semantic_results = await self.vector_store.search(
            query_vector=query_embedding,
            top_k=top_k * 2,
            filters=filters
        )
        
        # Step 2: 关键词检索
        keyword_results = await self.vector_store.keyword_search(
            query=query,
            top_k=top_k * 2,
            filters=filters
        )
        
        # Step 3: 图查询（如果启用）
        graph_results = []
        if self.graph_store:
            graph_results = await self.graph_store.search(
                query=query,
                top_k=top_k * 2,
                filters=filters
            )
        
        # Step 4: RRF 融合排序
        fused = self._reciprocal_rank_fusion(
            semantic_results, keyword_results, graph_results,
            weights=[0.6, 0.2, 0.2]
        )
        
        # Step 5: 阈值过滤
        filtered = [r for r in fused if r.score >= threshold]
        
        return filtered[:top_k]
    
    async def get_all(
        self,
        *,
        filters: Optional[Dict] = None,
        top_k: int = 20
    ) -> List[MemoryFact]:
        """获取所有记忆"""
        results = await self.vector_store.search(
            query_vector=[0] * self.config.embedding_dim,  # 虚拟查询
            top_k=top_k,
            filters=filters
        )
        return [
            MemoryFact(**r["payload"]) 
            for r in results
        ]
    
    async def update(
        self,
        memory_id: str,
        data: str,
        metadata: Optional[Dict] = None
    ) -> MemoryFact:
        """更新单条记忆"""
        # 获取旧记忆
        old = await self.vector_store.get(memory_id)
        if not old:
            raise ValueError(f"Memory {memory_id} not found")
        
        old_payload = old["payload"]
        
        # 创建新记忆
        new_fact = MemoryFact(
            id=memory_id,
            memory=data,
            user_id=old_payload.get("user_id"),
            agent_id=old_payload.get("agent_id"),
            run_id=old_payload.get("run_id"),
            metadata={**old_payload.get("metadata", {}), **(metadata or {})},
            updated_at=datetime.now().isoformat()
        )
        
        # 重新计算 embedding
        new_embedding = (await self._embed([data]))[0]
        
        # 更新存储
        await self.vector_store.update(
            vector_id=memory_id,
            vector=new_embedding,
            payload=new_fact.model_dump()
        )
        
        # 记录历史
        await self._add_history(new_fact, "UPDATE", old_memory=old_payload.get("memory"))
        
        return new_fact
    
    async def delete(self, memory_id: str) -> None:
        """删除单条记忆"""
        old = await self.vector_store.get(memory_id)
        if old:
            await self._add_history(
                MemoryFact(**old["payload"]), 
                "DELETE"
            )
        
        await self.vector_store.delete(memory_id)
    
    def _parse_messages(self, messages: List[Dict]) -> List[Dict]:
        """消息预处理"""
        parsed = []
        for msg in messages:
            parsed.append({
                "role": msg.get("role", "user"),
                "content": msg.get("content", "")
            })
        return parsed
    
    def _build_filters(
        self, user_id, agent_id, run_id
    ) -> Dict:
        """构建 filter"""
        filters = {}
        if user_id:
            filters["user_id"] = user_id
        if agent_id:
            filters["agent_id"] = agent_id
        if run_id:
            filters["run_id"] = run_id
        return filters
    
    async def _deduplicate_and_merge(
        self,
        new_facts: List[MemoryFact],
        existing: List[MemoryFact]
    ) -> List[MemoryFact]:
        """事实去重与合并"""
        merged = []
        
        for new_fact in new_facts:
            is_duplicate = False
            
            for old_fact in existing:
                # 计算相似度（简化版：hash 比较）
                if new_fact.hash == old_fact.hash:
                    is_duplicate = True
                    break
                
                # 模糊去重：如果内容高度相似，更新旧的
                similarity = self._compute_similarity(
                    new_fact.memory, old_fact.memory
                )
                if similarity > self.config.dedup_threshold:
                    # 更新旧记忆
                    await self.update(
                        old_fact.id,
                        new_fact.memory,
                        metadata={"merged_from": old_fact.id}
                    )
                    is_duplicate = True
                    break
            
            if not is_duplicate:
                merged.append(new_fact)
        
        return merged
    
    def _compute_similarity(self, text1: str, text2: str) -> float:
        """计算文本相似度（简化版）"""
        # 生产环境应用 embedding 余弦相似度
        words1 = set(text1)
        words2 = set(text2)
        intersection = words1 & words2
        union = words1 | words2
        return len(intersection) / len(union) if union else 0.0
    
    def _reciprocal_rank_fusion(
        self,
        *result_lists,
        weights: List[float] = None
    ) -> List[SearchResult]:
        """RRF 融合排序"""
        k = 60  # RRF 常数
        scores = {}
        
        for i, results in enumerate(result_lists):
            weight = weights[i] if weights and i < len(weights) else 1.0
            
            for rank, result in enumerate(results):
                mid = result["id"]
                rrf_score = weight / (k + rank + 1)
                
                if mid not in scores:
                    scores[mid] = {
                        "id": mid,
                        "score": 0.0,
                        "payload": result["payload"]
                    }
                scores[mid]["score"] += rrf_score
        
        # 转换为 SearchResult 并排序
        results = [
            SearchResult(
                memory=MemoryFact(**item["payload"]),
                score=item["score"],
                source="fusion"
            )
            for item in scores.values()
        ]
        results.sort(key=lambda x: x.score, reverse=True)
        
        return results
    
    async def _embed(self, texts: List[str]) -> List[List[float]]:
        """文本向量化（简化版，生产环境用 OpenAI embedding API）"""
        # 简化的 embedding 实现：基于字符频率的伪向量
        # 生产环境应替换为真实的 embedding 模型
        import hashlib
        
        embeddings = []
        for text in texts:
            # 用 hash 生成伪向量（仅用于演示）
            h = hashlib.md5(text.encode()).hexdigest()
            vec = [float(int(h[i:i+2], 16)) / 255.0 for i in range(0, 32, 2)]
            # 填充到目标维度
            vec = vec * (self.config.embedding_dim // len(vec) + 1)
            vec = vec[:self.config.embedding_dim]
            embeddings.append(vec)
        
        return embeddings
    
    async def _update_graph(self, facts: List[MemoryFact]):
        """更新图存储（简化版）"""
        if not self.graph_store:
            return
        
        for fact in facts:
            # 简化的实体抽取
            entities = self._extract_simple_entities(fact.memory)
            
            for entity_name, entity_type in entities:
                await self.graph_store.upsert_entity(
                    name=entity_name,
                    entity_type=entity_type
                )
                await self.graph_store.add_edge(
                    entity_name=entity_name,
                    memory_id=fact.id,
                    relation="MENTIONS"
                )
    
    def _extract_simple_entities(self, text: str) -> List[tuple]:
        """简化的实体抽取"""
        entities = []
        
        # 抽取人名模式
        import re
        person_patterns = [
            r"(?:我|你|他|她|咱们)\s*(?:叫|是|的名字是)\s*(\w+)",
        ]
        for pattern in person_patterns:
            matches = re.findall(pattern, text)
            for match in matches:
                entities.append((match, EntityType.PERSON))
        
        # 抽取技术概念
        tech_keywords = [
            "Python", "JavaScript", "Rust", "Go", "Docker", "Kubernetes",
            "React", "Vue", "Node.js", "PostgreSQL", "Redis", "Mem0"
        ]
        for keyword in tech_keywords:
            if keyword.lower() in text.lower():
                entities.append((keyword, EntityType.CONCEPT))
        
        return entities
    
    async def _add_history(
        self, 
        fact: MemoryFact, 
        operation: str,
        old_memory: Optional[str] = None
    ):
        """记录变更历史"""
        history = MemoryHistory(
            memory_id=fact.id,
            old_memory=old_memory,
            new_memory=fact.memory if operation != "DELETE" else None,
            operation=operation
        )
        # 简化版：打印到控制台
        # 生产环境应写入数据库
        print(f"[History] {operation}: {fact.memory[:50]}...")

4.6 图存储（简化版）

# stores/graph_store.py
from typing import List, Dict, Optional, Set
from ..core.types import EntityType


class GraphStore:
    """简化的实体-记忆二部图存储"""
    
    def __init__(self):
        # 实体节点
        self.entities: Dict[str, Dict] = {}
        # 记忆节点
        self.memories: Dict[str, Dict] = {}
        # 边：实体ID → 记忆ID 列表
        self.edges: Dict[str, Set[str]] = {}
        # 反向边：记忆ID → 实体ID 列表
        self.reverse_edges: Dict[str, Set[str]] = {}
    
    async def upsert_entity(
        self,
        name: str,
        entity_type: EntityType,
        metadata: Optional[Dict] = None
    ) -> str:
        """创建或更新实体"""
        entity_id = f"entity:{name.lower()}"
        
        if entity_id not in self.entities:
            self.entities[entity_id] = {
                "id": entity_id,
                "name": name,
                "type": entity_type,
                "metadata": metadata or {}
            }
            self.edges[entity_id] = set()
        
        return entity_id
    
    async def add_edge(
        self,
        entity_name: str,
        memory_id: str,
        relation: str = "MENTIONS"
    ) -> None:
        """添加边"""
        entity_id = f"entity:{entity_name.lower()}"
        
        if entity_id not in self.edges:
            self.edges[entity_id] = set()
        self.edges[entity_id].add(memory_id)
        
        if memory_id not in self.reverse_edges:
            self.reverse_edges[memory_id] = set()
        self.reverse_edges[memory_id].add(entity_id)
    
    async def search(
        self,
        query: str,
        top_k: int = 20,
        filters: Optional[Dict] = None
    ) -> List[Dict]:
        """图增强检索"""
        # 从查询中提取关键词
        keywords = query.lower().split()
        
        # 找到匹配的实体
        matched_entities = []
        for entity_id, entity in self.entities.items():
            name_lower = entity["name"].lower()
            for keyword in keywords:
                if keyword in name_lower or name_lower in keyword:
                    matched_entities.append(entity_id)
                    break
        
        # 从实体出发找关联的记忆
        memory_ids = set()
        for entity_id in matched_entities:
            memory_ids.update(self.edges.get(entity_id, set()))
        
        # 二跳检索
        second_hop = set()
        for memory_id in memory_ids:
            for entity_id in self.reverse_edges.get(memory_id, set()):
                second_hop.update(self.edges.get(entity_id, set()))
        memory_ids.update(second_hop)
        
        # 构建结果
        results = []
        for mid in list(memory_ids)[:top_k]:
            if mid in self.memories:
                results.append({
                    "id": mid,
                    "score": 0.5,  # 图检索的基础分数
                    "payload": self.memories[mid]
                })
        
        return results

4.7 完整演示

# examples/demo.py
import asyncio
from agent_memory.core.memory import Memory
from agent_memory.core.types import MemoryConfig


async def main():
    """演示长期记忆引擎的完整流程"""
    
    # 初始化记忆系统
    config = MemoryConfig(
        enable_graph=True,
        llm_model="gpt-4o-mini",
        embedding_dim=1536
    )
    memory = Memory(config)
    
    print("=" * 60)
    print("AI Agent 长期记忆引擎演示")
    print("=" * 60)
    
    # === 场景 1：初次对话，积累偏好 ===
    print("\n📝 场景 1：初次对话")
    print("-" * 40)
    
    messages_1 = [
        {"role": "user", "content": "我叫张三，在一家做 AI 的创业公司工作"},
        {"role": "assistant", "content": "你好张三！很高兴认识你。AI 创业公司听起来很有意思。"},
        {"role": "user", "content": "我们主要用 Python 和 PyTorch 做模型训练"},
        {"role": "assistant", "content": "Python + PyTorch 是很好的技术栈选择。"},
        {"role": "user", "content": "我个人比较喜欢用 VS Code 写代码，用 Vim 做快速编辑"},
    ]
    
    facts_1 = await memory.add(
        messages_1,
        user_id="zhangsan",
        infer=True
    )
    
    print(f"写入了 {len(facts_1)} 条记忆：")
    for f in facts_1:
        print(f"  - {f.memory} (置信度: {f.confidence})")
    
    # === 场景 2：检索记忆 ===
    print("\n🔍 场景 2：检索记忆")
    print("-" * 40)
    
    results = await memory.search(
        "张三用什么编程语言",
        filters={"user_id": "zhangsan"}
    )
    
    print(f"检索到 {len(results)} 条相关记忆：")
    for r in results:
        print(f"  - {r.memory.memory} (分数: {r.score:.3f})")
    
    # === 场景 3：跨会话记忆更新 ===
    print("\n🔄 场景 3：跨会话记忆更新")
    print("-" * 40)
    
    messages_2 = [
        {"role": "user", "content": "我最近开始用 Cursor 了，比 VS Code 好用很多"},
        {"role": "assistant", "content": "Cursor 确实很流行，它的 AI 辅助功能很强大。"},
        {"role": "user", "content": "对了，我们公司搬到了北京朝阳区"},
    ]
    
    facts_2 = await memory.add(
        messages_2,
        user_id="zhangsan",
        infer=True
    )
    
    print(f"更新了 {len(facts_2)} 条记忆：")
    for f in facts_2:
        print(f"  - {f.memory}")
    
    # === 场景 4：多跳关联查询 ===
    print("\n🔗 场景 4：多跳关联查询")
    print("-" * 40)
    
    results = await memory.search(
        "张三的工作相关",
        filters={"user_id": "zhangsan"}
    )
    
    print(f"检索到 {len(results)} 条相关记忆：")
    for r in results:
        print(f"  - {r.memory.memory} (分数: {r.score:.3f})")
    
    # === 场景 5：统计信息 ===
    print("\n📊 场景 5：记忆统计")
    print("-" * 40)
    
    all_memories = await memory.get_all(
        filters={"user_id": "zhangsan"},
        top_k=100
    )
    
    print(f"总共存储了 {len(all_memories)} 条记忆：")
    for m in all_memories:
        print(f"  - {m.memory}")
    
    print("\n" + "=" * 60)
    print("演示完成！")
    print("=" * 60)


if __name__ == "__main__":
    asyncio.run(main())

五、LoCoMo 基准与选型决策树

5.1 为什么 LoCoMo 是最难的基准

LoCoMo 是 2024 年提出的长对话记忆评测集，每个对话平均 600+ 轮、跨 35 个会话、横跨数月。它测试的能力包括：

• 时序推理：「你上个月说过什么时候去日本？」
• 多跳关联：「我提过的那家公司，CEO 是谁？」
• 隐式偏好：用户从来没明说，但反复表现的偏好
• 矛盾消解：用户三个月前说喜欢，最近改主意了

评分使用 J Score——综合「事实正确 + 时序正确 + 上下文相关」的复合分。

5.2 十种方案完整横评

5.3 选型决策树

你的 Agent 要面对什么场景？
│
├── 单轮事实问答（FAQ / 客服）
│   → 选 RAG，便宜够用
│
├── 短对话 + 强偏好记忆（推荐 / 个性化）
│   → 选 Mem0，性价比之王
│
├── 长对话 + 时序推理（私人助理 / 治疗咨询）
│   → 选 Mem0g，图谱在这里值回票价
│
├── 极致准确性 + 不在乎成本（医疗 / 法律）
│   → 选 Full-context，但要做好钱包准备
│
├── 大段文档阅读（合同审查 / 论文综述）
│   → 选 ReadAgent，分块摘要 + 按需展开
│
└── 自主决策何时存取（高级 Agent）
    → 选 A-Mem，LLM 自己决定

六、生产部署的五个隐藏坑

6.1 记忆爆炸

任何方案在跑 6 个月后，记忆条目都会膨胀到几十万条。必须设计淘汰策略：

class MemoryEviction:
    """记忆淘汰策略"""
    
    @staticmethod
    async def evict_by_frequency(store, threshold=0.1):
        """按访问频率淘汰"""
        all_memories = await store.get_all()
        for mem in all_memories:
            access_count = mem.metadata.get("access_count", 0)
            if access_count < threshold:
                await store.delete(mem.id)
    
    @staticmethod
    async def evict_by_recency(store, max_age_days=90):
        """按时间淘汰"""
        from datetime import datetime, timedelta
        cutoff = (datetime.now() - timedelta(days=max_age_days)).isoformat()
        
        all_memories = await store.get_all()
        for mem in all_memories:
            if mem.updated_at and mem.updated_at < cutoff:
                await store.delete(mem.id)
    
    @staticmethod
    async def evict_by_importance(store, min_confidence=0.3):
        """按置信度淘汰"""
        all_memories = await store.get_all()
        for mem in all_memories:
            if mem.confidence < min_confidence:
                await store.delete(mem.id)

6.2 错误记忆固化

LLM 抽取的事实如果错了，会持续被检索回来污染后续推理。解决方案：

1. 给用户「删除/更正」入口
2. 每次抽取记录置信度，置信度低的不直接落库
3. 定期人工抽检（每月 100 条）

6.3 跨 Agent 记忆冲突

多 Agent 系统里，不同 Agent 写入同一份记忆，最终结论自相矛盾。解决方案：

class WriteArbiter:
    """写入仲裁层"""
    
    async def arbitrate(self, new_fact, existing_facts):
        """仲裁写入冲突"""
        
        # 规则 1：时间优先（最新的覆盖旧的）
        for existing in existing_facts:
            if self._is_contradiction(new_fact, existing):
                if new_fact.created_at > existing.created_at:
                    return "UPDATE", existing.id
                else:
                    return "SKIP", None
        
        # 规则 2：角色权重（管理员 > 普通 Agent）
        # 规则 3：置信度优先
        
        return "ADD", None

6.4 Embedding 漂移

Embedding 模型升级后，旧向量和新向量不在同一个空间，检索质量下降。解决方案：

• 版本化 embedding 模型（embedding_model: "text-embedding-3-small-v2"）
• 升级后批量重算旧向量
• 或使用兼容性更好的模型

6.5 隐私与合规

记忆系统存储了用户的长期偏好和历史，必须考虑：

• 数据加密存储
• 访问控制（用户只能访问自己的记忆）
• 审计日志（谁在什么时候读写了什么）
• GDPR 合规（用户有权删除自己的记忆）

七、总结与展望

7.1 核心结论

1. 记忆是 2026 Agent 的核心竞争力：没有记忆的 Agent 只是无状态工具
2. 结构化长期记忆是工业级答案：准确率打 9 折，成本打 1 折
3. 图增强有价值但非必须：只有时序/多跳场景才需要图谱
4. Mem0 的设计哲学值得学习：简单、务实、可扩展

7.2 未来趋势

• 多模态记忆：不只记住文字，还能记住图片、视频、音频
• 联邦记忆：多个 Agent 共享记忆但保持隐私
• 记忆压缩：自动将低价值记忆压缩为高价值摘要
• 主动记忆：Agent 主动询问「你想让我记住什么？」
• 遗忘机制：模拟人类遗忘曲线，自动淘汰过时信息

7.3 一句话总结

2026 年的 Agent 记忆系统，不再是比谁准确率最高，而是比谁能在三维 Pareto（准确率 × 成本 × 延迟）上找到最务实的那个点。Mem0 用 67 分赢得 14 倍成本压缩，是 2026 年最重要的「够用美学」。

模型可以租，准确率可以买，但让你的 Agent 跨周跨月持续记住用户的能力，是真正的护城河。