OpenClaw 养虾教程⑦-让 AI 拥有“长期记忆”:90% 的开发者不会用的数据库集成技巧

导读

前一篇我们聊了命令执行安全，让 AI 能安全地执行系统命令。但这还不够——AI 需要能"记住"东西。

想象这些场景：

用户说"帮我记录一下，明天下午 3 点开会" → AI 需要存到数据库
用户问"上周我让你记的待办事项有哪些？" → AI 需要查询历史数据
用户说"把最近的订单数据整理成报表" → AI 需要聚合分析

没有数据库的 AI，就像只有短期记忆的人——聊完就忘，无法真正帮你管理业务。

这篇就解决三个问题：

如何给 AI 配数据库？
如何设计安全的数据库访问权限？
如何避免 SQL 注入等安全问题？

一、为什么 AI 需要数据库？

1.1 真实场景：我的待办事项管理系统

最近我们团队用 OpenClaw 做了一个智能待办助手，流程是这样的：

用户说"记个待办" → AI 解析内容 → 存入数据库 → 到点提醒 → 完成后更新状态

最初我们用文件存储，每个待办是一个 JSON 文件：

Python # 错误做法：用文件存储结构化数据 async def add_todo(self, todo: str): todos = await self.read_file("todos.json") todos.append(todo) await self.write_file("todos.json", json.dumps(todos))

问题很快出现了：

并发冲突：两个人同时添加待办，后写的覆盖先写的
查询困难：想找"上周未完成的待办"，要遍历所有文件
数据一致性：AI 误操作删了文件，数据全丢
性能问题：待办超过 1000 条后，每次读取都要加载整个文件

后来改用SQLite 数据库，问题全解决了：

Python # 正确做法：用数据库存储结构化数据 async def add_todo(self, todo: str, due_date: str): await self.db.execute( "INSERT INTO todos (content, due_date, status, created_at) VALUES (?, ?, ?, ?)", [todo, due_date, "pending", datetime.now()] ) async def get_pending_todos(self, since: str): rows = await self.db.fetch( "SELECT * FROM todos WHERE status = ? AND created_at >= ?", ["pending", since] ) return rows

效果：

✅ 并发安全：数据库自动处理锁
✅ 查询高效：索引支持复杂查询
✅ 数据可靠：事务保证一致性
✅ 易于扩展：轻松支持分类、标签、优先级

1.2 数据库的三大价值

价值	说明	案例
结构化存储	用表结构组织数据，避免文件混乱	用户表、订单表、日志表
高效查询	SQL 支持复杂查询和聚合	"查上周未完成的待办"
数据一致性	事务保证操作要么全成功、要么全失败	转账操作不会丢钱

二、数据库选型：SQLite vs PostgreSQL

2.1 我的选择标准

给 AI 配数据库，我主要看四点：

标准	说明	权重
部署简单	AI 能自己初始化，不需要 DBA	🔴 必须
权限控制	能限制 AI 只能访问特定表	🔴 必须
性能足够	支持并发查询，不卡顿	🟡 推荐
易于备份	能自动备份，防止数据丢失	🟡 推荐

2.2 SQLite：个人项目首选

优势：

✅ 零配置：一个文件就是数据库
✅ 无依赖：Python 内置 sqlite3 模块
✅ 易备份：直接复制 .db 文件即可
✅ 性能好：单文件、低并发场景足够用

劣势：

❌ 并发写入有锁（但 AI 场景通常单线程）
❌ 不支持复杂权限（但可以用应用层控制）

适用场景：

个人 AI 助手
小型团队工具
原型验证阶段

2.3 PostgreSQL：生产环境首选

优势：

✅ 并发能力强：支持多用户同时写入
✅ 权限精细：可以控制到表、列级别
✅ 功能丰富：支持 JSON、全文搜索、地理信息等
✅ 生态完善：备份、监控、高可用方案成熟

劣势：

