04-Mem0 的 SQLite 本地状态持久化

Mem0 的 SQLite 本地状态持久化

学习目标

本章分析 Mem0 如何使用 SQLite 管理两类本地数据：记忆变更历史与对话消息缓存。你将了解：

• SQLiteManager 的线程安全实现
• 自动表结构迁移逻辑
• 消息自动裁剪策略
• 批量操作的原子性保证

项目实践

两类数据表

SQLiteManager 管理两个表：

表名	用途	保留策略
history	记忆变更历史（ADD/UPDATE/DELETE）	永久保留
messages	最近对话消息缓存	自动裁剪，保留最近 N 条

线程安全设计

class SQLiteManager:
    def __init__(self, db_path=":memory:"):
        self.connection = sqlite3.connect(db_path, check_same_thread=False)
        self._lock = threading.Lock()

关键设计：

• check_same_thread=False：允许跨线程共享连接（SQLite 默认只允许创建连接的线程使用）
• threading.Lock()：所有写操作通过 _lock 串行化，防止并发写入冲突

每个写操作都遵循 BEGIN → COMMIT/ROLLBACK 模式：

def add_history(self, ...):
    with self._lock:
        try:
            self.connection.execute("BEGIN")
            self.connection.execute("INSERT INTO history ...", params)
            self.connection.execute("COMMIT")
        except Exception as e:
            self.connection.execute("ROLLBACK")
            raise

自动表结构迁移

Mem0 在初始化时自动检测 history 表的列结构，如有差异则迁移：

def _migrate_history_table(self):
    cur.execute("PRAGMA table_info(history)")
    old_cols = {row[1] for row in cur.fetchall()}

    expected_cols = {"id", "memory_id", "old_memory", "new_memory", "event",
                     "created_at", "updated_at", "is_deleted", "actor_id", "role"}

    if old_cols != expected_cols:
        # 迁移流程
        cur.execute("ALTER TABLE history RENAME TO history_old")
        cur.execute("CREATE TABLE history (...新结构...)")
        cur.execute(f"INSERT INTO history ({intersecting_cols}) SELECT ... FROM history_old")
        cur.execute("DROP TABLE history_old")

迁移策略：

1. 检测列差异
2. 旧表重命名为 history_old
3. 创建新表
4. 只拷贝交集列的数据
5. 删除旧表

任何步骤失败时回滚事务。

消息自动裁剪

save_messages() 在保存新消息后，自动清理旧消息：

# 保留最近 N 条消息
self.connection.execute("""
    DELETE FROM messages WHERE session_scope = ? AND id NOT IN (
        SELECT id FROM (
            SELECT id FROM messages WHERE session_scope = ?
            ORDER BY created_at DESC LIMIT 10
        )
    )
""", (session_scope, session_scope))

为什么使用双层子查询：SQLite 在 NOT IN 的子查询中不支持 ORDER BY + LIMIT。Mem0 使用双层子查询绕过这个限制：内层 SELECT ... ORDER BY ... LIMIT 先选出最近的 10 条 ID，外层 NOT IN 删除其余消息。

批量操作

batch_add_history() 使用 executemany 批量插入：

def batch_add_history(self, records):
    with self._lock:
        self.connection.execute("BEGIN")
        self.connection.executemany(
            "INSERT INTO history (...) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            [tuple_from_record(r) for r in records]
        )
        self.connection.execute("COMMIT")

批量操作在单个事务中完成，保证原子性。如果任何一条记录插入失败，整个事务回滚。

问题与规避

问题	对策
并发写入冲突	threading.Lock 串行化所有写操作
表结构变更导致数据丢失	自动迁移，只拷贝交集列
消息无限增长	每次保存后自动裁剪，保留最近 10 条
SQLite 文件锁冲突	check_same_thread=False + threading.Lock

设计取舍

为什么选 SQLite 而非文件存储？

优势：

• SQL 查询：支持按 memory_id、session_scope 过滤
• 事务保证：批量操作的原子性
• 自动迁移：表结构变更通过 SQL 完成
• 零依赖：Python 标准库内置

代价：

• 不支持高并发写入（单写者）
• 不适合多用户生产环境
• 数据量大时查询性能下降

为什么消息只保留 10 条？

10 条是 LLM 提取时使用的上下文窗口大小（get_last_messages(limit=10)）。保留更多消息没有意义——提取只用到最近 10 条，多余的只是存储浪费。

参考来源

• Mem0 源码：mem0/memory/storage.py
• SQLite 文档：https://www.sqlite.org/docs.html