在 大模型 应用日益普及的今天，如何有效地管理和利用模型的 上下文记忆，成为了提升 AI 智能体 性能的关键。本文将深入探讨一种基于 SQLite 的创新方法，帮助开发者构建更强大、更高效的 AI 智能体，使其在处理复杂任务时能够更好地保持状态、避免重复劳动并实现智能决策。

1. 传统方案的局限性：文件日志的瓶颈

在 AI 智能体 的早期开发阶段，常见的做法是使用日志文件来记录智能体的运行状态和上下文信息。作者最初也采用了这种方法，通过维护 agent-progress.log 文件，智能体可以记录每个阶段的关键信息，例如时间戳、阶段名称、执行的动作、需求描述、解决方案以及优先级等等。

这种方法在一定程度上解决了 上下文记忆 的问题，使得智能体在中断后可以从上次的非错误状态恢复，避免从头开始。然而，随着项目规模的扩大和日志文件体积的增长，这种方式的弊端逐渐显现。具体来说，当日志文件超过 5000 行时，智能体读取和写入日志的效率显著下降，导致整体性能降低。此外，频繁的文件操作也增加了系统的负担，限制了智能体的可扩展性。这种传统的文件日志方法，在处理大型项目和复杂任务时，暴露出明显的瓶颈。因此，寻找一种更高效、更可靠的持久化方案成为了必然的选择。

2. SQLite 的优势：轻量级、便携且功能强大

为了克服文件日志的局限性，作者将 AI 智能体 的 上下文记忆 系统迁移到 SQLite 数据库。SQLite 是一种轻量级的、嵌入式的关系型数据库管理系统，它具有以下显著优势：

• 轻量级和便携性： SQLite 数据库以单个文件的形式存在，易于部署和管理。无需独立的服务器进程，可以直接嵌入到应用程序中，降低了部署复杂性。
• SQL 查询能力： SQLite 支持标准的 SQL 查询语言，开发者可以使用复杂的 SQL 语句来检索和分析 AI 智能体 的运行数据，为智能决策提供支持。
• 插件支持： SQLite 具有良好的可扩展性，可以通过插件扩展其功能，例如全文检索 (FTS) 等。

这些特性使得 SQLite 成为构建 AI 智能体 持久化层的理想选择。特别是 SQLite 的 SQL 查询能力，为智能体提供了强大的数据分析工具，可以帮助智能体更好地理解自身的状态、识别潜在问题并制定相应的解决方案。例如，可以使用 SQL 查询来分析各个阶段的执行时间，找出性能瓶颈；也可以通过关键词搜索来查找特定事件或错误信息。

3. 基于 SQLite 的上下文记忆系统设计

基于 SQLite 的 上下文记忆 系统主要包含以下几个核心表：

• logs 表： 用于记录 AI 智能体 的运行日志，包括时间戳、阶段、消息和元数据。元数据采用 JSON 格式存储，可以灵活地记录各种上下文信息，例如任务 ID、 sprint ID、 ticket ID 等。

  CREATE TABLE IF NOT EXISTS logs (
    id      INTEGER PRIMARY KEY AUTOINCREMENT,
    ts      DATETIME DEFAULT CURRENT_TIMESTAMP,
    stage   TEXT NOT NULL,
    msg     TEXT NOT NULL,
    meta    JSON
  );
  

• memory 表： 用于存储 AI 智能体 的短期记忆，例如配置信息、任务状态等。每个记忆条目都有一个过期时间，可以自动清理过期的记忆。

  CREATE TABLE IF NOT EXISTS memory (
    key         TEXT PRIMARY KEY,
    value       BLOB NOT NULL,
    expires_at  DATETIME NOT NULL
  );
  

• checkpoints 表： 用于存储代码或数据的快照，例如版本控制信息、配置文件等。快照以 BLOB 格式存储，可以保证数据的完整性。

  CREATE TABLE IF NOT EXISTS checkpoints (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    ts        DATETIME DEFAULT CURRENT_TIMESTAMP,
    path      TEXT NOT NULL,
    summary   TEXT,
    snapshot  BLOB
  );
  

• fts 表： 用于实现全文检索功能，可以快速查找日志和快照中的内容。

  CREATE VIRTUAL TABLE IF NOT EXISTS fts USING fts5(content);
  

