OpenClaw记忆系统深度解析

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

-- 记忆片段表结构
CREATETABLEmemory_chunks(
idINTEGERPRIMARYKEYAUTOINCREMENT,
file_pathTEXTNOTNULL,-- 来源文件
start_lineINTEGERNOTNULL,-- 起始行
end_lineINTEGERNOTNULL,-- 结束行
contentTEXTNOTNULL,-- 内容文本
embeddingBLOB,-- 嵌入向量
content_hashTEXTNOTNULL,-- 内容哈希
created_atTIMESTAMP,
updated_atTIMESTAMP
);

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

# 混合搜索核心逻辑（简化）
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)

触发条件：剩余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"
}
}
}

# 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()