乐于分享
好东西不私藏

OpenClaw记忆系统深度解析

OpenClaw记忆系统深度解析

OpenClaw记忆系统深度解析:从底层原理到Python实现

探索AI助手如何"记住"过去:OpenClaw记忆系统的技术实现详解

引言:为什么AI需要记忆?

想象一下,如果你每次对话都要重新介绍自己,每次任务都要从头解释背景,那将是多么低效的体验。这正是早期AI助手的痛点——缺乏连续性。OpenClaw通过创新的记忆系统解决了这个问题,让AI助手能够持续学习、记住偏好、积累上下文

一、OpenClaw记忆系统的核心架构

1.1 文件系统存储:简单而强大的基础

OpenClaw采用纯Markdown文件作为记忆存储介质,这种设计的优势在于:

  • 人类可读:所有记忆都是明文Markdown,随时可查看
  • 版本可控:可与Git等版本控制系统无缝集成
  • 跨平台兼容:任何设备都能访问和编辑
  • 灾难恢复:文件损坏时易于修复

核心文件结构:

~/.openclaw/workspace/
├── MEMORY.md              # 长期记忆(仅主会话加载)
└── memory/
    ├── 2026-03-19.md      # 每日日志
    ├── 2026-03-18.md
    └── ...

1.2 两层记忆模型

短期记忆:每日日志

  • 文件名:memory/YYYY-MM-DD.md
  • 存储当日所有活动和笔记
  • 每次会话自动读取当天和前一天的日志
  • 提供连续的上下文信息

长期记忆:MEMORY.md

  • 精选的重要信息
  • 安全设计:仅在私密会话中加载
  • 包含重要决策、偏好设置、关键事实
  • 定期从每日日志中提炼更新

二、底层实现原理详解

2.1 记忆索引机制

SQLite向量数据库

OpenClaw使用SQLite存储记忆片段和嵌入向量,支持:

  1. 向量搜索:语义相似性匹配
  2. 全文本搜索:关键词精确匹配
  3. 混合搜索:结合两者的优势
-- 记忆片段表结构
CREATETABLEmemory_chunks(
idINTEGERPRIMARYKEYAUTOINCREMENT,
file_pathTEXTNOTNULL,-- 来源文件
start_lineINTEGERNOTNULL,-- 起始行
end_lineINTEGERNOTNULL,-- 结束行
contentTEXTNOTNULL,-- 内容文本
embeddingBLOB,-- 嵌入向量
content_hashTEXTNOTNULL,-- 内容哈希
created_atTIMESTAMP,
updated_atTIMESTAMP
);

2.2 嵌入向量生成

OpenClaw支持多种嵌入模型:

  1. 本地模型(优先):   - embeddinggemma-300m-qat-q8_0-GGUF(约0.6GB)   - 完全离线,无需网络   - 自动下载和缓存

  2. 云服务(备用):   - OpenAI text-embedding-3-small   - Google Gemini Embeddings   - Voyage AI   - Mistral AI

自动降级策略:

本地模型 → OpenAI → Gemini → Voyage → Mistral → 关键词搜索

2.3 混合搜索算法

OpenClaw的搜索不是简单的向量搜索,而是智能混合算法

# 混合搜索核心逻辑(简化)
defhybrid_search(query, max_results=5):
    # 1. 向量搜索(语义理解)
    vector_results = vector_search(query, max_results * 2)

    # 2. 关键词搜索(精确匹配)
    keyword_results = keyword_search(query, max_results * 2)

    # 3. 分数融合
    for result in vector_results:
        result['final_score'] = result['vector_score'] * 0.7  # 70%权重

    for result in keyword_results:
        if result not in results_map:
            result['final_score'] = result['keyword_score'] * 0.3  # 30%权重
        else:
            # 合并分数
            result['final_score'] += result['keyword_score'] * 0.3

    # 4. 时间衰减(提升近期记忆)
    for result in all_results:
        age_days = calculate_age(result['timestamp'])
        decay = exp(-age_days / 30)  # 30天半衰期
        result['final_score'] *= decay

    # 5. 返回排序结果
    return sorted(all_results, key=lambda x: x['final_score'], reverse=True)

2.4 记忆压缩机制

当会话接近上下文窗口限制时,OpenClaw触发记忆压缩

触发条件:剩余token < 软阈值
执行步骤:
1. 暂停用户交互
2. 提醒AI写入重要记忆
3. AI选择关键信息写入MEMORY.md
4. 清理过时上下文
5. 继续会话

