基于Forma架构的灵活记忆引擎,为OpenClaw提供持久化记忆和上下文管理能力
OpenClaw记忆引擎是一个为AI助手设计的灵活数据存储系统,解决以下核心问题:
- 灵活性 - 支持任意类型的记忆,无需预定义Schema
- 性能 - 高频记忆快速检索(<50ms)
- 类型安全 - 使用JSON Schema校验记忆数据
- AI-Ready - 与LLM的结构化输出无缝集成
✅ EAV + 热表架构 - 灵活存储 + 高性能查询 ✅ JSON Schema校验 - 类型安全,防止脏数据 ✅ 全文搜索 - 基于SQLite FTS5的快速搜索 ✅ 记忆关系图 - 建立记忆之间的关联 ✅ 标签系统 - 灵活的分类和过滤 ✅ 访问统计 - 自动追踪记忆访问频率
pip install jsonschemafrom memory_engine import MemoryEngine
# 创建引擎实例
engine = MemoryEngine("my_memory.db")# 注册Person Schema
person_schema = {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"name": {
"type": "string",
"x-memory-column": "text_01" # 映射到热表
},
"role": {
"type": "string",
"x-memory-column": "text_02"
},
"importance": {
"type": "integer",
"minimum": 1,
"maximum": 10,
"x-memory-column": "integer_01"
},
"email": {"type": "string"},
"notes": {"type": "string"}
},
"required": ["name"]
}
engine.register_schema("person", "Person", person_schema)memory_id = engine.write_memory(
schema_id="person",
data={
"name": "张总",
"role": "客户",
"importance": 9,
"email": "zhang@example.com",
"notes": "预算50万,对新方案感兴趣"
},
tags=["客户", "重要"]
)# 普通查询
results = engine.query_memories(
schema_id="person",
filters={"importance": 9},
limit=10
)
# 全文搜索
search_results = engine.search_memories("张总")from openclaw_skill import MemorySkill
# 初始化技能
skill = MemorySkill()
# 记住某件事
skill.remember(
memory_type="person",
name="张总",
role="客户",
importance=9,
tags=["客户", "重要"]
)
# 回忆某件事
result = skill.recall(query="张总")
print(result['memories'])User: 记住,张总是我的重要客户,预算50万
OpenClaw: [调用 skill.remember()]
✓ 已记住: 张总
User: 张总是谁?
OpenClaw: [调用 skill.recall("张总")]
张总是你的重要客户,预算50万,重要性9/10
┌─────────────────────────────────────────────────────────────┐
│ Memory Engine API │
│ ┌──────────────┬──────────────┬──────────────┐ │
│ │ Write Memory │ Query Memory │ Update Memory│ │
│ └──────────────┴──────────────┴──────────────┘ │
└────────────────────────┬────────────────────────────────────┘
│
┌───────────────┼───────────────┐
│ │ │
▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ JSON Schema │ │ Hot Table │ │ EAV Table │
│ Validator │ │ (快速检索) │ │ (灵活存储) │
└──────────────┘ └──────────────┘ └──────────────┘
│ │
└───────┬───────┘
▼
┌──────────────┐
│ SQLite DB │
└──────────────┘
memory_main - 热表
- 存储高频访问的字段
- 预分配列(text_01, integer_01等)
- 有B-tree索引,查询快速
memory_eav - EAV表
- 存储所有字段(包括热字段)
- 灵活,无需DDL
- 按需聚合
memory_tags - 标签表
- 支持多标签
- 快速过滤
memory_relations - 关系表
- 记忆之间的关联
- 支持知识图谱
引擎内置了6种常用记忆类型:
{
"name": "张总",
"role": "客户",
"company": "ABC公司",
"importance": 9,
"email": "zhang@abc.com",
"notes": "预算50万"
}{
"title": "与张总电话沟通",
"category": "call",
"occurred_at": 1738483200,
"duration": 30,
"participants": ["张总", "我"],
"summary": "讨论新方案"
}{
"title": "准备方案PPT",
"status": "in_progress",
"priority": 8,
"due_date": 1738569600,
"assignee": "我"
}{
"title": "新产品创意",
"category": "产品",
"content": "基于AI的...",
"potential_score": 8
}{
"topic": "项目进度讨论",
"platform": "telegram",
"occurred_at": 1738483200,
"summary": "确认了下周的里程碑"
}{
"title": "项目需求文档",
"doc_type": "report",
"file_path": "/path/to/doc.pdf",
"summary": "详细的需求说明"
}注册一个新的记忆Schema
写入一条记忆
参数:
schema_id: Schema IDdata: 记忆数据(dict)tags: 标签列表relations: 关系列表[{"to": memory_id, "type": "related_to"}]
返回: memory_id或None
查询记忆
参数:
schema_id: 按Schema过滤filters: 字段过滤{"name": "张总"}tags: 按标签过滤limit/offset: 分页
返回: 记忆列表
全文搜索记忆
参数:
query: 搜索关键词schema_id: 按Schema过滤limit: 返回数量
返回: 记忆列表
记住某件事
回忆某件事
忘记某件事(软删除)
建立记忆关联
获取相关记忆
引擎会自动追踪字段访问频率,建议将高频字段映射到热表:
# 在JSON Schema中使用 x-memory-column
{
"name": {
"type": "string",
"x-memory-column": "text_01" # 映射到热表
}
}使用Forma的CTE + JSON_AGG模式,一次查询返回所有数据:
WITH filtered AS (
SELECT memory_id FROM memory_main WHERE ...
),
eav_json AS (
SELECT memory_id, JSON_GROUP_ARRAY(...) AS attributes
FROM memory_eav WHERE memory_id IN (SELECT memory_id FROM filtered)
GROUP BY memory_id
)
SELECT m.*, e.attributes
FROM memory_main m
LEFT JOIN eav_json e ON m.memory_id = e.memory_id- 写入延迟: <10ms
- 查询延迟: <50ms (热表) / <200ms (含EAV聚合)
- 搜索延迟: <100ms (FTS5全文搜索)
# 创建自定义Schema
custom_schema = {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {
"field1": {"type": "string", "x-memory-column": "text_01"},
"field2": {"type": "integer", "x-memory-column": "integer_01"},
# ... 更多字段
}
}
engine.register_schema("custom", "Custom Type", custom_schema)当数据量增长到百万级,可以扩展为冷热分离架构:
PostgreSQL (热数据) + DuckDB/Parquet (冷数据)
参考Forma的第三篇文章实现。
OpenClaw记忆引擎/
├── 00_架构设计.md # 详细的架构设计文档
├── schema.sql # 数据库Schema
├── memory_engine.py # 核心引擎实现
├── openclaw_skill.py # OpenClaw Skill接口
├── schemas.json # 预定义Schema模板
└── README.md # 本文件
- 数据库: SQLite 3.35+ (支持FTS5)
- 语言: Python 3.10+
- 依赖: jsonschema
- 添加更新记忆的API
- 实现记忆合并功能
- 添加记忆版本历史
- 开发Web UI管理界面
- 支持PostgreSQL后端
- 实现冷热分离架构
MIT License
Created by AI_Roland 2026-02-02