SQLite 做 AI 工作流持久化，够用吗？

SELECT step_id, status, output, created_at
FROM workflow_steps
WHERE execution_id = 'abc-123'
ORDER BY step_seq;

pip install sqlite3-utils  # 可选，方便初始化；标准库自带 sqlite3

import sqlite3
import json
import time
from pathlib import Path

DB_PATH = Path("agent_state.db")

def init_db():
    """创建步骤状态表，只需运行一次。"""
    conn = sqlite3.connect(str(DB_PATH))
    conn.execute("PRAGMA journal_mode=WAL")  # 开启 WAL，崩溃安全
    conn.execute("""
        CREATE TABLE IF NOT EXISTS workflow_steps (
            execution_id  TEXT NOT NULL,
            step_seq      INTEGER NOT NULL,
            step_id       TEXT NOT NULL,
            status        TEXT NOT NULL DEFAULT 'pending',
            input_json    TEXT,
            output_json   TEXT,
            error_msg     TEXT,
            created_at    REAL NOT NULL,
            updated_at    REAL NOT NULL,
            PRIMARY KEY (execution_id, step_seq)
        )
    """)
    conn.commit()
    conn.close()

def start_step(exec_id: str, seq: int, step_id: str, input_data: dict):
    """步骤开始前写入 running 状态——这是 checkpoint。"""
    conn = sqlite3.connect(str(DB_PATH))
    now = time.time()
    conn.execute("""
        INSERT INTO workflow_steps (execution_id, step_seq, step_id,
                                    status, input_json, created_at, updated_at)
        VALUES (?, ?, ?, 'running', ?, ?, ?)
    """, (exec_id, seq, step_id, json.dumps(input_data), now, now))
    conn.commit()
    conn.close()

def complete_step(exec_id: str, seq: int, output_data: dict):
    """步骤成功完成，更新状态和输出。"""
    conn = sqlite3.connect(str(DB_PATH))
    conn.execute("""
        UPDATE workflow_steps
        SET status = 'completed', output_json = ?, updated_at = ?
        WHERE execution_id = ? AND step_seq = ?
    """, (json.dumps(output_data), time.time(), exec_id, seq))
    conn.commit()
    conn.close()

def fail_step(exec_id: str, seq: int, error: str):
    """步骤失败，记录错误信息。"""
    conn = sqlite3.connect(str(DB_PATH))
    conn.execute("""
        UPDATE workflow_steps
        SET status = 'failed', error_msg = ?, updated_at = ?
        WHERE execution_id = ? AND step_seq = ?
    """, (error, time.time(), exec_id, seq))
    conn.commit()
    conn.close()

def find_last_running(exec_id: str) -> dict | None:
    """崩溃恢复：找到最后一个 running 状态的步骤。"""
    conn = sqlite3.connect(str(DB_PATH))
    row = conn.execute("""
        SELECT step_seq, step_id, input_json
        FROM workflow_steps
        WHERE execution_id = ? AND status = 'running'
        ORDER BY step_seq DESC LIMIT 1
    """, (exec_id,)).fetchone()
    conn.close()
    if row:
        return {"seq": row[0], "step_id": row[1], "input": json.loads(row[2])}
    return None

# ---- 模拟一次 Agent 执行 ----

def run_agent(exec_id: str):
    """带崩溃恢复的 Agent 执行循环。"""
    init_db()

    # 检查是否有未完成的步骤（崩溃恢复场景）
    last_running = find_last_running(exec_id)
    if last_running:
        print(f"恢复执行：从步骤 {last_running['seq']} ({last_running['step_id']}) 重试")
        start_seq = last_running["seq"]
    else:
        start_seq = 0

    steps = [
        ("plan",   {"prompt": "帮我分析这段代码的性能瓶颈"}),
        ("search", {"query": "Python profiling best practices"}),
        ("analyze", {"code": "def process(items): ..."}),
        ("report", {"format": "markdown"}),
    ]

    for i in range(start_seq, len(steps)):
        step_id, input_data = steps[i]
        start_step(exec_id, i, step_id, input_data)

        try:
            # 这里放真实的 LLM / 工具调用逻辑
            # 模拟：假装执行成功
            output = {"result": f"{step_id} 步骤输出（模拟）"}
            complete_step(exec_id, i, output)
            print(f"步骤 {i} ({step_id}) 完成")
        except Exception as e:
            fail_step(exec_id, i, str(e))
            print(f"步骤 {i} ({step_id}) 失败: {e}")
            break  # 失败后停止，下次可从该步恢复

# 运行
run_agent("exec-001")

# 查看完整执行轨迹
conn = sqlite3.connect(str(DB_PATH))
rows = conn.execute("""
    SELECT step_seq, step_id, status, output_json
    FROM workflow_steps WHERE execution_id = 'exec-001'
    ORDER BY step_seq
""").fetchall()
for r in rows:
    print(r)
conn.close()