SQL、事务与数据基础：让 Agent 的状态能被查清楚

数据库可以先理解成一套按规则组织的电子账本。商品还剩几件、任务进行到哪一步，都可以记录进去，之后按条件查出来。

先分清三个名字：数据库是组织好的数据；SQLite 是负责存取和管理这些数据的软件；SQL 是告诉软件“存什么、查什么、改什么”的语言。接下来从一条库存记录开始，一步步把它们用起来。

0. 数据库、SQL、SQLite，到底谁是谁？

如果把数据库里的数据想成按栏目整理好的账本，那么 SQLite 就像管理账本的管理员。你通过 SQL 提出要求，例如“查一下书还剩几件”，它负责找到记录并返回结果。

三个名字分别属于不同层次：

SQLite 是一个轻量的数据库软件。它能直接在程序中工作：通常把数据存在一个文件里，程序通过它读写这个文件。Python 提供了 sqlite3 模块，让我们用 Python 调用 SQLite；在课程使用的 Python 安装中，写 import sqlite3 就能使用，接下来的例子无需另装数据库服务器。

SQLite、MySQL、PostgreSQL 都是具体的数据库软件。后两者通常会运行一个独立服务，应用连接服务来读写数据；SQLite 直接嵌在应用中工作。这是选择 SQLite 做第一次实验的原因：先把数据存取学明白，再学习服务器、账号和连接配置。SQLite 的工作方式 · Python 的 sqlite3 模块

Agent 也会用到这些能力。例如助手替你创建工单后，把工单编号、处理状态和结果存进数据库，下次查询时就能找回进度。模型负责理解任务，程序执行动作，数据库负责保存动作产生的记录。

1. 表、行、列：先看懂一条库存记录

在这类数据库中，数据按“表”组织。表很像一张有列名的电子表格。假设有一张叫 inventory 的库存表：

横着看，book 和 2 合起来是一行，也就是一条记录：“商品 book 还有两件”。竖着看，sku 是放商品编号的列，stock 是放库存数量的列。以后增加其他商品，就在同一张表里增加行。

sku 是商品编号的英文缩写，这里用 book 表示书这个商品；inventory 和 stock 分别是“库存表”和“库存数量”的英文名称。它们是我们给表和列起的名字。

建表就是先告诉数据库：“我要一张库存表，它有商品编号、库存数量这两列。”列名确定了，写数据时才能知道每个值应该放在哪里。我们会把编号列写成 TEXT，表示文本；把数量列写成 INTEGER，表示整数。

2. 写下第一段 SQL：存入、查询，再修改

现在可以给数据库发指令了。先认识四个动作，每个动作都能对应刚才那张表：

新建一个学习文件夹，在里面创建 sql_basics.py，复制下面的完整代码并用编辑器保存。在资源管理器打开这个文件夹，点击上方地址栏，输入 powershell 并回车。在打开的命令窗口输入 python sql_basics.py，再按回车运行。

macOS：先在 Python 篇的学习文件夹中打开终端，执行 source .venv/bin/activate。再输入 cd ，把当前例子的文件夹拖入终端并回车，运行下面的命令。

# 按固定输入观察SQL、事务与数据基础的处理过程和返回结果。
import sqlite3

db = sqlite3.connect(":memory:")

db.execute("CREATE TABLE inventory (sku TEXT, stock INTEGER)")
db.execute("INSERT INTO inventory (sku, stock) VALUES ('book', 2)")

row = db.execute("SELECT sku, stock FROM inventory WHERE sku = 'book'").fetchone()
print("修改前：", row)

db.execute("UPDATE inventory SET stock = 1 WHERE sku = 'book'")
db.commit()

row = db.execute("SELECT sku, stock FROM inventory WHERE sku = 'book'").fetchone()
print("修改后：", row)

db.close()

预期输出：

修改前： ('book', 2)
修改后： ('book', 1)

先按从上到下的顺序读这段程序：