❌ 部署复杂：需要单独安装服务
❌ 维护成本：需要 DBA 或运维知识

适用场景：

多用户 AI 系统
生产环境部署
需要高可用和灾备

2.4 我的建议

分阶段选型：

阶段 1（个人使用）：SQLite - 快速验证想法 - 零运维成本 - 数据量 < 10GB 阶段 2（团队使用）：SQLite + 定期备份 - 支持 3-5 人小团队 - 每天自动备份到云存储 - 数据量 < 50GB 阶段 3（生产环境）：PostgreSQL - 支持多用户并发 - 主从复制 + 自动故障转移 - 数据量无上限

这篇先用 SQLite 演示，PostgreSQL 的配置在第⑨篇生产环境部署时再详细讲。

三、SQLite 实战：给 AI 配个"记事本"

3.1 初始化数据库

Python import sqlite3 from datetime import datetime # 1. 创建数据库连接 db_path = "/workspace/ai_assistant.db" conn = sqlite3.connect(db_path) # 2. 创建表结构 conn.executescript(""" -- 待办事项表 CREATE TABLE IF NOT EXISTS todos ( id INTEGER PRIMARY KEY AUTOINCREMENT, content TEXT NOT NULL, status TEXT DEFAULT 'pending',-- pending, completed, cancelled priority INTEGER DEFAULT 0,-- 0-5，数字越大优先级越高 due_date TEXT,-- ISO 格式：2026-04-08T15:00:00 created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ); -- 用户表 CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, email TEXT UNIQUE, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); -- 日志表 CREATE TABLE IF NOT EXISTS logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, level TEXT,-- INFO, WARNING, ERROR message TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); -- 创建索引（加速查询） CREATE INDEX IF NOT EXISTS idx_todos_status ON todos(status); CREATE INDEX IF NOT EXISTS idx_todos_due_date ON todos(due_date); CREATE INDEX IF NOT EXISTS idx_logs_level ON logs(level); """) conn.commit()

3.2 在 OpenClaw 智能体中使用

Python @agent class PersonalAssistant: """个人助手智能体 - 带数据库支持""" def __init__(self): self.db_path = "/workspace/ai_assistant.db" self.conn = sqlite3.connect(self.db_path) self.conn.row_factory = sqlite3.Row# 返回字典格式 async def add_todo(self, content: str, due_date: str = None, priority: int = 0): """添加待办事项""" cursor = self.conn.cursor() cursor.execute( """INSERT INTO todos (content, due_date, priority) VALUES (?, ?, ?)""", [content, due_date, priority] ) self.conn.commit() return cursor.lastrowid# 返回新插入的 ID async def get_todos(self, status: str = None, limit: int = 10): """查询待办事项""" cursor = self.conn.cursor() if status: cursor.execute( """SELECT * FROM todos WHERE status = ? ORDER BY priority DESC, due_date ASC LIMIT ?""", [status, limit] ) else: cursor.execute( """SELECT * FROM todos ORDER BY priority DESC, due_date ASC LIMIT ?""", [limit] ) rows = cursor.fetchall() return [dict(row) for row in rows] async def complete_todo(self, todo_id: int): """标记待办为完成""" cursor = self.conn.cursor() cursor.execute( """UPDATE todos SET status = 'completed', updated_at = ? WHERE id = ?""", [datetime.now().isoformat(), todo_id] ) self.conn.commit() return cursor.rowcount > 0 async def log_event(self, level: str, message: str): """记录日志""" cursor = self.conn.cursor() cursor.execute( """INSERT INTO logs (level, message) VALUES (?, ?)""", [level, message] ) self.conn.commit()

3.3 智能体调用示例

Python # 初始化智能体 assistant = PersonalAssistant() # 添加待办 todo_id = await assistant.add_todo( content="下午 3 点开会", due_date="2026-04-08T15:00:00", priority=3 ) print(f"待办已创建，ID: {todo_id}") # 查询所有待办 todos = await assistant.get_todos(limit=20) for todo in todos: print(f"[{todo['status']}] {todo['content']} (优先级：{todo['priority']})") # 标记为完成 await assistant.complete_todo(todo_id) # 记录日志 await assistant.log_event("INFO", f"用户完成了待办 #{todo_id}")