通过这四个表的协同工作，AI 智能体 可以有效地管理和利用其 上下文记忆，从而实现更智能的行为。

4. 任务管理：让 AI 智能体更专注

任务管理 是提升 AI 智能体 效率的关键环节。通过将任务分解为更小的子任务，并使用 SQLite 记录每个子任务的状态，可以帮助智能体更好地跟踪任务进度、避免重复劳动并实现智能恢复。文章中提供了 task_start 和 task_end 两个辅助函数，用于创建和结束任务。

• task_start 函数： 用于创建一个新的任务，并将任务信息存储到 memory 表中。任务信息包括任务描述、状态、创建时间以及其他元数据。该函数还会生成一个唯一的任务 ID，并将其记录到日志中。
• task_end 函数： 用于标记一个任务完成，并将任务状态更新为 “done”。该函数还会记录任务结果、完成时间以及其他元数据。

此外，make_daily_agenda 函数可以用于生成每日的任务列表，帮助 AI 智能体 快速了解当前的任务状态。通过读取 memory 表中状态为 “in_progress” 的任务，并将其描述信息汇总到一条日志中，可以为智能体提供一个清晰的任务概览。

这种基于 SQLite 的 任务管理 方法，使得 AI 智能体 能够更好地组织和执行任务，避免在多个任务之间切换时丢失 上下文记忆。

5. Python 代码示例：SQLite 驱动的 AI 智能体

文章提供了一段 Python 代码示例，展示了如何使用 SQLite 构建 AI 智能体 的 上下文记忆 系统。这段代码包含以下关键组件：

• MemoryDB 类： 用于封装 SQLite 数据库的操作，包括连接数据库、创建表、写入日志、存储记忆、添加快照以及执行查询等。
• log 函数： 用于简化日志写入操作，自动添加时间戳和元数据。
• task_start 和 task_end 函数： 用于创建和结束任务。
• make_daily_agenda 函数： 用于生成每日的任务列表。
• 命令行接口 (CLI)： 提供了一系列命令行工具，用于管理 SQLite 数据库，例如初始化数据库、写入日志、存储记忆、添加快照以及执行查询等。

这段代码示例展示了如何将 SQLite 数据库集成到 AI 智能体 中，实现 上下文记忆 和 任务管理 功能。开发者可以根据自己的需求，对这段代码进行扩展和定制，构建更强大的 AI 智能体。

from __future__ import annotations
"""sqlite_memory — durable memory layer + **task-aware** CLI for the Dev-Assistant.
Changelog (v2.0 · 2025-05-29)
─────────────────────────────
* Global **`--db`** flag (unchanged)
* **`init`** sub-command creates/validates schema
* **Enriched logging**: helper `log()` embeds timestamp + meta → logs table + FTS
* **Task helpers** (`task_start`, `task_end`, `make_daily_agenda`)
* **House-keeping**: `vacuum` sub-command removes expired memory rows
* **FTS triggers** keep index in sync automatically (no manual insert needed)
"""
from pathlib import Path
from typing import Any, List, Optional, Type, Dict
import argparse
import datetime as _dt
import json
import sqlite3
import sys
import uuid
import os

_DEFAULT_DB = "agent_memory.db"

# ╭──────────────────────────────────────────────────────────────────────────╮
# │ Helper utilities                                                        │
# ╰──────────────────────────────────────────────────────────────────────────╯

def now() -> str:
    return _dt.datetime.utcnow().isoformat(sep=" ", timespec="seconds")

def _in_days(days: int) -> str:
    return (_dt.datetime.utcnow() + _dt.timedelta(days=days)).isoformat(
        sep=" ", timespec="seconds"
    )

# ╭──────────────────────────────────────────────────────────────────────────╮
# │ Core wrapper                                                             │
# ╰──────────────────────────────────────────────────────────────────────────╯