1. import sqlite3：载入 Python 用来操作 SQLite 的模块。
2. sqlite3.connect(":memory:")：打开一个放在内存里的数据库。db 是接下来操作它的连接对象，可以理解成“通往这份数据库的入口”。这里的数据随着连接关闭而释放；如果要存在文件中，连接时可以使用 "learning.db" 这样的文件名。
3. db.execute("...")：让数据库执行引号里的 SQL。第一条先建表，第二条往表里增加一行。
4. 第一条查询：从库存表里找出 book 这一行，把编号和数量读出来。
5. UPDATE：把这一行的库存改成 1。db.commit() 确认这次修改。
6. 第二条查询：再读同一行，看到数量已经变化。最后 db.close() 关闭连接。

SQL 是字符串里的指令，Python 负责把这条指令交给 SQLite 执行。例如查询语句可以拆成这样：

SELECT sku, stock
FROM inventory
WHERE sku = 'book';

SELECT sku, stock 表示“读取编号和库存这两列”；FROM inventory 表示“从库存表里读”；WHERE sku = 'book' 表示“只找编号是 book 的行”。连起来就是：“查一下库存表里，book 的编号和库存数量。”SQL 中的 = 在这里表示比较是否相等。

fetchone() 表示“取出查询结果中的一行”。打印的 ('book', 2) 是一个元组，可以理解成这一行各列的值按顺序排在一起：第一个是商品编号，第二个是库存数量。查不到时得到 None，表示没有这条记录。

你也可以把修改语句中的 stock = 1 改成 stock = 5，保存后再运行，第二行就会变成 ('book', 5)。每次运行都从新建的内存数据库开始，所以第一行仍是 ('book', 2)。到这里，建表、增加记录、读取记录和修改记录已经连成了一条完整流程。

如果要删除一行，对应的指令是 DELETE FROM inventory WHERE sku = 'book'：删除的是这条商品记录，表本身还在。和 SELECT、UPDATE 一样，WHERE 指定这次操作针对哪些行。

3. 事务把一组变化放在边界内

想象收银员给你结账：扣掉一件库存，同时留下购买记录。这两笔账应一起生效。数据库的事务就是把一组修改捆在一起：都成功便提交，途中失败便撤销这组尚未提交的修改。这里的动作都发生在同一个数据库里；给外部系统发出的请求另有自己的处理流程。

下面三个英文词是 SQL 命令。BEGIN 开始这组修改，COMMIT 确认保存，ROLLBACK 撤销尚未提交的修改。SQL 中 -- 后面是给人看的注释。

BEGIN;  -- 开始一组修改
UPDATE inventory SET stock = stock - 1 WHERE sku = 'book' AND stock >= 1;
-- 程序确认成功修改一行，再写入预留记录
INSERT INTO reservations VALUES ('r1', 'book', 1, 'succeeded');
COMMIT;  -- 一起保存库存和预留记录

这里的 stock 是剩余可用库存，扣一件就减1；预留表另加一行。两次写入放在 BEGIN 与 COMMIT 之间，同属一个事务。下一节通过完整程序建立两张表并运行这组变化，变量值通过参数传入。

如果条件不满足，执行 ROLLBACK，库存和预留记录就恢复到本次修改前。可以把它理解成收银员取消这次结账。事务持续得越久，其他写入等候的时间可能越长；常见做法是用短事务保存状态，等外部调用完成后再开一笔事务更新结果。

4. 先亲手跑通一笔正常预留

下面的源码用 Python 自带的 SQLite 创建两张临时表：inventory 存库存，reservations 存预留记录。初始 book 有 2 件；请求 r1 预留 1 件后，正常结果应是库存 1、预留记录 1。程序随后还会演示重复提交和故意失败；先只盯住输出中的 initial 与 first 两行。

在电脑上新建一个空文件夹，例如 F:\agentdepth-sql-lab。复制下方完整代码，保存为 transactions.py 和 test_transactions.py 到同一文件夹。在资源管理器打开该文件夹，在地址栏输入 powershell 并回车。逐行运行：

"""用SQLite保存库存和预订记录，观察提交、重复请求和失败回滚。"""
import sqlite3


