Lite SQLite — 轻量级本地数据库

专为 OpenClaw 智能体优化的超轻量级 SQLite 数据库管理方案，内存占用极低（约 2–5 MB），存储开销极小。

为何选择 SQLite？

✅ 零配置 — 无需服务端、无需配置，纯文件驱动 ✅ 内存占用极小 — 典型使用场景下仅需 2–5 MB RAM ✅ 性能卓越 — 每秒可执行数百万次查询 ✅ 高度便携 — 单一 .db 文件即可完整承载全部数据 ✅ 高可靠性 — 支持 ACID 事务，具备崩溃防护能力 ✅ 跨平台兼容 — 凡 Python 可运行之处，SQLite 均可运行

核心特性

• 内存模式（:memory:）支持临时数据操作（性能更优！）
• WAL（Write-Ahead Logging）模式支持高并发访问
• 连接池（Connection Pooling）
• 自动化的 Schema 迁移
• 内置备份与恢复功能
• 查询优化提示（Query Optimization Hints）

快速入门

基础数据库操作

from sqlite_connector import SQLiteDB

# 创建数据库（默认启用 WAL 模式）
db = SQLiteDB("agent_data.db")

# 创建表
db.create_table("memos", {
    "id": "INTEGER PRIMARY KEY AUTOINCREMENT",
    "title": "TEXT NOT NULL",
    "content": "TEXT",
    "created_at": "TEXT DEFAULT CURRENT_TIMESTAMP",
    "tags": "TEXT"
})

# 插入数据
db.insert("memos", [title="First memo", content="Hello world", tags="test"])

# 查询数据
results = db.query("SELECT * FROM memos WHERE tags = ?", ("test",))

# 更新数据
db.update("memos", "id = ?", [content="Updated content"], (1,))

# 删除数据
db.delete("memos", "id = ?", (1,))

# 关闭连接
db.close()

内存数据库（最快模式）

# 最快模式：纯内存运行，无磁盘 I/O
db = SQLiteDB(":memory:")

# 适用于临时性操作
db.create_table("temp", {...})

# 数据仅在当前会话生命周期内存在  
# 推荐用于缓存、中间计算或临时存储

性能优化

关键配置项

import sqlite3

# 启用 WAL 模式（Write-Ahead Logging）— 性能提升 3–4 倍
conn = sqlite3.connect("agent_data.db")
conn.execute("PRAGMA journal_mode=WAL")

# 关闭同步写入（加快写入速度；配合正确关闭流程仍可保证崩溃安全）
conn.execute("PRAGMA synchronous=NORMAL")

# 内存优化
conn.execute("PRAGMA cache_size=-64000")  # 设置 64 MB 缓存
conn.execute("PRAGMA page_size=4096")

# 将临时存储置于内存中
conn.execute("PRAGMA temp_store=MEMORY")

查询优化技巧

# 对高频查询字段建立索引
db.create_index("memos", "tags")
db.create_index("memos", "created_at")

# 使用预编译语句（本封装已自动支持）
db.query("SELECT * FROM memos WHERE id = ?", (id,))

# 批量插入处理大规模数据集
db.batch_insert("memos", rows_data)

预定义 Schema

智能体备忘录 Schema（记忆存储）

db.create_table("agent_memos", {
    "id": "INTEGER PRIMARY KEY AUTOINCREMENT",
    "agent_id": "TEXT NOT NULL",           # 创建该记录的智能体 ID
    "key": "TEXT NOT NULL",               # 查找键（Lookup key）
    "value": "TEXT",                      # 存储值
    "priority": "INTEGER DEFAULT 0",       # 检索排序优先级
    "created_at": "TEXT DEFAULT CURRENT_TIMESTAMP",
    "expires_at": "TEXT"                  # 可选 TTL（过期时间）
})

# 创建索引
db.create_index("agent_memos", "agent_id")
db.create_index("agent_memos", "key")
db.create_index("agent_memos", "expires_at")

会话日志 Schema