四、安全设计：防止 AI"乱动"数据库

4.1 风险场景

给 AI 数据库权限，就像给小孩一把刀——用得好能切菜，用不好会伤手。

可能的风险：

SQL -- 风险 1：AI 误删数据 DELETE FROM todos;-- 没有 WHERE 条件，清空全表！ -- 风险 2：AI 泄露敏感数据 SELECT * FROM users;-- 查询所有用户信息 -- 风险 3：SQL 注入攻击（如果 AI 拼接用户输入） user_input = "'; DROP TABLE todos; --" cursor.execute(f"SELECT * FROM todos WHERE content = '{user_input}'") -- 实际执行：SELECT * FROM todos WHERE content = ''; DROP TABLE todos; --'

4.2 三层防护机制

参考命令执行的安全设计，数据库也要三层防护：

┌─────────────────────────────────────┐│   第一层：SQL 白名单（允许的操作）       │├─────────────────────────────────────┤│   第二层：参数化查询（防止注入）         │├─────────────────────────────────────┤│   第三层：权限隔离（限制访问范围）       │└─────────────────────────────────────┘

第一层：SQL 白名单

只允许 AI 执行预定义的 SQL 操作：

Python class SafeDatabase: """安全数据库包装器""" def __init__(self, db_path: str): self.conn = sqlite3.connect(db_path) # 定义允许的 SQL 模板 self.allowed_queries = { "insert_todo": """ INSERT INTO todos (content, due_date, priority) VALUES (?, ?, ?) """, "select_todos": """ SELECT * FROM todos WHERE status = ? ORDER BY priority DESC LIMIT ? """, "update_todo_status": """ UPDATE todos SET status = ?, updated_at = ? WHERE id = ? """, "insert_log": """ INSERT INTO logs (level, message) VALUES (?, ?) """, # 注意：没有 DELETE 操作！ } async def execute(self, query_name: str, params: list): """执行预定义的 SQL 查询""" if query_name not in self.allowed_queries: raise PermissionError(f"不允许的查询：{query_name}") sql = self.allowed_queries[query_name] cursor = self.conn.cursor() cursor.execute(sql, params)# 参数化查询 self.conn.commit() return cursor.fetchall()

效果：

✅ AI 只能执行预定义的查询
✅ 无法执行 DROP TABLE、DELETE 等危险操作
✅ 无法访问未授权的表

第二层：参数化查询

永远不要拼接 SQL：

Python # ❌ 错误做法：字符串拼接（SQL 注入漏洞） user_input = request.get("content") cursor.execute(f"SELECT * FROM todos WHERE content = '{user_input}'") # ✅ 正确做法：参数化查询 user_input = request.get("content") cursor.execute("SELECT * FROM todos WHERE content = ?", [user_input])

为什么参数化能防注入？

用户输入：'; DROP TABLE todos; -- 拼接方式： SELECT * FROM todos WHERE content = ''; DROP TABLE todos; --' → 实际执行两条语句，表被删了！参数化方式： SELECT * FROM todos WHERE content = ? 参数：["'; DROP TABLE todos; --"] → 整个输入被当作一个字符串值，不会执行

第三层：权限隔离

用不同数据库文件隔离敏感数据：

Python # 主数据库：AI 可以读写 main_db = SafeDatabase("/workspace/ai_main.db") # 表：todos, logs, settings # 敏感数据库：AI 只能读，不能写 sensitive_db = SafeDatabase("/workspace/ai_sensitive.db") # 表：users, api_keys, credentials # 权限：只允许 SELECT，不允许 INSERT/UPDATE/DELETE

或者用视图限制访问范围：