def connect():
    # 在内存中创建数据库；由下面的 BEGIN/COMMIT/ROLLBACK 明确控制事务。
    db = sqlite3.connect(":memory:", isolation_level=None)
    db.executescript("""
        CREATE TABLE inventory (
            sku TEXT PRIMARY KEY NOT NULL,
            stock INTEGER NOT NULL CHECK(stock >= 0)
        );
        CREATE TABLE reservations (
            request_id TEXT PRIMARY KEY NOT NULL,
            sku TEXT NOT NULL,
            quantity INTEGER NOT NULL CHECK(quantity > 0),
            status TEXT NOT NULL CHECK(status IN ('succeeded'))
        );
        INSERT INTO inventory VALUES ('book', 2);
    """)
    return db


def reserve(db, request_id, sku, quantity, *, fail_after_update=False):
    """预订商品：扣减库存，并保存这次预订的编号和数量。"""
    if not isinstance(request_id, str) or not request_id.strip():
        raise ValueError("request_id must be non-empty")
    if not isinstance(sku, str) or not sku.strip():
        raise ValueError("sku must be non-empty")
    if type(quantity) is not int or quantity <= 0:
        raise ValueError("quantity must be a positive integer")
    # 把扣库存和记预订单放在同一事务中：一起成功，或一起撤销。
    db.execute("BEGIN IMMEDIATE")
    try:
        previous = db.execute(
            "SELECT sku, quantity FROM reservations WHERE request_id = ?",
            (request_id,),
        ).fetchone()
        # 同一请求编号已办理过时，核对商品和数量，再返回已有结果。
        if previous is not None:
            if previous != (sku, quantity):
                raise ValueError("idempotency key reused with different input")
            db.execute("COMMIT")
            return "replayed"
        changed = db.execute(
            "UPDATE inventory SET stock = stock - ? WHERE sku = ? AND stock >= ?",
            (quantity, sku, quantity),
        ).rowcount
        if changed != 1:
            raise ValueError("unknown sku or insufficient stock")
        if fail_after_update:
            raise RuntimeError("injected failure before reservation insert")
        db.execute(
            "INSERT INTO reservations VALUES (?, ?, ?, 'succeeded')",
            (request_id, sku, quantity),
        )
        db.execute("COMMIT")
        return "created"
    except BaseException:
        if db.in_transaction:
            db.execute("ROLLBACK")
        raise


def snapshot(db):
    stock = db.execute("SELECT stock FROM inventory WHERE sku = 'book'").fetchone()[0]
    count = db.execute("SELECT COUNT(*) FROM reservations").fetchone()[0]
    return f"stock={stock}, reservations={count}"


def main():
    db = connect()
    try:
        print("initial:", snapshot(db))
        print("first:", reserve(db, "r1", "book", 1), snapshot(db))
        print("retry:", reserve(db, "r1", "book", 1), snapshot(db))
        for label, key, quantity, inject in [
            ("different input", "r1", 2, False),
            ("partial failure", "r2", 1, True),
            ("insufficient", "r3", 2, False),
        ]:
            try:
                reserve(db, key, "book", quantity, fail_after_update=inject)
            except (ValueError, RuntimeError) as error:
                print(f"{label}: {type(error).__name__}; {snapshot(db)}")
        print("after rollback:", reserve(db, "r2", "book", 1), snapshot(db))
    finally:
        db.close()


if __name__ == "__main__":
    main()

# 核对SQL、事务与数据基础的正常结果和错误分支；断言失败时会显示对应的测试。
import sqlite3
import unittest
from transactions import connect, reserve, snapshot


