用 SQLite 实现 memU：为 openclaw 等主动代理提供跨会话 episodic/semantic 记忆

在构建 24/7 主动代理如 openclaw 时，核心挑战在于实现跨会话的持久记忆：代理需记住用户偏好、历史事件，并在无明确指令时主动行动。memU 框架提供文件系统式分层记忆（资源→事项→类别），通过连续学习（memorize）和智能检索（retrieve）显著降低 LLM token 消耗。本文聚焦用 SQLite 复现 memU 的 episodic（事件性）和 semantic（语义性）记忆，实现本地化、低成本的会话跨越召回与自主状态管理。相较原生 PostgreSQL/pgvector，SQLite 方案更适合单机部署，结合 sqlite-vss 扩展支持向量搜索。

memU 记忆架构简析与 SQLite 适配理由

memU 将记忆视为文件系统：类别如文件夹自动组织主题，事项如文件存储提取事实（偏好、技能），资源如挂载点保留原始对话 / 文档。“记忆即文件系统，文件系统即记忆” 设计确保结构化导航与跨引用，形成知识图谱 [1]。证据显示，此架构在 Locomo 基准上达 92.09% 准确率，支持双模检索：RAG 快速上下文组装（毫秒级），LLM 深度推理（秒级）。

SQLite 适配优势：零依赖本地存储，sqlite-vss 提供高效向量索引（HNSW），适合 <10 万条记录的代理场景。风险限于大规模扩展（>1M 事项转 PG），但通过智能衰减（低价值事件定期合并 / 删除）保持轻量。落地参数：数据库文件 agent_memory.db，PRAGMA journal_mode=WAL 提升并发；embedding_dim=1536（text-embedding-3-small）。

SQLite Schema 设计：分层 episodic/semantic 支持

核心三表实现 hierarchical 结构，支持 episodic（时间戳事件，如 “用户查询 NVDA 股价”）与 semantic（提炼事实，如 “用户偏好科技股”）。

-- 资源表：原始输入
CREATE TABLE resources (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id TEXT NOT NULL,
    modality TEXT CHECK(modality IN ('conversation', 'document', 'image')),  -- memU 支持多模态
    content BLOB,  -- 文本/路径/URL
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
    hash TEXT UNIQUE  -- 去重
);

-- 事项表：episodic/semantic 事实，带向量
CREATE VIRTUAL TABLE items USING vss0(
    id INTEGER PRIMARY KEY,
    resource_id INTEGER,
    type TEXT CHECK(type IN ('episodic', 'semantic')),  -- episodic:事件, semantic:事实
    content TEXT,
    embedding BLOB,  -- 1536-dim vector
    importance REAL DEFAULT 1.0,  -- 0-1 分数，用于衰减
    FOREIGN KEY(resource_id) REFERENCES resources(id)
);

-- 类别表：自动聚类主题
CREATE TABLE categories (
    id INTEGER PRIMARY KEY,
    name TEXT,
    parent_id INTEGER,
    summary TEXT,
    embedding BLOB,
    user_id TEXT,
    FOREIGN KEY(parent_id) REFERENCES categories(id)
);

参数清单：

向量索引：vss0 HNSW，ef_construction=128，ef_search=64（召回 / 速度平衡）。
索引：CREATE INDEX idx_items_user_timestamp ON items(user_id, timestamp DESC);
衰减阈值：importance <0.3 且 age> 30 天 → 合并至 semantic 或删除。

此 schema 启用 session-spanning：user_id 隔离多用户，timestamp 支持时序召回。

核心 Pipeline：memorize 与 retrieve

1. memorize ()：连续学习管道

后台摄入新交互，LLM 提取事项 / 类别，立即嵌入存储。伪代码（Python + openai/sqlite-vss）：

import sqlite3
import openai
from vss import VSSModel  # sqlite-vss

client = openai.OpenAI()
vss_model = VSSModel('text-embedding-3-small')