class MemoryDB:
    """SQLite-backed memory + task store with WAL & FTS5."""

    def __init__(self, db_path: str | Path = _DEFAULT_DB):
        self.db_path = Path(db_path)
        self.conn = sqlite3.connect(self.db_path, check_same_thread=False)
        self.conn.row_factory = sqlite3.Row
        with self.conn:
            self.conn.execute("PRAGMA journal_mode = WAL;")
            self.conn.execute("PRAGMA synchronous  = NORMAL;")
            self._init_schema()

    # ───────────────────────────── schema ──────────────────────────────

    def _init_schema(self):
        self.conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS logs (
              id     INTEGER PRIMARY KEY AUTOINCREMENT,
              ts     DATETIME DEFAULT CURRENT_TIMESTAMP,
              stage  TEXT NOT NULL,
              msg    TEXT NOT NULL,
              meta   JSON
            );
            CREATE TABLE IF NOT EXISTS memory (
              key         TEXT PRIMARY KEY,
              value       BLOB NOT NULL,
              expires_at  DATETIME NOT NULL
            );
            CREATE TABLE IF NOT EXISTS checkpoints (
              id        INTEGER PRIMARY KEY AUTOINCREMENT,
              ts        DATETIME DEFAULT CURRENT_TIMESTAMP,
              path      TEXT NOT NULL,
              summary   TEXT,
              snapshot  BLOB
            );
            CREATE VIRTUAL TABLE IF NOT EXISTS fts USING fts5(content);
            """
        )
        # FTS triggers so we never forget to update the index
        self.conn.executescript(
            """
            CREATE TRIGGER IF NOT EXISTS logs_ai AFTER INSERT ON logs BEGIN
              INSERT INTO fts(content) VALUES (NEW.msg);
            END;
            CREATE TRIGGER IF NOT EXISTS checkpoints_ai AFTER INSERT ON checkpoints BEGIN
              INSERT INTO fts(content) VALUES (NEW.summary);
            END;
            """
        )

    # ───────────────────────────── low-level ops ──────────────────────────────

    def write_log(self, stage: str, msg: str, meta: Optional[dict] = None):
        with self.conn:
            self.conn.execute(
                "INSERT INTO logs(stage,msg,meta) VALUES (?,?,?)",
                (stage, msg, json.dumps(meta) if meta else None),
            )

    def remember(self, key: str, value: Any, ttl_days: int = 30):
        expires_at = _in_days(ttl_days)
        with self.conn:
            self.conn.execute(
                "REPLACE INTO memory(key,value,expires_at) VALUES (?,?,?)",
                (key, json.dumps(value), expires_at),
            )

    def get_memory(self, key: str, type_: Type | None = None) -> Optional[Any]:
        cur = self.conn.execute(
            "SELECT value FROM memory WHERE key=? AND expires_at>?",
            (key, now()),
        )
        row = cur.fetchone()
        if not row:
            return None
        raw = json.loads(row[0])
        return type_(raw) if type_ else raw

    def add_checkpoint(self, path: str | Path, summary: str):
        p = Path(path)
        blob = p.read_bytes() if p.exists() else None
        with self.conn:
            self.conn.execute(
                "INSERT INTO checkpoints(path,summary,snapshot) VALUES (?,?,?)",
                (str(p), summary, blob),
            )

    def search(self, query: str = "*", limit: int = 20) -> List[sqlite3.Row]:
        cur = self.conn.cursor()
        if query == "*":
            cur.execute("SELECT stage,msg,ts FROM logs ORDER BY ts DESC LIMIT ?", (limit,))
        else:
            cur.execute("SELECT content FROM fts WHERE fts MATCH ? LIMIT ?", (query, limit))
        return cur.fetchall()

    def vacuum_expired(self):
        with self.conn:
            self.conn.execute("DELETE FROM memory WHERE expires_at<=?", (now(),))

    def close(self):
        self.conn.close()

# ╭──────────────────────────────────────────────────────────────────────────╮
# │ High-level helpers (logging & tasks)                                    │
# ╰──────────────────────────────────────────────────────────────────────────╯
# NOTE: These are top-level functions so user code can `from sqlite_memory import log, task_start …`
# They rely on *one* global DB instance created by `get_db()` below.

_DB_SINGLETON: MemoryDB | None = None

def get_db(path: str | Path = _DEFAULT_DB) -> MemoryDB:
    global _DB_SINGLETON
    if _DB_SINGLETON is None:
        _DB_SINGLETON = MemoryDB(path)
    return _DB_SINGLETON

# ── enriched log wrapper ───────────────────────────────────────────────────

def log(stage: str, msg: str, **meta):
    db = get_db()
    db.write_log(stage, msg, {**meta, "ts": now()} if meta else {"ts": now()})

# ── task helpers ───────────────────────────────────────────────────────────

_TASK_KEY_PREFIX = "task:"

def task_start(desc: str, **extra) -> str:
    """Create a task; return its 8-char id."""
    tid = str(uuid.uuid4())[:8]
    payload: Dict[str, Any] = {
        "desc": desc,
        "status": "in_progress",
        "created_at": now(),
        **extra,
    }
    db = get_db()
    db.remember(f"{_TASK_KEY_PREFIX}{tid}", payload, ttl_days=30)
    log("task:start", desc, task_id=tid)
    return tid

def task_end(tid: str, result: str, **extra):
    db = get_db()
    key = f"{_TASK_KEY_PREFIX}{tid}"
    task = db.get_memory(key) or {}
    task.update({
        "status": "done",
        "result": result,
        "closed_at": now(),
        **extra,
    })
    db.remember(key, task, ttl_days=30)
    log("task:end", result, task_id=tid)

def _open_tasks(db: MemoryDB):
    cur = db.conn.execute(
        "SELECT key,value FROM memory WHERE key LIKE ? AND expires_at>?",
        (f"{_TASK_KEY_PREFIX}%", now()),
    )
    for key, raw in cur.fetchall():
        task = json.loads(raw)
        if task.get("status") != "done":
            yield key, task

def make_daily_agenda(db: MemoryDB):
    """Write one log line summarising all open tasks."""
    open_tasks = list(_open_tasks(db))
    if not open_tasks:
        return
    summary = ", ".join(f"{tid.split(':')[1]}: {t['desc']}" for tid, t in open_tasks)
    db.write_log("agenda", f"open tasks → {summary}")

# ╭──────────────────────────────────────────────────────────────────────────╮
# │ CLI                                                                     │
# ╰──────────────────────────────────────────────────────────────────────────╯

def _parse_cli(argv: List[str]):
    parser = argparse.ArgumentParser("sqlite_memory CLI")
    parser.add_argument("--db", default=_DEFAULT_DB, help="Path to SQLite file")
    sub = parser.add_subparsers(dest="cmd", required=True)
    # core ops
    sub.add_parser("init", help="Initialise / upgrade schema only")
    p_log = sub.add_parser("log", help="Write log entry")
    p_log.add_argument("stage")
    p_log.add_argument("msg")
    p_rem = sub.add_parser("remember", help="Remember key → value")
    p_rem.add_argument("key")
    p_rem.add_argument("value")
    p_rem.add_argument("--ttl", type=int, default=30)
    p_get = sub.add_parser("get", help="Get remembered value if not expired")
    p_get.add_argument("key")
    p_cp = sub.add_parser("checkpoint", help="Snapshot a file")
    p_cp.add_argument("path")
    p_cp.add_argument("summary")
    p_search = sub.add_parser("search", help="FTS query")
    p_search.add_argument("query")
    p_search.add_argument("--limit", type=int, default=20)
    # task ops
    p_ts = sub.add_parser("task-start", help="Start new task")
    p_ts.add_argument("desc")
    p_te = sub.add_parser("task-end", help="Mark task done")
    p_te.add_argument("tid")
    p_te.add_argument("result")
    # maintenance
    sub.add_parser("vacuum", help="Delete expired memory rows")
    return parser.parse_args(argv)

def _cli():
    args = _parse_cli(sys.argv[1:])
    db = get_db(args.db)
    match args.cmd:
        case "init":
            print(f"✅ {args.db} ready (schema up to date).")
        case "log":
            log(args.stage, args.msg)
            print("OK: event logged")
        case "remember":
            db.remember(args.key, args.value, ttl_days=args.ttl)
            print("OK: value remembered")
        case "get":
            print(db.get_memory(args.key))
        case "checkpoint":
            db.add_checkpoint(args.path, args.summary)
            print("OK: checkpoint stored")
        case "search":
            rows = db.search(args.query, limit=args.limit)
            for r in rows:
                # row could be from logs (stage,msg,ts) or fts (content)
                if "content" in r.keys():
                    print(r["content"])
                else:
                    print(f"{r['ts']} · {r['stage']}: {r['msg']}")
        case "task-start":
            tid = task_start(args.desc)
            print(f"task {tid} started")
        case "task-end":
            task_end(args.tid, args.result)
            print(f"task {args.tid} completed")
        case "vacuum":
            db.vacuum_expired()
            print("Expired memory rows purged")
        case _:
            print("unknown command", file=sys.stderr)
            sys.exit(1)
    db.close()

if __name__ == "__main__":
    _cli()

6. AI 智能体 Prompt 指南：确保一致性和可控性

为了确保 AI 智能体 能够正确地使用 SQLite 数据库，需要提供清晰的 Prompt 指南。文章中提供了一份详细的 Prompt 指南，包括以下内容：

• 任务描述： 明确 AI 智能体 的任务目标，即“代码、回忆和继续”，确保智能体在不丢失 上下文记忆 的情况下快速编写、测试和调试代码。
• 持久化引擎： 介绍 SQLite 数据库的特性和使用方法，包括数据库的路径、WAL 模式以及 CLI 工具的使用方法。
• 日志记录指南： 详细说明如何使用 db.write_log 函数记录日志，包括日志消息的描述性、元数据的捕获以及信息量的控制。
• 任务管理方法： 介绍如何使用 task_start 和 task_end 函数管理任务，包括任务状态的更新、依赖关系的记录以及每日任务列表的生成。
• 数据库模式： 提供 SQLite 数据库的表结构定义，帮助 AI 智能体 了解数据库的组织方式。
• 会话循环： 描述 AI 智能体 的会话流程，包括启动/恢复、澄清、计划/工作以及总结等步骤。
• 测试驱动开发： 建议采用测试驱动开发方法，确保代码的质量和可靠性。
• Slash 命令： 提供一系列用户命令，用于管理 AI 智能体，例如 /plan、/status、/search、/checkpoint 和 /rollback 等。
• 规范化阶段： 定义一系列规范化的阶段，例如 init、plan、fetch、build、test、deploy、checkpoint、done 和 error 等，帮助 AI 智能体 更好地跟踪任务进度。
• 错误处理流程： 描述错误处理流程，包括错误信息的记录、修复建议以及回滚操作等。
• 语气： 建议采用简洁、可操作的语气，并提供“需要更多？”选项，方便用户进行深入研究。

这份 Prompt 指南为 AI 智能体 提供了一个清晰的框架，确保智能体能够正确地使用 SQLite 数据库，并有效地完成任务。

7. 测试驱动开发：保障代码质量

文章强调了测试驱动开发 (TDD) 的重要性，建议开发者在开发 AI 智能体 时采用 TDD 方法。通过先编写测试用例，然后再编写代码，可以确保代码的质量和可靠性。文章提供了一份 TDD 指南，包括以下内容：

• 业务逻辑层： 关注业务逻辑的正确性，采用 pytest 和 pytest-asyncio 等工具进行测试。
• CLI 封装层： 关注命令行接口的正确性，采用 pytest 和 click.testing 等工具进行测试。
• 任务辅助层： 关注任务辅助函数的正确性，采用内存数据库进行测试。
• 集成层： 关注整个系统的集成性，采用真实数据库文件进行测试。

通过 TDD 方法，可以有效地减少 Bug，提高代码质量，并确保 AI 智能体 能够稳定运行。

8. 结论：SQLite 赋能 AI 智能体，提升效率和可靠性

总而言之，将 SQLite 数据库集成到 AI 智能体 中，可以有效地解决 上下文记忆 的问题，提升智能体的效率和可靠性。通过使用 SQLite 数据库，AI 智能体 可以：

• 持久化存储上下文信息： 避免在中断后丢失状态，实现智能恢复。
• 高效管理任务： 跟踪任务进度，避免重复劳动，实现智能任务管理。
• 灵活查询数据： 使用 SQL 语句分析运行数据，为智能决策提供支持。

虽然现在 大模型 技术发展迅速，但结合 SQLite 这样的传统工具，反而能够更好地发挥 AI 智能体 的潜力。 这种方法不仅可以应用于 AI 智能体 的开发，还可以推广到其他需要持久化存储和数据分析的场景，具有广泛的应用前景。随着 大模型 技术的不断发展，我们有理由相信，基于 SQLite 的 上下文记忆 系统将在未来的 AI 智能体 开发中发挥越来越重要的作用。