SQL -- 创建视图，只暴露部分字段 CREATE VIEW todos_public AS SELECT id, content, status, due_date-- 不暴露 priority 和 created_at FROM todos; -- AI 只能查询视图，不能直接访问原表 GRANT SELECT ON todos_public TO ai_user;

五、实战案例：构建智能待办助手

5.1 需求背景

我们需要一个智能体，能：

听懂用户的自然语言（"明天下午 3 点开会"）
解析并提取关键信息（时间、内容、优先级）
存入数据库
到点提醒
支持查询和更新

5.2 智能体设计

Python @agent class TodoAssistant: """智能待办助手""" def __init__(self): self.db = SafeDatabase("/workspace/todo_assistant.db") self.init_database() def init_database(self): """初始化数据库""" self.db.execute("init_schema", []) async def parse_and_add(self, user_input: str): """解析自然语言并添加待办""" # 1. 用 AI 解析用户输入 parsed = await self.llm.parse_todo(user_input) # 返回：{"content": "开会", "due_date": "2026-04-09T15:00:00", "priority": 3} # 2. 存入数据库 todo_id = await self.db.execute("insert_todo", [ parsed["content"], parsed["due_date"], parsed["priority"] ]) # 3. 确认 return f"已创建待办 #{todo_id}：{parsed['content']}，截止时间：{parsed['due_date']}" async def list_todos(self, filter: str = "all"): """列出待办""" if filter == "pending": rows = await self.db.execute("select_todos", ["pending", 20]) elif filter == "completed": rows = await self.db.execute("select_todos", ["completed", 20]) else: rows = await self.db.execute("select_todos", [None, 20]) # 格式化输出 output = [] for row in rows: icon = "✅" if row["status"] == "completed" else "⏳" output.append(f"{icon} [{row['id']}] {row['content']} (截止：{row['due_date']})") return "\n".join(output) if output else "暂无待办" async def complete(self, todo_id: int): """标记完成""" success = await self.db.execute("update_todo_status", [ "completed", datetime.now().isoformat(), todo_id ]) if success: return f"待办 #{todo_id} 已标记为完成" else: return f"未找到待办 #{todo_id}" async def check_and_remind(self): """检查并发送提醒""" now = datetime.now().isoformat() # 查询即将到期的待办 rows = await self.db.execute("select_todos_by_due_date", [now, 10]) for row in rows: await self.send_message( "user", f"⏰ 提醒：待办「{row['content']}」即将到期（{row['due_date']}）" )

5.3 使用示例

用户对话：

用户：记个待办，明天下午 3 点开会 AI：已创建待办 #1：开会，截止时间：2026-04-09T15:00:00 用户：再记一个，本周五前提交周报，优先级高 AI：已创建待办 #2：提交周报，截止时间：2026-04-12T23:59:59，优先级：4 用户：我有哪些待办？ AI： ⏳ [1] 开会 (截止：2026-04-09T15:00:00) ⏳ [2] 提交周报 (截止：2026-04-12T23:59:59) 用户：#1 完成了 AI：待办 #1 已标记为完成用户：查看未完成的待办 AI： ⏳ [2] 提交周报 (截止：2026-04-12T23:59:59)

六、踩坑记录：我们交过的"学费"

6.1 坑 1：数据库文件被锁

问题：

Python # 错误做法：多个连接同时写入 conn1 = sqlite3.connect("ai.db") conn2 = sqlite3.connect("ai.db") # 第二个连接会报错：database is locked

解决方案：

Python # 正确做法：单例模式，复用连接 class DatabaseSingleton: _instance = None def __new__(cls, db_path: str): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance.conn = sqlite3.connect(db_path, check_same_thread=False) return cls._instance

6.2 坑 2：时区问题导致查询错误

问题：

Python # 错误做法：混用时区 due_date = "2026-04-09T15:00:00"# 用户输入（北京时间） now = datetime.utcnow().isoformat()# 当前时间（UTC） # 比较时差了 8 小时！