关键配置:

{
"compaction":{
"reserveTokensFloor":20000,
"memoryFlush":{
"enabled":true,
"softThresholdTokens":4000,
"systemPrompt":"会话即将压缩,请保存持久记忆",
"prompt":"将重要笔记写入memory/YYYY-MM-DD.md"
}
}
}

三、Python实现示例

以下是一个简化的OpenClaw记忆系统Python实现:

# memory_system.py
importjson
importos
importsqlite3
importnumpyasnp
fromdatetimeimport datetime
frompathlibimport Path

classOpenClawMemory:
"""OpenClaw记忆系统Python实现"""

    def__init__(self, workspace_path="."):
        self.workspace = Path(workspace_path)
        self.memory_file = self.workspace / "MEMORY.md"
        self.memory_dir = self.workspace / "memory"

        # 初始化文件系统
        self.memory_dir.mkdir(parents=True, exist_ok=True)

        # 初始化数据库
        self.db_path = self.workspace / "memory_index.db"
        self._init_database()

    def_init_database(self):
"""初始化SQLite数据库"""
        self.conn = sqlite3.connect(str(self.db_path))
        cursor = self.conn.cursor()

        # 创建记忆片段表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS memory_chunks (
                id INTEGER PRIMARY KEY,
                file_path TEXT,
                content TEXT,
                embedding BLOB,
                created_at TIMESTAMP
            )
        ''')

        # 创建全文搜索表
        cursor.execute('''
            CREATE VIRTUAL TABLE IF NOT EXISTS memory_fts 
            USING fts5(content, content='memory_chunks')
        ''')

        self.conn.commit()

    defwrite(self, content, category="general"):
"""写入记忆"""
        # 1. 写入MEMORY.md
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        with open(self.memory_file, 'a', encoding='utf-8') as f:
            f.write(f"\n## [{timestamp}] {category}\n{content}\n")

        # 2. 写入每日日志
        today = datetime.now().strftime("%Y-%m-%d")
        daily_file = self.memory_dir / f"{today}.md"
        with open(daily_file, 'a', encoding='utf-8') as f:
            f.write(f"- {timestamp}: {content[:100]}...\n")

        # 3. 索引记忆
        self._index_content(content, str(self.memory_file))

        return True

    def_index_content(self, content, file_path):
"""索引内容到数据库"""
        # 分块(每块约400 tokens)
        chunk_size = 1000  # 简化版
        chunks = [content[i:i+chunk_size] 
                 for i in range(0, len(content), chunk_size)]

        cursor = self.conn.cursor()
        for chunk in chunks:
            # 生成嵌入向量
            embedding = self._generate_embedding(chunk)

            # 插入数据库
            cursor.execute('''
                INSERT INTO memory_chunks (file_path, content, embedding, created_at)
                VALUES (?, ?, ?, ?)
            ''', (file_path, chunk, 
                  embedding.tobytes() if embedding is not None else None,
                  datetime.now().isoformat()))

            # 插入全文搜索
            chunk_id = cursor.lastrowid
            cursor.execute(
                "INSERT INTO memory_fts (rowid, content) VALUES (?, ?)",
                (chunk_id, chunk)
            )

        self.conn.commit()

    def_generate_embedding(self, text):
"""生成嵌入向量(简化版)"""
        # 实际应用中会调用嵌入模型API
        vector = np.random.randn(384).astype(np.float32)
        norm = np.linalg.norm(vector)
        return vector / norm if norm > 0 else vector

    defsearch(self, query, max_results=5):
"""混合搜索"""
        results = []

        # 向量搜索
        query_embedding = self._generate_embedding(query)
        if query_embedding is not None:
            vector_results = self._vector_search(query_embedding, max_results)
            results.extend(vector_results)

        # 关键词搜索
        keyword_results = self._keyword_search(query, max_results)
        results.extend(keyword_results)

        # 去重和排序
        seen = {}
        final_results = []
        for result in results:
            key = result.get('content', '')[:100]
            if key not in seen:
                seen[key] = True
                final_results.append(result)

        return sorted(final_results, 
                     key=lambda x: x.get('score', 0), 
                     reverse=True)[:max_results]

    def_vector_search(self, query_embedding, limit):
"""向量搜索实现"""
        cursor = self.conn.cursor()
        cursor.execute(
            "SELECT id, content FROM memory_chunks WHERE embedding IS NOT NULL"
        )

        results = []
        for row_id, content in cursor.fetchall():
            # 获取存储的向量
            cursor.execute(
                "SELECT embedding FROM memory_chunks WHERE id = ?",
                (row_id,)
            )
            embedding_bytes = cursor.fetchone()[0]

            if embedding_bytes:
                chunk_embedding = np.frombuffer(embedding_bytes, dtype=np.float32)
                similarity = np.dot(query_embedding, chunk_embedding)

                if similarity > 0.3:
                    results.append({
                        'id': row_id,
                        'content': content[:200],
                        'score': float(similarity),
                        'type': 'vector'
                    })

        return sorted(results, key=lambda x: x['score'], reverse=True)[:limit]

    def_keyword_search(self, query, limit):
"""关键词搜索实现"""
        cursor = self.conn.cursor()
        cursor.execute('''
            SELECT rowid, content FROM memory_fts 
            WHERE memory_fts MATCH ? 
            ORDER BY rank LIMIT ?
        ''', (f'"{query}"', limit))

        results = []
        for row_id, content in cursor.fetchall():
            # 计算简单相关性
            score = 0.5
            for word in query.lower().split():
                if word in content.lower():
                    score += 0.1

            results.append({
                'id': row_id,
                'content': content[:200],
                'score': score,
                'type': 'keyword'
            })

        return results

    defclose(self):
"""清理资源"""
        if self.conn:
            self.conn.close()

# 使用示例
if __name__ == "__main__":
    # 初始化
    memory = OpenClawMemory("/tmp/openclaw_demo")

    # 写入记忆
    memory.write("OpenClaw是一个强大的AI助手平台", "技术")
    memory.write("记忆系统支持向量搜索和关键词搜索", "功能")

    # 搜索记忆
    results = memory.search("OpenClaw 记忆")
    for i, result in enumerate(results, 1):
        print(f"结果{i}: {result['content']} (分数: {result['score']:.3f})")

    # 清理
    memory.close()

四、高级特性与优化

4.1 QMD后端(实验性)

OpenClaw支持QMD作为记忆后端,提供: - BM25 + 向量 + 重排器的三阶段搜索 - 完全本地运行,无需API密钥 - 自动模型下载和管理

4.2 记忆缓存机制

  • 嵌入向量缓存:避免重复计算
  • 搜索结果缓存:加速常见查询
  • 增量索引:只处理变更文件

4.3 安全与隐私

  1. 会话隔离:不同会话的记忆不共享
  2. 文件权限:严格的文件访问控制
  3. 敏感信息过滤:可配置的隐私过滤器

4.4 性能优化

  • 批量处理:支持OpenAI Batch API
  • 异步索引:不阻塞用户交互
  • 智能分块:动态调整块大小

五、实际应用场景

5.1 持续学习助手

  • 记住用户偏好(语言、风格、主题)
  • 积累项目上下文(代码、文档、决策)
  • 跟踪任务进展和待办事项

5.2 团队协作工具

  • 共享团队知识和最佳实践
  • 记录会议纪要和决策
  • 维护项目文档和指南

5.3 个人知识管理

  • 构建个人知识库
  • 记录学习和研究笔记
  • 整理灵感和创意

六、未来发展方向

  1. 多模态记忆:支持图像、音频记忆
  2. 主动记忆:AI主动识别和保存重要信息
  3. 记忆关联:建立记忆之间的语义连接
  4. 记忆迁移:在不同设备间同步记忆

结语

OpenClaw的记忆系统展示了简单与强大的完美结合。通过文件系统存储、混合搜索算法和智能压缩机制,它解决了AI助手的"健忘症"问题,为真正持续智能奠定了基础。

这种设计哲学值得所有AI系统借鉴:最复杂的功能可以用最优雅的方式实现。OpenClaw证明了,即使是最先进的AI技术,也可以建立在人类可理解、可控制的基础之上。

记忆不是数据的堆积,而是智慧的沉淀。OpenClaw让AI拥有了持续成长的能力,而不仅仅是重复回答。


扩展阅读: - OpenClaw官方文档 🔗 https://docs.openclaw.ai - SQLite向量扩展 🔗 https://github.com/asg017/sqlite-vec - QMD搜索引擎 🔗 https://github.com/tobi/qmd - 嵌入模型比较 🔗 https://huggingface.co/spaces/mteb/leaderboard