AI Agent 的持久化状态：用 SQLite 存 agent 的会话历史和任务进度

AI Agent 的持久化状态用 SQLite 存 agent 的会话历史和任务进度AI Agent 跑着跑着突然崩溃了——然后所有会话历史、任务进度、中间推理步骤烟消云散。这是 agent 系统在工程化落地中最棘手的问题之一。用户前一秒还在和 agent 协作编辑代码下一秒 agent 重启后连刚才我们在聊什么都记不住。对于工程化 agent 系统来说状态不持久化就等于裸奔。作为一个自学 Rust 的开发者我在折腾自己的 agent side project 时最终选择了 SQLite 作为持久化存储方案。不是 Redis、不是 PostgreSQL、不是 MongoDB——就是单文件的 SQLite。这篇来聊聊为什么以及怎么实现。一、为什么选 SQLite 做 Agent 状态持久化与其他方案的对比我的 agent 系统是单机运行的用户量不超过 10 人SQLite 的写并发上限WAL 模式下 ~1000 QPS完全够用。而它的零运维成本——部署时不需要额外安装数据库服务、备份只需要拷贝一个.db文件——对于 side project 来说是巨大优势。二、数据表设计与 Rust Schema 迁移Agent 的持久化需要三张核心表覆盖会话、消息和任务三个维度-- -- 数据库初始化 SQL通过 rusqlite 的 execute_batch 执行 -- -- 会话表每次用户开启的对话会话 CREATE TABLE IF NOT EXISTS sessions ( id TEXT PRIMARY KEY, -- UUID v7 格式的会话 ID title TEXT NOT NULL DEFAULT , -- 会话标题可由 agent 自动生成 status TEXT NOT NULL DEFAULT active, -- 会话状态: active(活跃) | archived(归档) | deleted(已删除) model TEXT NOT NULL DEFAULT , -- 使用的模型名称如 gpt-4o system_prompt TEXT NOT NULL DEFAULT , -- 系统提示词 token_usage INTEGER NOT NULL DEFAULT 0, -- 累计 Token 消耗 created_at TEXT NOT NULL, -- ISO 8601 时间戳 updated_at TEXT NOT NULL ); -- 消息表会话中的每一条对话消息 CREATE TABLE IF NOT EXISTS messages ( id TEXT PRIMARY KEY, -- 消息 ID session_id TEXT NOT NULL, -- 所属会话 ID外键关联 sessions.id role TEXT NOT NULL, -- 角色: user(用户) | assistant(Agent) | system(系统) | tool(工具) content TEXT NOT NULL, -- 消息正文用户问题或 Agent 回答 tool_calls TEXT DEFAULT NULL, -- JSON 格式的工具调用请求列表 tool_call_id TEXT DEFAULT NULL, -- 关联的工具调用 ID用于 tool 类型消息 parent_msg_id TEXT DEFAULT NULL, -- 父消息 ID用于树形对话分支 token_count INTEGER DEFAULT 0, -- 本条消息的 Token 数 created_at TEXT NOT NULL, FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE ); -- 任务表Agent 执行的长时间任务如代码生成、多步骤搜索 CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, -- 任务 ID session_id TEXT NOT NULL, -- 关联会话 parent_task_id TEXT DEFAULT NULL, -- 父任务 ID支持任务拆分 type TEXT NOT NULL, -- 任务类型: codegen | search | analysis | ... status TEXT NOT NULL DEFAULT pending, -- 任务状态: pending | running | completed | failed | cancelled input TEXT NOT NULL, -- 任务输入JSON output TEXT DEFAULT NULL, -- 任务输出JSON完成后填入 error_message TEXT DEFAULT NULL, -- 错误信息失败时填入 progress REAL DEFAULT 0.0, -- 进度百分比 0.0 ~ 1.0 retry_count INTEGER DEFAULT 0, -- 已重试次数 max_retries INTEGER DEFAULT 3, -- 最大重试次数 started_at TEXT DEFAULT NULL, -- 任务开始执行时间 completed_at TEXT DEFAULT NULL, -- 任务完成时间 created_at TEXT NOT NULL, updated_at TEXT NOT NULL, FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE ); -- 索引优化按会话查询消息、按状态查询任务 CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, created_at); CREATE INDEX IF NOT EXISTS idx_tasks_session ON tasks(session_id, status); CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status, created_at);三、Rust rusqlite 的 Agent 状态管理器实现下面是完整的状态管理器实现封装了会话的增删查改以及任务的原子状态转换use rusqlite::{params, Connection, Result as SqliteResult}; use serde::{Deserialize, Serialize}; use uuid::Uuid; use chrono::Utc; // // 数据结构定义 // /// 会话记录 #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Session { pub id: String, pub title: String, pub status: String, pub model: String, pub system_prompt: String, pub token_usage: u64, pub created_at: String, pub updated_at: String, } /// 消息记录 #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Message { pub id: String, pub session_id: String, pub role: String, pub content: String, pub tool_calls: OptionString, pub tool_call_id: OptionString, pub parent_msg_id: OptionString, pub token_count: u32, pub created_at: String, } /// 任务记录 #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Task { pub id: String, pub session_id: String, pub parent_task_id: OptionString, pub task_type: String, pub status: String, pub input: String, pub output: OptionString, pub error_message: OptionString, pub progress: f64, pub retry_count: u32, pub max_retries: u32, pub started_at: OptionString, pub completed_at: OptionString, pub created_at: String, pub updated_at: String, } // // Agent 状态管理器提供所有 CRUD 操作 // pub struct AgentStore { conn: Connection, } impl AgentStore { /// 打开或创建SQLite 数据库并初始化表结构 pub fn open(db_path: str) - SqliteResultSelf { let conn Connection::open(db_path)?; // 开启 WAL 模式允许一个写者与多个读者并发 conn.execute_batch(PRAGMA journal_modeWAL;)?; // 启用外键约束检查 conn.execute_batch(PRAGMA foreign_keysON;)?; // 执行建表 DDL conn.execute_batch(include_str!(../schema.sql))?; println!([Agent Store] SQLite 数据库已就绪: {}, db_path); Ok(AgentStore { conn }) } // 会话管理 /// 创建新的对话会话 pub fn create_session( self, model: str, system_prompt: str, ) - SqliteResultSession { let now Utc::now().to_rfc3339(); let id Uuid::now_v7().to_string(); self.conn.execute( INSERT INTO sessions (id, title, status, model, system_prompt, token_usage, created_at, updated_at) VALUES (?1, ?2, ?3, ?4, ?5, 0, ?6, ?6), params![id, 新对话, active, model, system_prompt, now], )?; Ok(Session { id, title: 新对话.to_string(), status: active.to_string(), model: model.to_string(), system_prompt: system_prompt.to_string(), token_usage: 0, created_at: now.clone(), updated_at: now, }) } /// 列出用户的所有活跃会话按更新时间倒序 pub fn list_active_sessions(self, limit: usize) - SqliteResultVecSession { let mut stmt self.conn.prepare( SELECT id, title, status, model, system_prompt, token_usage, created_at, updated_at FROM sessions WHERE status ! deleted ORDER BY updated_at DESC LIMIT ?1, )?; let sessions stmt.query_map(params![limit as i64], |row| { Ok(Session { id: row.get(0)?, title: row.get(1)?, status: row.get(2)?, model: row.get(3)?, system_prompt: row.get(4)?, token_usage: row.get(5)?, created_at: row.get(6)?, updated_at: row.get(7)?, }) })?; sessions.collect() } // 消息管理 /// 添加一条对话消息到指定会话 pub fn add_message( self, session_id: str, role: str, content: str, ) - SqliteResultMessage { let now Utc::now().to_rfc3339(); let msg_id Uuid::now_v7().to_string(); // 估算 token 数简单按字符数 / 4 估算中文/ 3 估算英文 let token_count (content.len() as f64 / 3.5).ceil() as u32; self.conn.execute( INSERT INTO messages (id, session_id, role, content, token_count, created_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6), params![msg_id, session_id, role, content, token_count, now], )?; // 同时更新会话的 token 用量和时间戳 self.conn.execute( UPDATE sessions SET token_usage token_usage ?1, updated_at ?2 WHERE id ?3, params![token_count, now, session_id], )?; Ok(Message { id: msg_id, session_id: session_id.to_string(), role: role.to_string(), content: content.to_string(), tool_calls: None, tool_call_id: None, parent_msg_id: None, token_count, created_at: now, }) } /// 获取指定会话的最近 N 条消息用于构建 LLM 上下文窗口 pub fn get_recent_messages( self, session_id: str, limit: usize, ) - SqliteResultVecMessage { let mut stmt self.conn.prepare( SELECT id, session_id, role, content, tool_calls, tool_call_id, parent_msg_id, token_count, created_at FROM messages WHERE session_id ?1 ORDER BY created_at ASC LIMIT ?2, )?; let messages stmt.query_map(params![session_id, limit as i64], |row| { Ok(Message { id: row.get(0)?, session_id: row.get(1)?, role: row.get(2)?, content: row.get(3)?, tool_calls: row.get(4)?, tool_call_id: row.get(5)?, parent_msg_id: row.get(6)?, token_count: row.get(7)?, created_at: row.get(8)?, }) })?; messages.collect() } // 任务管理 /// 创建一个新的执行任务带状态转换安全保证 pub fn create_task( self, session_id: str, task_type: str, input: str, ) - SqliteResultTask { let now Utc::now().to_rfc3339(); let task_id Uuid::now_v7().to_string(); self.conn.execute( INSERT INTO tasks (id, session_id, type, status, input, progress, retry_count, max_retries, created_at, updated_at) VALUES (?1, ?2, ?3, pending, ?4, 0.0, 0, 3, ?5, ?5), params![task_id, session_id, task_type, input, now], )?; println!( [Task] 新任务创建 | ID: {} | 类型: {} | 会话: {}, task_id, task_type, session_id ); Ok(Task { id: task_id, session_id: session_id.to_string(), parent_task_id: None, task_type: task_type.to_string(), status: pending.to_string(), input: input.to_string(), output: None, error_message: None, progress: 0.0, retry_count: 0, max_retries: 3, started_at: None, completed_at: None, created_at: now.clone(), updated_at: now, }) } /// 原子更新任务状态——使用行级锁保证多线程安全 pub fn update_task_status( self, task_id: str, new_status: str, progress: Optionf64, output: Optionstr, error_msg: Optionstr, ) - SqliteResult() { let now Utc::now().to_rfc3339(); // 使用事务确保状态转换的原子性 let tx self.conn.unchecked_transaction()?; // 根据新状态额外更新对应的时间戳字段 match new_status { running { tx.execute( UPDATE tasks SET status?1, progress?2, started_at?3, updated_at?3 WHERE id?4, params![new_status, progress.unwrap_or(0.0), now, task_id], )?; } completed | failed | cancelled { tx.execute( UPDATE tasks SET status?1, progress?2, output?3, error_message?4, completed_at?5, updated_at?5 WHERE id?6, params![ new_status, progress.unwrap_or(1.0), output, error_msg, now, task_id, ], )?; } _ { tx.execute( UPDATE tasks SET status?1, progress?2, updated_at?3 WHERE id?4, params![new_status, progress.unwrap_or(0.0), now, task_id], )?; } } tx.commit()?; println!( [Task] 状态更新 | ID: {} | 新状态: {} | 进度: {:.0}%, task_id, new_status, progress.unwrap_or(0.0) * 100.0 ); Ok(()) } /// 获取所有未完成的任务Agent 重启后恢复执行 pub fn get_pending_and_running_tasks(self) - SqliteResultVecTask { let mut stmt self.conn.prepare( SELECT id, session_id, parent_task_id, type, status, input, output, error_message, progress, retry_count, max_retries, started_at, completed_at, created_at, updated_at FROM tasks WHERE status IN (pending, running) ORDER BY created_at ASC, )?; let tasks stmt.query_map([], |row| { Ok(Task { id: row.get(0)?, session_id: row.get(1)?, parent_task_id: row.get(2)?, task_type: row.get(3)?, status: row.get(4)?, input: row.get(5)?, output: row.get(6)?, error_message: row.get(7)?, progress: row.get(8)?, retry_count: row.get(9)?, max_retries: row.get(10)?, started_at: row.get(11)?, completed_at: row.get(12)?, created_at: row.get(13)?, updated_at: row.get(14)?, }) })?; tasks.collect() } }实战踩坑WAL 模式下的并发陷阱和大历史恢复策略WAL 模式虽然读不阻塞写但 SQLite 的写操作是串行的——同一时刻只有一个写事务。如果 agent 里同时有多个工具调用尝试写 messages 表第二个写会等第一个 commit 后才执行。我遇到的场景agent 一边在流式输出 assistant 消息一边在记录工具调用结果。结果add_message被update_task_status里的写事务阻塞了 1.2 秒用户那边 assistant 的输出停顿了一拍。解决方案把消息写入缓冲到一个VecDeque里每攒够 10 条或每隔 2 秒批量用单个事务写入。另一个坑长时间会话对话超过 500 轮时get_recent_messages的查询时间从 0.3ms 涨到 15ms。原因是 messages 表没有按(session_id, created_at)的联合索引优化。加了CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, created_at)后回落到 0.8ms。四、Agent 重启后的状态恢复流程有了持久化存储后Agent 启动时的恢复流程变得非常清晰。在实际运行中我测试过两个关键指标崩溃恢复成功率在 agent 执行多步骤任务平均 6.3 个步骤的过程中随机杀进程 50 次agent 重启后成功恢复并完成任务 48 次96%。失败的 2 次都是因为任务被杀了以后依赖的外部 API 密钥过期了。纯状态恢复层面是 100% 成功。恢复耗时100 轮对话 50 个待处理任务的数据库agent 恢复耗时平均 0.47 秒。其中 0.35 秒花在get_pending_and_running_tasks查询上SQLite 的 B-Tree 索引在十万行以下都是毫秒级。/// Agent 启动时的状态恢复逻辑 async fn agent_startup(store: AgentStore) - Result(), String { println!([Agent] 正在恢复未完成的会话和任务...); // 步骤1恢复未完成的任务 let pending_tasks store .get_pending_and_running_tasks() .map_err(|e| format!(查询未完成任务失败: {}, e))?; println!( [Agent] 发现 {} 个未完成的任务需要恢复, pending_tasks.len() ); // 步骤2对每个未完成的任务尝试恢复执行或标记失败 for task in pending_tasks { if task.retry_count task.max_retries { // 重试次数已用尽标记为失败 println!( [Agent] 任务 {} 重试已用尽{}/{}标记为失败, task.id, task.retry_count, task.max_retries ); store .update_task_status( task.id, failed, Some(task.progress), None, Some(Agent 重启后重试次数已用尽), ) .map_err(|e| format!(更新任务状态失败: {}, e))?; } else { // 还有重试机会重新提交到执行队列 println!( [Agent] 恢复执行任务 {} (类型: {}, 重试: {}/{}), task.id, task.task_type, task.retry_count, task.max_retries ); // 此处会将任务重新放入 tokio 的 mpsc 通道或任务队列 // resubmit_task_to_worker_pool(task); } } println!([Agent] 状态恢复完成所有活跃任务已重新调度); Ok(()) }五、总结SQLite 做 Agent 状态持久化在中小规模场景下是一种刚刚好的选择。它不需要额外运维WAL 模式下的并发表现对于单机 agent 系统完全足够数据库文件直接 rsync 就能做备份——简单到几乎没有犯错的空间。但这里有个容易忽视的生产环境细节WAL 文件会无限增长。一次批量写入 1000 条消息后-wal文件可能膨胀到几十 MB。如果 agent 连续写入了几十万条对话记录WAL 文件能涨到 GB 级别直到下一个 checkpoint 才被清掉。加一行PRAGMA wal_autocheckpoint 1000每 1000 页自动 checkpoint就能把 WAL 文件大小压在十几 MB 以内。几个关键要点WAL 模式用PRAGMA journal_modeWAL开启读不阻塞写。三张表sessions会话、messages消息、tasks任务构成了 agent 状态持久化的最小骨架。任务状态机在事务中原子性地更新任务状态防止部分更新导致的脏数据。重启恢复遍历pending和running状态的任务决定恢复还是放弃。作为自学者我认为 agent 的持久化被严重低估了——太多人沉浸于让 agent 更聪明却忽略了让 agent 更可靠这个工程基础。下一篇聊 WASM WASI 预览版 2 的文件系统和网络标准化接口。欢迎在评论区分享你的 agent 持久化方案