解决方案：

Python # 正确做法：统一时区 from datetime import datetime, timezone, timedelta # 全部用 UTC 存储 due_date = "2026-04-09T07:00:00Z"# 北京时间 15:00 = UTC 07:00 now = datetime.now(timezone.utc).isoformat() # 或者全部用本地时区 tz = timezone(timedelta(hours=8)) due_date = "2026-04-09T15:00:00+08:00" now = datetime.now(tz).isoformat()

6.3 坑 3：数据库文件丢失

问题：

AI 误操作删除了 /workspace/ai.db 文件 → 所有数据丢失

解决方案：

YAML # 配置自动备份 backup: enabled: true source: "/workspace/ai.db" destination: "/backup/ai.db" schedule: "0 2 * * *"# 每天凌晨 2 点 retention_days: 7

6.4 坑 4：查询性能随数据量下降

问题：

Python # 错误做法：没有索引 cursor.execute("SELECT * FROM todos WHERE status = ?", ["pending"]) # 数据量少时没问题，超过 10000 条后查询变慢

解决方案：

SQL -- 创建索引 CREATE INDEX idx_todos_status ON todos(status); CREATE INDEX idx_todos_due_date ON todos(due_date); -- 查询时 EXPLAIN 分析 EXPLAIN QUERY PLAN SELECT * FROM todos WHERE status = ?; -- 确认使用了索引

七、最佳实践清单

7.1 数据库设计

实践	说明	优先级
用参数化查询	永远不要拼接 SQL	🔴 必须
预定义 SQL 模板	限制 AI 只能执行允许的查询	🔴 必须
创建必要索引	加速常用查询	🔴 必须
统一时区	全部用 UTC 或本地时区	🔴 必须
定期备份	防止数据丢失	🟡 推荐
用事务包装	保证数据一致性	🟡 推荐

7.2 安全配置模板

Python class SafeDatabase: """安全数据库配置模板""" def __init__(self, db_path: str): self.db_path = db_path self.conn = sqlite3.connect(db_path) self.conn.row_factory = sqlite3.Row # 1. 允许的 SQL 模板 self.allowed_queries = { "insert": "INSERT INTO {table} ({columns}) VALUES ({placeholders})", "select": "SELECT {columns} FROM {table} WHERE {conditions} LIMIT ?", "update": "UPDATE {table} SET {updates} WHERE id = ?", # 注意：没有 DELETE！ } # 2. 允许的表 self.allowed_tables = ["todos", "logs", "users"] # 3. 禁止的字段（敏感信息） self.denied_columns = ["password", "api_key", "token"] async def execute(self, query_type: str, table: str, params: list): """安全执行查询""" # 验证表名 if table not in self.allowed_tables: raise PermissionError(f"不允许访问表：{table}") # 验证查询类型 if query_type not in self.allowed_queries: raise PermissionError(f"不允许的查询类型：{query_type}") # 执行参数化查询 sql = self.allowed_queries[query_type].format(table=table, ...) cursor = self.conn.cursor() cursor.execute(sql, params) self.conn.commit() return cursor.fetchall()

7.3 备份策略

YAML # 备份配置 backup: # SQLite sqlite: source: "/workspace/ai.db" destination: "/backup/ai.db" schedule: "0 2 * * *"# 每天凌晨 2 点 compression: true# 压缩备份 encryption: true# 加密备份（如果有敏感数据） # PostgreSQL postgresql: host: "localhost" database: "ai_assistant" backup_command: "pg_dump -Fc" destination: "/backup/ai_pg.dump" schedule: "0 3 * * *" retention_days: 7

八、进阶技巧

8.1 用 WAL 模式提升并发性能

Python # 启用 WAL（Write-Ahead Logging）模式 conn.execute("PRAGMA journal_mode=WAL") # 效果：读写不阻塞，适合多用户场景

8.2 用连接池管理数据库连接