class ReservationsTest(unittest.TestCase):
    def setUp(self):
        self.db = connect()
        self.addCleanup(self.db.close)

    def test_success_and_replay_only_decrement_once(self):
        self.assertEqual(reserve(self.db, "a", "book", 1), "created")
        self.assertEqual(reserve(self.db, "a", "book", 1), "replayed")
        self.assertEqual(snapshot(self.db), "stock=1, reservations=1")

    def test_key_reuse_with_changed_input_is_rejected(self):
        reserve(self.db, "a", "book", 1)
        with self.assertRaisesRegex(ValueError, "different input"):
            reserve(self.db, "a", "book", 2)
        self.assertEqual(snapshot(self.db), "stock=1, reservations=1")
        self.assertFalse(self.db.in_transaction)

    def test_partial_write_rolls_back_and_retry_can_succeed(self):
        with self.assertRaises(RuntimeError):
            reserve(self.db, "a", "book", 1, fail_after_update=True)
        self.assertEqual(snapshot(self.db), "stock=2, reservations=0")
        self.assertEqual(reserve(self.db, "a", "book", 1), "created")

    def test_insufficient_stock_changes_nothing(self):
        with self.assertRaises(ValueError):
            reserve(self.db, "a", "book", 3)
        self.assertEqual(snapshot(self.db), "stock=2, reservations=0")

    def test_boolean_and_negative_quantity_are_not_valid(self):
        for value in [True, False, 0, -1, 1.0, "1"]:
            with self.subTest(value=value), self.assertRaises(ValueError):
                reserve(self.db, "a", "book", value)

    def test_bound_key_is_data_not_sql(self):
        key = "x'); DROP TABLE inventory; --"
        self.assertEqual(reserve(self.db, key, "book", 1), "created")
        self.assertEqual(reserve(self.db, key, "book", 1), "replayed")
        self.assertEqual(snapshot(self.db), "stock=1, reservations=1")

    def test_unique_constraint_still_protects_direct_writes(self):
        reserve(self.db, "a", "book", 1)
        with self.assertRaises(sqlite3.IntegrityError):
            self.db.execute("INSERT INTO reservations VALUES ('a', 'book', 1, 'succeeded')")
        self.assertEqual(snapshot(self.db), "stock=1, reservations=1")


if __name__ == "__main__":
    unittest.main(verbosity=2)

ls
python --version
python -c "import sqlite3; print(sqlite3.sqlite_version)"
python -B -X utf8 transactions.py

先用 Get-ChildItem 确认两个文件在当前目录，再运行命令并查看 SQLite 版本。如果系统找不到 py，回到 Python 基础检查安装；如果找不到 transactions.py，检查终端所在目录和文件扩展名。正常输出开头应是 initial: stock=2, reservations=0，接着是 first: created stock=1, reservations=1。这两行说明一次预留已经同时改变两张表；下一节会逐行解释后面的回放、冲突与回滚。

5. 复盘实验：故意失败后为什么没有扣库存

只看 BEGIN 和 COMMIT，容易把事务误解成两行固定写法。接下来用 Python 自带的 SQLite 执行 SQL，对比失败前后两张表的值，观察同一数据库中的原子变化。

第 4 节已经把程序跑起来了。回到同一个文件夹，再运行配套测试：

python -B -X utf8 -m unittest -v test_transactions

-B 禁止写入字节码缓存，与事务行为无关；-X utf8 统一终端编码。程序输出的前两行是正常路径，后面几行才是我们要分析的重复请求与回滚。测试通过不能代替读懂状态变化，接下来按两张表逐步复盘。

两张表分别负责什么

inventory 保存当前可用库存，初始有两本书。reservations 保存已完成的预留记录，一次业务请求最多一条。这里把预留与扣库存放在同一事务，因此预留表只有成功状态；没有为了演示状态机而写入一个无法解释的 running 状态。

同一请求最多一行预留，库存至少为0：先把这两个约束在两张表中实现，再考虑退款和订单等业务关系。

建表代码逐行看什么

connect() 用 sqlite3.connect(":memory:") 创建临时数据库。它只存在于当前程序的内存里，程序结束后记录会消失；这方便我们反复练习同一组输入。executescript() 一次执行多条建表和初始化语句：

CREATE TABLE inventory (
    sku TEXT PRIMARY KEY NOT NULL,
    stock INTEGER NOT NULL CHECK(stock >= 0)
);
INSERT INTO inventory VALUES ('book', 2);

逐词读这段：CREATE TABLE 是建表；括号内定义列；TEXT 和 INTEGER 是字段类型；NOT NULL 禁止空值；PRIMARY KEY 保证定位键唯一；CHECK 把最低限度的业务规则交给数据库再守一遍。最后一行插入一本代号为 book 的商品，初始可用库存为 2。应用代码先检查输入，数据库约束再检查最终写入。其他程序写同一张表时，也会受这些约束限制。