async def memorize(db_path: str, content: str, user_id: str, modality: str='conversation'):
    conn = sqlite3.connect(db_path)
    # 1. 存资源
    cur = conn.cursor()
    cur.execute("INSERT OR IGNORE INTO resources (user_id, content, modality) VALUES (?, ?, ?)",
                (user_id, content, modality))
    resource_id = cur.lastrowid
    
    # 2. LLM 提取（系统提示：提取 3-5 事项，分类 episodic/semantic）
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": "提取关键事实..."}, {"role": "user", "content": content}]
    )
    items = parse_extraction(response.choices[0].message.content)  # [{"type": "semantic", "content": "..."}]
    
    # 3. 嵌入 & 存事项
    for item in items:
        emb = vss_model.encode(item['content'])
        cur.execute("INSERT INTO items (resource_id, type, content, embedding) VALUES (?, ?, ?, ?)",
                    (resource_id, item['type'], item['content'], emb))
    
    # 4. 更新类别（相似搜索聚类）
    cats = cluster_items(cur, items)
    for cat in cats:
        cat_emb = vss_model.encode(cat['summary'])
        cur.execute("INSERT OR REPLACE INTO categories ...")
    
    conn.commit()
    return {"items": len(items), "categories": len(cats)}

参数：

LLM 预算：max_tokens=512 / 调用，temperature=0.1（确定性提取）。
批处理：>10 事项时分批，rate_limit=10/min。
去重：content hash，相似度 > 0.95 跳过。

2. retrieve ()：双模召回

RAG 模式快速（embedding knn），LLM 模式预测意图。

async def retrieve(db_path: str, query: str, user_id: str, method: str='rag', top_k: int=5):
    conn = sqlite3.connect(db_path)
    q_emb = vss_model.encode(query)
    
    if method == 'rag':
        # 向量搜索
        results = conn.execute("SELECT * FROM items WHERE vss_search(embedding, ?) LIMIT ?", (q_emb, top_k))
        return format_context(results.fetchall())
    else:  # 'llm'
        rag_context = retrieve(..., method='rag', top_k=20)
        # LLM 精炼意图 & 扩展查询
        refined = client.chat.completions.create(model="gpt-4o-mini", messages=[...])
        return retrieve(..., query=refined, top_k=top_k)

参数：

相似阈值：cosine > 0.75 返回。
混合模式：RAG 先筛，LLM 后验（token 节省 80%）。

Proactive Loop：自主状态管理

为 openclaw 等代理添加后台线程，实现 24/7 监控：

输入监控：hook 代理 I/O，每 30s 批 memorize 新事件。
模式匹配：embedding 比对历史（阈值 0.8），触发 retrieve。
意图预测：LLM 给定上下文，输出 action（如 “预取 NVDA 数据”）或建议。
执行 / 通知：低风险自主（如更新 todo），高风险问用户。

伪代码循环：

import asyncio
import threading

async def proactive_loop(db_path: str, agent_queue: asyncio.Queue):
    while True:
        # 拉新输入
        new_contents = await agent_queue.get()
        await memorize(db_path, new_contents, user_id)
        
        # 快速扫描
        recent_embs = get_recent_embeddings(300)  # 最近 5min
        matches = vector_match(recent_embs, threshold=0.8)
        if matches:
            context = await retrieve(db_path, matches[0]['content'])
            action = await llm_decide(context)  # "suggest: reschedule meeting"
            agent_queue.put(action)  # 推回代理
        await asyncio.sleep(30)  # poll_interval

工程参数 / 清单：

参数	值	说明
poll_interval	30s	平衡响应 / 负载
match_threshold	0.8	cosine，调低增敏感
max_context_items	10	LLM 输入限
decay_interval	1h	跑 importance 衰减
token_budget_daily	10k	超阈值降级 RAG

监控点：

Prometheus 指标：memory_size (rows), retrieval_hit_rate (>90%), token_saved (vs full context)。
告警：db_size >1GB，recall_accuracy <85%（人工验证集）。
回滚：备份 db 前部署，A/B 测试新 pipeline 1 周。

集成与风险缓解

对 openclaw：monkey-patch input/output hooks 调用 memorize；状态管理用 categories ['pending_tasks'] 持久 todo。测试：模拟 100 会话，验证召回 F1>0.9。

风险：

向量漂移：每月 re-embed 10% 随机样本（drift_score>0.2）。
SQLite 瓶颈：>50 并发用 WAL + connection pool=20。

此方案使代理从被动响应转向自主进化，token 降至原 1/10。实际部署见 memUBot 示例 [1]。

资料来源： [1] https://github.com/NevaMind-AI/memU
[2] OpenClaw 文档与 proactive-agent 技能（playbooks.com）。

（正文字数：1256）