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存储记忆片段和嵌入向量,支持:
- 向量搜索:语义相似性匹配
- 全文本搜索:关键词精确匹配
- 混合搜索:结合两者的优势
-- 记忆片段表结构
CREATETABLEmemory_chunks(
idINTEGERPRIMARYKEYAUTOINCREMENT,
file_pathTEXTNOTNULL,-- 来源文件
start_lineINTEGERNOTNULL,-- 起始行
end_lineINTEGERNOTNULL,-- 结束行
contentTEXTNOTNULL,-- 内容文本
embeddingBLOB,-- 嵌入向量
content_hashTEXTNOTNULL,-- 内容哈希
created_atTIMESTAMP,
updated_atTIMESTAMP
);
2.2 嵌入向量生成
OpenClaw支持多种嵌入模型:
-
本地模型(优先): -
embeddinggemma-300m-qat-q8_0-GGUF(约0.6GB) - 完全离线,无需网络 - 自动下载和缓存 -
云服务(备用): - 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 安全与隐私
- 会话隔离:不同会话的记忆不共享
- 文件权限:严格的文件访问控制
- 敏感信息过滤:可配置的隐私过滤器
4.4 性能优化
- 批量处理:支持OpenAI Batch API
- 异步索引:不阻塞用户交互
- 智能分块:动态调整块大小
五、实际应用场景
5.1 持续学习助手
- 记住用户偏好(语言、风格、主题)
- 积累项目上下文(代码、文档、决策)
- 跟踪任务进展和待办事项
5.2 团队协作工具
- 共享团队知识和最佳实践
- 记录会议纪要和决策
- 维护项目文档和指南
5.3 个人知识管理
- 构建个人知识库
- 记录学习和研究笔记
- 整理灵感和创意
六、未来发展方向
- 多模态记忆:支持图像、音频记忆
- 主动记忆:AI主动识别和保存重要信息
- 记忆关联:建立记忆之间的语义连接
- 记忆迁移:在不同设备间同步记忆
结语
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
夜雨聆风