查询结果也不是一个裸数字。fetchone() 返回一行，常见形式是元组，例如 (2,)；查不到时返回 None。(2,) 末尾的逗号表示只有一个元素的元组，[0] 才取出单元格里的 2。初学者若直接对 None[0] 取值会报错，因此真实查单接口应先分支处理“没有记录”。

为什么先检查键，再执行条件更新

核心过程在 reserve() 中。先获取写事务，再查询已有预留；相同键、相同输入返回 replayed，不再扣库存。相同键、不同输入则抛错，防止调用方把两个业务意图混在一起。

接下来只让满足库存条件的一行发生变化：

changed = db.execute(
    "UPDATE inventory SET stock = stock - ? WHERE sku = ? AND stock >= ?",
    (quantity, sku, quantity),
).rowcount
if changed != 1:
    raise ValueError("unknown sku or insufficient stock")

rowcount 为零不是 Python 异常，却是业务失败：商品不存在或库存不够。忽略它，后面的预留插入仍可能成功，造成“有订单、没扣库存”。这里故意把两种失败合在一个错误中；对用户展示时可以细分，但不能用另一次无锁查询替代条件更新。

SQL 中的问号是参数绑定位置，不是字符串拼接。测试会把带引号和 SQL 片段的请求键作为普通值写入，再检查库存表仍然存在。参数绑定不能为任意表名、排序方向或整段 SQL 授权；如果这些部分来自模型，需要固定候选项与权限控制。Python sqlite3 参数绑定说明

把 reserve() 按控制流拆开，就不会只看到一大段代码：

except BaseException 在教学代码中是为了让键盘中断等异常也触发清理，然后原样抛出；业务服务通常还会在更外层区分可重试异常、输入错误和进程终止。不要捕获后静默返回“成功”。

观察结果，不只观察“没有报错”

运行程序后，按请求顺序核对库存和预留记录：

initial: stock=2, reservations=0
first: created stock=1, reservations=1
retry: replayed stock=1, reservations=1
different input: ValueError; stock=1, reservations=1
partial failure: RuntimeError; stock=1, reservations=1
insufficient: ValueError; stock=1, reservations=1
after rollback: created stock=0, reservations=2

第四行检查键的输入一致性。第五行最重要：代码已经扣了库存，但在插入预留前故意抛出 RuntimeError；显式 ROLLBACK 让库存回到 1，预留仍为 1。最后用同一个 r2 重试成功，说明失败事务没有留下占用这个键的半条业务记录。

注意区分“本地明确回滚”与“客户端不知道是否提交”。这个实验捕获到的注入异常属于前者；如果是数据库连接断开或外部服务响应丢失，不能仅凭异常就断言没有完成。应重新连接并按业务键核对，仍无法确定时保留未知状态。

两个容易误用的事务细节

代码选择 isolation_level=None 并显式发送 BEGIN IMMEDIATE、COMMIT、ROLLBACK，避免依赖 Python 隐式开启事务。Python 新版本还提供 autocommit 参数；不要混用两套事务管理方式，再根据函数名猜它是否已经提交。Python 事务控制说明

SQLite 的 BEGIN IMMEDIATE 会尝试立即取得写事务；有其他写者时可能失败。实验只有一个内存连接，没有测锁等待、多个进程或持久化恢复。把它迁移到 PostgreSQL 时，必须按该数据库的隔离和锁行为重做并发测试，不能照搬 SQLite 结论。SQLite 事务文档

还有一个细节：约束错误不总会自动撤销本事务之前的所有语句。因此示例在异常出口显式检查 in_transaction 并回滚；不能简单认为“INSERT 报错，所以之前 UPDATE 一定没了”。SQLite 冲突处理语义

试着破坏一次，再修回来

先复制实验文件，避免覆盖可对照的版本。只在副本里把异常出口的 db.execute("ROLLBACK") 替换成 pass，保留缩进（不能直接注释掉，否则 if 的空语句块会导致语法错误）。再次执行，你会发现注入失败后的库存和连接状态不再符合预期，后续事务也可能无法正常开始。这比记住“事务要回滚”更有用：你能具体指出哪一步留下了问题。

再把第二次请求的数量改成 2，但仍使用 r1。合理结果是输入冲突，而不是 replayed 或再次扣库存。幂等键不是一句“查到就跳过”，它还表达了两次调用应该是同一个业务动作。