db.create_table("session_logs", {
    "id": "INTEGER PRIMARY KEY AUTOINCREMENT",
    "session_id": "TEXT NOT NULL",
    "agent": "TEXT NOT NULL",
    "message": "TEXT",
    "metadata": "TEXT",                   # JSON 格式字符串
    "created_at": "TEXT DEFAULT CURRENT_TIMESTAMP"
})

db.create_index("session_logs", "session_id")
db.create_index("session_logs", "created_at")

缓存 Schema（基于 TTL）

db.create_table("cache", {
    "id": "INTEGER PRIMARY KEY AUTOINCREMENT",
    "key": "TEXT UNIQUE NOT NULL",
    "value": "BLOB",                      # 支持二进制数据
    "created_at": "TEXT DEFAULT CURRENT_TIMESTAMP",
    "expires_at": "TEXT NOT NULL"
})

# 自动清理已过期条目
db.query("DELETE FROM cache WHERE expires_at < ?", (datetime.now().isoformat(),))

db.create_index("cache", "key")
db.create_index("cache", "expires_at")

高级特性

连接池（Connection Pooling）

from sqlite_connector import ConnectionPool

# 创建连接池以支持并发访问
pool = ConnectionPool("agent_data.db", max_connections=5)

# 获取连接
conn = pool.get_connection()
# 使用 conn...
pool.release_connection(conn)

自动备份

# 手动备份数据库
db.backup("agent_data_backup.db")

# 启用每日自动备份
db.auto_backup("backups/", "daily")

Schema 迁移

# 若列不存在则添加新列
db.add_column("memos", "updated_at", "TEXT DEFAULT CURRENT_TIMESTAMP")

# 迁移现有数据（重命名字段等）
db.migrate("memos", {
    "old_column": "new_column"
})

性能基准测试

典型性能表现

内存占用

• 基础内存占用：2–5 MB
• 10 万行数据：约 10–15 MB
• 100 万行数据：约 50–100 MB
• 内存模式（:memory:）：约为数据实际大小 + 系统开销

OpenClaw 智能体最佳实践

1. 正确选择运行模式

# 临时操作请使用 :memory:
temp_db = SQLiteDB(":memory:")

# 持久化存储请使用文件数据库
persist_db = SQLiteDB("agent_storage.db")

2. 合理创建索引

# WHERE 子句中涉及的字段务必建索引
db.create_index("table", "column_name")

# 复合查询可建立多列联合索引
db.create_index("table", "col1, col2")

3. 批量操作替代逐条操作

# ❌ 避免逐条插入（性能差）：
for row in rows:
    db.insert("table", row)  # 慢！

# ✅ 推荐批量插入（高性能）：
db.batch_insert("table", rows)  # 快！

4. 利用 TTL 实现数据自动过期

# 自动清理过期缓存
db.cleanup_expired("cache", "expires_at")
# 自动清理 7 天前的日志
db.cleanup_old("logs", "created_at", days=7)

5. 定期执行 VACUUM 优化空间

# 在大量 DELETE 后回收磁盘空间
db.vacuum()  # 建议在系统空闲时段执行

DuckDB 替代方案（面向分析场景）

如需执行复杂分析类查询（聚合、大表关联等），建议选用 DuckDB：

import duckdb

conn = duckdb.connect(":memory:")

# 在复杂分析任务上显著优于 SQLite
conn.execute("""
    SELECT COUNT(*) as rows,
           AVG(value) as avg_value
    FROM large_table
""").fetchall()

何时选用 DuckDB：

• 超大规模数据集分析（>1 亿行）
• 复杂聚合与多表 JOIN
• 列式数据处理操作
• 统计分析与机器学习预处理

何时选用 SQLite：

• 事务型操作（增删改查）
• 中小规模数据集（<1 亿行）
• 点查（Point Query）与单行更新
• 通用目的持久化存储

常见使用模式

1. 备忘录存储

def save_memo(db, agent_id, key, value, ttl_hours=24):
    expires_at = (datetime.now() + timedelta(hours=ttl_hours)).isoformat()
    db.insert("agent_memos", {
        "agent_id": agent_id,
        "key": key,
        "value": json.dumps(value),
        "expires_at": expires_at
    })