Python from sqlite3 import connect from contextlib import contextmanager class DatabasePool: def __init__(self, db_path: str, pool_size: int = 5): self.db_path = db_path self.pool = [connect(db_path) for _ in range(pool_size)] @contextmanager def get_connection(self): conn = self.pool.pop() try: yield conn finally: self.pool.append(conn) # 使用 pool = DatabasePool("ai.db") with pool.get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT * FROM todos")

8.3 用触发器自动维护数据

SQL -- 自动更新 updated_at 字段 CREATE TRIGGER update_todos_timestamp AFTER UPDATE ON todos BEGIN UPDATE todos SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; -- 自动清理过期数据 CREATE TRIGGER cleanup_old_logs AFTER INSERT ON logs BEGIN DELETE FROM logs WHERE created_at < datetime('now', '-30 days'); END;

九、常见问题

Q1: AI 能否直接执行用户输入的 SQL？

答：绝对不行！

即使用户要求，也不要让 AI 执行任意 SQL：

Python # ❌ 危险做法 user_sql = request.get("sql") cursor.execute(user_sql) # ✅ 正确做法：用预定义接口 action = request.get("action") if action == "add_todo": cursor.execute("INSERT INTO todos ...", params) elif action == "list_todos": cursor.execute("SELECT * FROM todos ...", params)

Q2: 如何处理数据库迁移（表结构变更）？

答：用迁移脚本：

Python # migrations/001_add_priority.py def migrate(conn): cursor = conn.cursor() cursor.execute("ALTER TABLE todos ADD COLUMN priority INTEGER DEFAULT 0") conn.commit() # 在智能体启动时执行 def run_migrations(): for migration_file in sorted(os.listdir("migrations")): module = importlib.import_module(f"migrations.{migration_file}") module.migrate(conn)

Q3: SQLite 数据量上限是多少？

答：

理论上限：140TB
实际建议：< 10GB（单文件）
超过 10GB：考虑 PostgreSQL 或分库

十、本章小结

核心要点

AI 需要数据库：文件存储不适合结构化数据
SQLite 适合个人项目：零配置、易备份、性能足够
三层防护机制：SQL 白名单 → 参数化查询 → 权限隔离
永远不要拼接 SQL：参数化查询防注入
定期备份：防止数据丢失

安全检查清单

在给 AI 配数据库前，确认：
使用参数化查询（不拼接 SQL）
预定义 SQL 模板（限制操作范围）
创建必要索引（加速查询）
统一时区（避免时间错误）
配置自动备份（防止数据丢失）
敏感数据隔离（不同数据库或视图）

十一、下一章预告

第⑧篇：《让 AI 能"上网"：Web 服务与 API 集成》

数据库让 AI 能"记住"，但要让 AI 真正"干活"，还需要连接外部服务：

如何让 AI 调用 HTTP API？
如何安全地存储 API 密钥？
如何设计重试和超时机制？
如何集成微信、飞书、GitHub 等第三方服务？

敬请期待！

系列文章导航

篇序	主题	状态
①	从原理到实战	✅ 已发布
②	记忆系统的艺术	✅ 已发布
③	子代理协作模式	✅ 已发布
④	工具系统详解	✅ 已发布
⑤	文件操作实战指南	✅ 已发布
⑥	命令执行与安全控制	✅ 已发布
⑦	数据库与持久化存储	📝 本篇
⑧	Web 服务与 API 集成	⏳ 待写
⑨	生产环境部署指南	⏳ 待写
⑩	AI Coding 团队落地实战	⏳ 待写

关于本系列

这是 OpenClaw 养虾教程系列的第⑦篇。我会在每周二、周五更新，用 5 周时间完成全部 10 篇。

关注我，获取：

每周 2 篇技术干货
AI 工程化落地实战经验
可复用的代码和配置模板

本文基于真实项目经验撰写，所有配置和代码均已脱敏处理。欢迎转发，转载请注明出处。