配套七项单元测试检查正常提交、重复键、变更输入、部分失败回滚、库存不足、非法数量、参数绑定与唯一约束（部分在同一测试中验证）。接入持久数据库后，还要增加多连接同时预留的检查。

6. 用唯一约束识别重复请求

给 request_id 建唯一约束，可以阻止同一键插入两次；但它不会自动知道两次请求是否真的是同一业务意图，也不会替你返回第一次执行结果。应用仍需捕获冲突、查询已有记录，并定义 running、succeeded 与 unknown 状态下的处理。

当客户端超时，服务端可能已经提交。再次插入同一键如果被唯一约束拒绝，这个拒绝本身是有用信号：先查原记录，再决定是否继续。不要把唯一冲突统一显示为系统故障，更不要换一个新键绕过约束。

7. Agent 记录要区分输入、动作和结果

一条执行记录要回答：谁发起、哪次请求、使用哪版输入、当前状态、结果在哪里、失败属于哪类。request_id 是一次业务请求的编号；同一次业务重试复用这个编号，就能查出它之前是否已经处理过。

状态可以用 queued（排队）、running（执行中）、succeeded（成功）、failed（失败）、unknown（结果尚未确认）表示。后续查单、重试和人工处理根据这些状态选择下一步。

建议至少分开保存三类对象：原始任务摘要、工具动作与参数摘要、工具响应和最终状态。这样可以回答“模型想做什么”“程序实际调用了什么”“服务端返回了什么”。敏感输入要脱敏或只保留哈希与引用，不能因为数据库有权限控制就无限期保存完整 prompt。

数据表并不是日志系统的替代品：日志更适合时间线和诊断，数据库更适合状态查询和约束。两者之间用 request_id、trace_id 等关联，但要明确保留期、访问范围和删除策略。

8. 失败重试为什么会产生重复订单

客户端等待超时不等于服务端没执行。第一次请求可能已经创建订单，只是响应丢失；客户端再发一个新键，数据库看到的是两个合法插入。修复方向包括：稳定业务键加唯一约束、提供按键查单接口、把创建和后续步骤拆成可恢复状态机。选择哪一种取决于业务能否查询和补偿。

跨服务写入、消息发送和外部支付可能发生在数据库提交之后。为它们保留业务键与处理状态，响应丢失时按业务键查单，查不清时交给人工核对。

9. 可查询比“保存了 JSON”更重要

把所有内容塞进一个 JSON 字段初期很快，但状态、时间和关联键难以约束。反过来，把每个字都拆成列也会让模型输出难以演进。可以先把需要筛选、唯一和排序的字段结构化，其余结果保存为受控 JSON，并为格式变化保留版本号。

迁移字段时同时更新写入与读取逻辑，并考虑已有记录怎样读取。测试应覆盖旧版本读取、未知状态和重复键冲突。数据格式是接口的一部分，和 HTTP 响应一样需要版本管理。

10. 练习与实践

11. 总结与下一步

留给下一个开发者的记录可以这样写：

• 稳定键 r1 标识同一次业务动作；同输入再次提交时，应查到原结果，而不是再扣一次库存。
• 事务把本数据库中的库存更新与预留记录放在一起；参数绑定保证输入按数据处理。外部支付不跟着 SQLite 事务一起回滚。
• 离线实验里，r1 同输入重放后库存仍为 1；把数量改成 2 会被拒绝；注入失败后库存也仍为 1。把表结构、执行命令和三条输出一起留在作品里。

事务与参数绑定可进一步阅读 SQLite 事务 和 Python sqlite3 占位符。下一篇读向量与余弦相似度。

想继续研究事务，可以先看 SQLite 事务文档，再对照 PostgreSQL 事务与约束，注意两者的锁和并发行为不能照搬。

参考资料

• Python sqlite3 教程与占位符：这里连接、查询结果与参数绑定的直接依据。
• SQLite 事务 与 冲突处理：理解显式事务、并发限制和约束错误的边界。
• PostgreSQL 事务教程 与 约束：进入服务端关系型数据库时的下一步阅读。