2. 会话持久化

def save_session(db, session_id, agent, message, metadata=None):
    db.insert("session_logs", {
        "session_id": session_id,
        "agent": agent,
        "message": message,
        "metadata": json.dumps(metadata) if metadata else None
    })

3. 缓存层封装

def cache_get(db, key):
    if expired_key := db.query_one(
        "SELECT value FROM cache WHERE key = ? AND expires_at > ?",
        (key, datetime.now().isoformat())
    ):
        return json.loads(expired_key)
    return None

def cache_set(db, key, value, ttl_seconds=3600):
    expires_at = (datetime.now() + timedelta(seconds=ttl_seconds)).isoformat()
    db.insert_or_replace("cache", {
        "key": key,
        "value": json.dumps(value),
        "expires_at": expires_at
    })

错误处理

try:
    db.insert("metrics", {...})
except sqlite3.IntegrityError:
    # 主键/唯一约束冲突
    pass
except sqlite3.OperationalError:
    # 表不存在 或 数据库被锁
    pass

体积优化建议

减小存储体积

1. 选用合适的数据类型：

   • 数值型字段优先使用 INTEGER 而非 TEXT
   • 浮点型字段优先使用 REAL 而非 TEXT
   • 使用 CHECK 约束进行数据校验

2. 合理归一化数据结构：

   • JSON 字符串统一存为 TEXT
   • 变长字符串使用 TEXT
   • 避免冗余字段重复存储

3. 定期执行 VACUUM：

   db.vacuum()  # 删除操作后释放磁盘空间
   

4. 启用 WAL 模式替代传统日志（journal）：

   conn.execute("PRAGMA journal_mode=WAL")
   

从其他存储迁移

从 JSON 文件迁移

# 将 JSON 加载至 SQLite
import json

with open("data.json") as f:
    data = json.load(f)

db.create_table("json_data", {key: "TEXT" for key in data[0].keys()})
db.batch_insert("json_data", data)

从 CSV 文件迁移

import pandas as pd

df = pd.read_csv("data.csv")
df.to_sql("csv_data", conn, if_exists="replace", index=False)

故障排查

数据库被锁（Database Locked）

# 启用 WAL 模式提升并发能力
conn.execute("PRAGMA journal_mode=WAL")

# 或使用连接池缓解竞争
pool = ConnectionPool("db.db", timeout=5.0)

查询缓慢

# 查看查询执行计划
plan = conn.execute("EXPLAIN QUERY PLAN SELECT * FROM ...").fetchall()

# 添加缺失索引
db.create_index("table", "column")

# 更新统计信息以辅助查询优化器
conn.execute("ANALYZE")

数据库文件过大

# 查看数据库物理尺寸
size_info = conn.execute("PRAGMA page_count, page_size").fetchone()
print(f"Size: {(page_count * page_size) / (1024*1024):.2f} MB")

# 执行 VACUUM 回收已删除数据占用的空间
db.vacuum()

命令行工具（CLI）

配套的 sqlite_cli.py 提供便捷的命令行接口：

# 创建数据库
python scripts/sqlite_cli.py create agent_data.db

# 创建表
python scripts/sqlite_cli.py create-table agent_memos -c id:INTEGER:P -c title:TEXT -c content:TEXT

# 插入数据
python scripts/sqlite_cli.py insert agent_memos '{"title": "Test", "content": "Hello"}'

# 查询数据
python scripts/sqlite_cli.py query "SELECT * FROM agent_memos"

# 优化数据库（VACUUM + ANALYZE）
python scripts/sqlite_cli.py optimize agent_data.db

参考资料

• SQLite 官方文档： https://www.sqlite.org/docs.html
• Python sqlite3 模块文档： https://docs.python.org/3/library/sqlite3.html
• DuckDB 文档： https://duckdb.org/docs/
• SQLite 性能优化指南： https://www.sqlite.org/optoverview.html