ORM学习：从 Python 对象到数据库事务

读 Agent 后端代码时，容易混在一起的是三件事：数据库里的记录、Python 中表示记录的对象，以及负责操作它们的 SQLAlchemy Session。这篇笔记沿着一次任务的保存过程，把对象映射、事务、Repository、关联查询和重复请求串起来。

先记住一条主线： 数据库保存记录，Python 使用对象，Session 管理对象变化，数据库事务决定哪些修改一起提交。 再沿着这条线理解 N+1、幂等、异步访问和数据库迁移，会更容易看清每个工具负责什么。

下面使用 SQLAlchemy 2.x 的概念和简化的 Agent 场景。AgentSession / Turn / Run / Message 是教学中的业务名称，具体字段、表关系和 database.transaction() 的实现仍需对照 TARS-Agent 源码确认。代码片段用于解释操作顺序，省略了 Model、数据库连接配置和外围函数；含 await 的片段应放在异步函数中。它们不是已经核对并运行过的项目完整代码。

1. 先分清：数据、表示数据的对象、操作数据的对象

用户提交任务以后，Agent 需要记住用户说了什么、任务执行到哪一步、最后成功还是失败。

在我们的场景里，数据库负责保存这些信息。Python 代码则需要读取和修改它们。

数据库、SQL、ORM 分别解决什么问题？

数据库 负责存储和管理数据。在关系型数据库里，这些数据主要按表、行、列组织。

SQL 是程序向数据库表达操作要求的语言。例如，查询某个 Run、插入一条 Message、修改任务状态。

ORM 让程序可以通过 Python 对象表达这些操作。它把类与表、对象属性与表字段对应起来，再通过 SQL 与数据库交互。ORM 减少了手写 SQL 的工作，但没有让 SQL 消失。SQLAlchemy 文档

例如，数据库有一条记录：

Python 里可以通过一个 Run 对象表示它，读取 run.id 和 run.status。

这里要分清三个名字：

Model 是映射规则，对象是 Python 中的具体数据，记录是数据库中的具体数据。 在我们讨论的简单映射里，一个已保存的 Run 对象对应一条 Run 记录。SQLAlchemy 文档

创建对象和查询对象，不是一件事

假设 Run 已经是正确配置的 ORM Model：

run = Run(id="r001", status="pending")

这一行的输入是 ID 和状态，输出是一个 Python 对象。它没有通过 Session 发出数据库写入操作，所以 不会仅凭这一行就在数据库里新增记录 。SQLAlchemy 文档

再看：

run = await db_session.get(Run, run_id)

输入是 Model 类和要查找的主键。输出是对应的 ORM 对象；找不到时返回 None。

这里表达的是“取得已有记录的对象表示”，不是“新建一条记录”。而且，同一个 Session 已经管理着这个对象时，get() 可能直接复用它，不一定再发一次 SQL。SQLAlchemy 文档

查询记录后，代码得到的是它的 Python 对象表示。这里有一个容易混淆的地方：

新出现一个 Python 对象，不等于数据库新增了一条记录。

Engine 和 Session 各负责什么？

Engine 负责数据库连接的配置和提供。 它知道使用哪种数据库、哪种驱动、连接到哪里。Engine 不是数据库本身，也不等于一条固定连接。SQLAlchemy 文档

SQLAlchemy Session 是业务代码操作数据库的主要对象。 它管理 ORM 对象，组织查询和写入，并管理事务的提交、回滚。数据库事务的实际执行与保障，仍然由数据库完成。SQLAlchemy 文档

最容易混的两个名字是：

例如，db_session 可以查询出一个 chat_session。前者负责操作，后者是被操作的数据。

以后读代码时，可以主动把业务变量 session 在脑中改叫 chat_session 。这样不容易把两者混在一起。

2. 改了 Python 属性，数据库为什么也能更新？

先看关键代码。这里假定 run_id 来自业务输入：

run = await db_session.get(Run, run_id)
if run is None:
    raise ValueError("Run 不存在")

run.status = "completed"
await db_session.flush()

第一行取得 Run 对象。中间的判断处理“查不到”的情况。属性赋值先修改 Python 对象，最后的 flush() 才把需要执行的修改发送给数据库。

为什么不用另外写一条 UPDATE？

因为这个对象由当前 db_session 管理。SQLAlchemy 为映射属性设置了变化跟踪机制，能够记录属性修改。到 flush() 时，它根据对象对应的表、主键和字段变化，决定需要执行什么 SQL。SQLAlchemy 文档

这里有三个边界。

第一，ORM 对象不是一创建就归某个 Session 管。 自己调用 Run(...) 创建的新对象，通常要经过 db_session.add(run) 才进入这个 Session 的管理范围。通过 Session 查询得到的对象，则已经由它管理。SQLAlchemy 文档

第二，“跟踪变化”不是后台线程一直盯着对象。 也不是不断重新查询数据库。它主要跟踪当前管理对象在 Python 这一侧发生的修改。别的请求修改了数据库，你手上的对象不一定立刻同步更新。SQLAlchemy 文档

第三，修改对象不等于已经提交。 对象要仍在相应 Session 的管理中，修改要经过 flush，事务还要成功提交。对象脱离 Session 后，不能认为继续改属性就会自动保存。SQLAlchemy 文档

所以完整理解是：

受 Session 管理的 ORM 对象发生修改 → SQLAlchemy 记录变化 → flush 执行 SQL → commit 提交事务。

3. 事务：把“执行过”与“提交了”分开

用户提交任务时，我们可能需要创建 Turn、创建 Run、保存用户 Message，并更新 Agent 会话状态。

我们不希望 Run 创建失败了，却留下其他几项已经提交的数据。因此，需要把这些修改放进同一笔事务。

事务就是把一组数据库操作作为一个整体处理。 这里最需要理解的是原子性：这组修改要么一起提交，要么一起撤销，不留下只成功一半的结果。PostgreSQL

add / flush / commit / rollback 的分工

注意，add() 不等于“修改对象属性”，也不等于“立即执行 INSERT”。对于已经由 Session 查询出来的对象，修改后一般不需要再 add() 一遍。SQLAlchemy 文档

SQL 已经执行，为什么还能回滚？

假设数据库原本没有 r001，下面各步都成功执行：

run = Run(id="r001", status="pending")
db_session.add(run)
await db_session.flush()

run.status = "running"
await db_session.flush()

await db_session.rollback()

按顺序看：

两次 flush 仍然属于同一笔事务。 flush 不会自动把一笔事务拆成两笔。 数据库具有撤销未提交事务的能力，不是 SQLAlchemy 在出错后自己补一条 DELETE。SQLAlchemy 文档

理解回滚时，先分开两个阶段：

执行 SQL 和最终提交，是两个不同阶段。

“提交后保存”也不代表数据以后不能再改。它表示这笔事务已经确认，不能再靠后续事务的 rollback 撤销它。

例如，事务一提交了 r001，事务二插入 r002 后回滚。结果是 r001 保留，r002 不保留。同一个 Session 可以先后管理多笔事务，不是“一生只有一笔”。SQLAlchemy 文档

为什么有时要先 flush，不直接 commit？

因为你可能需要先执行 SQL，再继续完成同一笔事务。

例如，Message 的主键由数据库生成。先插入并 flush，取得生成的 ID，再用这个 ID 创建相关数据。最后一起 commit。这样中途失败时，仍然可以一起回滚。SQLAlchemy 文档

没写 flush，也不一定没执行 SQL

SQLAlchemy 有 自动 flush ：在默认配置下，执行某些 ORM 查询前，会先把已有的待写入变化 flush 到数据库。commit 也会处理尚未 flush 的变化。

因此：

• 没写 commit()，不等于数据库没执行过 SQL。
• 没显式写 flush()，也不等于没有发生 flush。 SQLAlchemy 文档

退出 async with，到底会不会提交？

要看进入的是什么上下文，不能只看见 async with 就判断会提交。

这里的 SessionFactory 特指通过 async_sessionmaker(...) 配置好的 AsyncSession 创建器。

写法 / 正常退出时的行为

async with SessionFactory() as db_session:

关闭 Session，但不自动替你提交；未提交事务会被清理

async with SessionFactory.begin() as db_session:

管理事务；代码块正常结束时提交，异常向外抛出时回滚，最后关闭 Session

async with database.transaction() as db_session:

项目自己的封装，必须看实现

所以， 业务代码里没出现 commit，可能是外层代为提交，也可能根本没提交。 读代码时，要继续往外层追踪提交和回滚的位置。SQLAlchemy 事务上下文文档

4. Agent 的一次任务，不应该直接等同于一笔数据库事务

事务解决的是“一组数据库修改不能只完成一半”。Agent 任务还包括调用模型、执行工具、等待外部服务。

在我们的简化场景中，可以这样安排：

接收任务
   │
   ▼
短事务一
创建 Turn、Run、用户 Message，更新会话状态
   │
   └─ 提交
        │
        ▼
调用 LLM、执行工具
不让前面的数据库事务一直开着
        │
        ▼
短事务二
保存结果，更新 Run 和相关状态
   │
   └─ 提交

这个拆法的理由有两层。

先看一致性。 创建 Turn、Run、用户 Message 为什么放一起？主要因为它们在业务上应该一起成立。执行时间短不是根本原因，而是让这笔事务容易保持短小的条件。

再看等待成本。 事务已经使用数据库连接后，跨着几十秒的模型调用不结束，就可能长期占用连接及事务资源；涉及写入时，还可能影响其他请求。Session 文档也建议把事务范围与具体数据库工作对应起来，而不是无限延长。SQLAlchemy 文档

还有一个不能忽略的事实：

数据库 rollback 不能撤销普通的文件写入，也不能追回已经发出去的邮件。

数据库事务所保护的是参与这笔事务的数据库操作。外部工具并不会因为被写在同一个 Python 代码块里，就自动加入数据库事务。PostgreSQL

因此，任务开始时的数据库事务提交成功，只能说明“任务已被正确记录”。后续 LLM 仍然可能失败。这时应再用一笔事务记录失败，而不是幻想回滚已经提交的接收记录。

即使把外部操作写在事务代码块里，保护范围也不会随之扩大：

数据库事务仍然能保护其中的数据库修改，但不能因此把整个 Agent 任务变成可统一撤销的操作。

5. Repository：给数据库操作起业务名字

不用 Repository，你完全可以直接调用 SQLAlchemy Session。

当多处业务都要查询同一种数据时，就会重复写查询条件、执行查询、处理结果。Repository 把这些操作集中到一起，让业务代码通过 get_session()、add_run() 这样的名字调用它。它是一种项目组织方式，不是使用 ORM 的必要条件。Cosmic Python

下面只展示 Repository 的构造函数和查询方法。后面例子使用的新增方法在此省略：

class StateRepository:
    def __init__(self, db_session):
        self.db_session = db_session

    async def get_session(self, session_id):
        return await self.db_session.get(AgentSession, session_id)

构造函数的输入是一个已经存在的 db_session。它只是把这个对象的引用保存下来， 没有另建数据库 Session 。

get_session() 的输入是 Agent 会话 ID。它用保存下来的数据库 Session 查找记录，返回 AgentSession 对象或 None。这个方法不是在数据库里新建会话。

Service、Repository、Session、Model 怎么分工？

这是我们采用的分工方式，不意味着所有项目都必须有四个独立的类。Service 也可以只是一个业务函数。Repository 的价值是集中数据访问代码，而不是单纯多套一层名字。Cosmic Python

把一次数据库操作逐步读清楚

为解释下例，先明确两个教学假设：database.transaction() 正常退出会提交，异常向外抛出时回滚；Repository 的新增方法使用传入的 Session，但不自行 commit。这里的封装行为需要回到实际项目中核对。

async with database.transaction() as db_session:
    repository = StateRepository(db_session)
    chat_session = await repository.get_session(session_id)
    if chat_session is None:
        raise ValueError("会话不存在")

    await repository.add_run(run)
    await repository.add_message(message)
    chat_session.status = "running"

输入是会话 ID，以及已经构造好的 Run、Message 对象。

第一行取得这次数据库工作使用的 Session。第二行让 Repository 使用它。随后查出 Agent 会话记录，检查是否存在，再登记需要保存的 Run 和 Message。

最后一行修改的是 业务数据对象 chat_session。这个对象仍由当前 db_session 管理，因此不必专门调用 repository.update_session() 才能被跟踪。

在上述假设下，正常退出时，这几项修改才一起提交。

这里还要守住一个边界： 本例由上层决定事务范围，所以 Repository 不应在每个新增方法里自行 commit。 否则，第一个方法已经提交，后面失败时就无法把整组修改一起撤销。这是事务范围推出来的设计要求，不是 Repository 名字自带的规则。SQLAlchemy 文档

6. 表之间的关系：归属、约束、对象访问、联表查询

在我们的示例里：

• Agent Session 表示一次持续的会话。
• Turn 表示这次会话中的一轮用户提交。
• Run 表示围绕这一轮任务的一次执行。
• Message 保存用户消息、助手回复等内容。

这只是本文的业务定义。 不能仅凭这些类名，断定任何 Agent 项目都这样设计。尤其是一个 Turn 是否允许多个 Run，需要看项目对重试、重新执行的定义。

外键放在哪里？

假设表里有：

这表示两个 Turn 都属于会话 s001，即一个 Session 对多个 Turn。

session_id 列记录归属。给它加上指向 sessions.id 的 ForeignKey 外键约束 ，数据库还能检查非空引用是否有效，防止 Turn 指向不存在的 Session。若不允许这个字段为空，还需要相应的非空约束。SQLAlchemy 文档

使用 SQLite 时，还要确认 每个数据库连接都开启了外键检查 。Model 中声明了 ForeignKey，还需要数据库里实际存在相应约束，并在运行时启用检查，才能依赖它拒绝非法引用。TARS-Agent 的具体配置仍需核对。SQLAlchemy 文档

relationship 为什么又出现了？

数据库里已经通过 turns.session_id 记录关系。但 Python 代码还希望方便地访问相关对象：

turn.session 表示这个 Turn 对应的会话；chat_session.turns 表示这个会话下的 Turn 集合。

这就是 relationship：ORM 对象之间的关联配置 。它不是在 sessions 表中再增加一个“Turn ID 列表”，也不意味着相关数据已经提前查出来了。SQLAlchemy 文档

JOIN 是另一件事

JOIN 是在查询时，根据条件组合不同表中的记录。例如用 sessions.id = turns.session_id，查出会话及其相关 Turn。

因此，四件事不要混：

有外键，不代表每次查询自动 JOIN；有 relationship，也不代表关联数据已经在内存里。 SQLAlchemy 关联加载文档

7. 查询性能：既不要多拿，也不要反复拿

把十万条 Message 全部读进 Python，会增加数据读取、结果传递和 ORM 对象构造的成本，也可能消耗大量内存。但是否真的“内存爆炸”，还取决于每条消息大小和运行环境，不能仅凭十万条这个数量判断。

先限制本次真正需要的数据

查询一个会话的最近消息，可以分成三个要求：

先按 session_id 筛选，再按时间倒序排列，最后用 LIMIT 限制条数。

例如，LIMIT 50 的意思是本次最多返回 50 条，不是“先查出十万条，再在 Python 中取前五十条”。分页则允许用户继续分批读取其他数据。筛选、排序和限制条数都是数据库查询可以完成的工作。SQLAlchemy 文档

查询范围要根据业务场景决定。界面只展示最近消息，与批量审查完整历史，需要的查询范围不同。

懒加载到底发生在什么时候？

假设 Run.messages 配置成懒加载，而且此次查询没有加载它：

第一次只查出了 50 个 Run。此时，Run 的 ID、状态等已加载字段可以读取，但对应 Message 集合还没取回。

当代码真正访问某个 run.messages 时，ORM 才尝试查询这个 Run 的 Message。这叫 懒加载：关联数据等用到时再取 。它由映射配置和本次查询方式决定，不是数据库看到点号就自动发挥的能力。SQLAlchemy 文档

为什么需要这种方式？因为只展示 Run 状态时，根本不需要读取 Message。

但逐个访问时，会出现另一种问题：

一次查询：取得 50 个 Run
接着访问第 1 个 Run 的 messages：再查一次
接着访问第 2 个 Run 的 messages：再查一次
……
总计可能是 1 + 50 = 51 次查询

这就是 N+1 。这里的 runs 是一组 Run 对象，不是“一个包含整张表的 ORM 对象”。额外的 N 次查询也不一定是 JOIN，常见情况是分别查询 messages WHERE run_id = ...。SQLAlchemy 文档

预加载是不是“一次大 JOIN”？

不一定。

预加载 是事先告诉 ORM：“后面会用到这些关联数据，请提前加载。”

它可以通过 JOIN 一起取，也可以先查询 Run，再批量查询这些 Run 的 Message。重点是避免逐个重复查询，不是强求只执行一条 SQL。

还要注意，单纯在查询中写了 JOIN，不一定就完成了 run.messages 的填充；需要相应的 ORM 加载配置。SQLAlchemy 文档

选择加载方式时，可以先问页面真正要显示什么：

• 只看 Run 状态，不加载 Message。
• 确定要读取 Message，就考虑批量加载。
• 只要最后一条回复，也不应该默认把全部历史回复都取回来。

异步边界： 前面的懒加载时间线是在说明 N+1 的原理。普通 AsyncSession 代码直接访问尚未加载的 run.messages，可能因为隐式数据库 I/O 而报错，不一定顺利执行出 51 次查询。异步代码通常应明确加载所需数据。SQLAlchemy 文档

索引能帮什么忙？

索引是数据库额外维护的查找结构。合适的索引可以减少查找、筛选、排序时需要处理的数据。

例如，经常按 session_id 查 Message，就可以考虑相关索引。数据库还可以利用合适的多列索引支持组合条件和排序。SQLite

但索引不是“让所有 SQL 底层都变快”。它占空间，写入数据时也需要维护。即使每次查询因索引变快了，N+1 所产生的大量独立查询仍然存在。

所以， 索引是否合适、返回多少数据、执行多少次查询，是三个不同问题。 SQLAlchemy 文档

8. 幂等与并发：不能重复，也不能只做一半

先把几个 ID 分开

在本文的示例中：

client_message_id 不是天然等于 Message 表的主键，也不会自己“创建 Turn”。 后端业务代码收到新提交以后，才决定创建 Turn、Run 和 Message。

本文约定：同一次提交的网络重试继续使用原来的 client_message_id。这样，后端才有依据识别“这是刚才那次提交，不是新任务”。

这里的 幂等 ，指同一次操作重复请求，不额外产生本不该重复的业务效果。我们希望网络重试返回已有任务，而不是再创建第二套任务。

数据库具体长什么样？

一种示例设计是把编号保存在 turns 表：

后端再次收到 c001，就能先找到 t001，再找到已有任务。

给这一列加 UNIQUE，就是要求数据库不允许另一条记录也使用相同的非空编号。 这是额外的唯一约束，不是把这一列变成主键。 如果每次提交都必须有编号，还要把该列设为非空；唯一约束与非空约束负责不同的事。SQLAlchemy 文档

这里先采用“全表唯一”的示例。实际也可以规定“同一个会话内唯一”，通过 (session_id, client_message_id) 的组合唯一约束实现。唯一范围要与业务要求一致，不能混用。TARS-Agent 实际采用哪一种， 目前未确认 。SQLAlchemy 文档

为什么“先查一下”还不够？

两个请求不需要在完全相同的一纳秒查询，也可能出问题：

时间向下

请求 A                         请求 B
查 c001：没有
                               查 c001：没有
准备插入
                               准备插入
插入
                               插入

两个查询都发生在已有结果对另一个请求可见之前，就可能都查不到。

因此，问题在于：

查询是否存在和执行插入，是分开的操作。其他请求可以在它们之间执行。

不需要用“Python 被翻译成汇编后重排”来解释。即使每个请求内部完全按源码顺序执行，上面的交错仍然成立。

UNIQUE 和事务怎样配合？

UNIQUE 约束的是：这个键不能重复。

事务保证的是：这一组数据库修改不能只提交一半。

例如，请求 B 因为重复编号而创建 Turn 失败，它在同一事务中已经做过的其他相关修改也应该一起回滚。这样既不重复，也不留下半套记录。普通事务本身并不自动把“先查再插”变成可靠去重。SQLAlchemy 文档

处理流程可以理解为：

收到请求
   │
   ├─ 已存在相同幂等键 → 返回已有任务
   │
   └─ 尚不存在 → 在事务里创建相关记录
                    │
                    ├─ 提交成功 → 返回新任务
                    │
                    └─ 发生目标唯一约束冲突
                           → 回滚失败事务
                           → 在可用的新事务中查询已有任务

冲突后不能忽略异常，直接继续使用当前 Session。 如果 flush 因约束等错误失败，即使底层数据库事务已经回滚，也要先由事务上下文或显式 await db_session.rollback() 完成 Session 的恢复，再继续使用它。也不能把所有数据库错误都当成重复提交，必须确认冲突来自预期的幂等约束。SQLAlchemy 文档

最后还有一个范围限制： Turn 的唯一约束只保证对应数据库记录不重复，不自动保证邮件、文件操作等外部工具永远只执行一次。 外部执行需要单独设计防重。这正是“数据库事务不等于整个 Agent 任务”的延伸。

9. 异步和迁移：核心思想没变，但使用边界要清楚

AsyncSession：等待数据库时，不堵住其他协程

AsyncSession 仍然管理 ORM 对象和事务。主要区别是，可能涉及数据库等待的操作通过异步接口执行。

db_session.add(run) 主要登记对象，通常不需要数据库 I/O，所以不用 await。flush() 可能发送 SQL 并等待结果，所以异步版需要 await。SQLAlchemy 文档

await 的作用不是“防止 CPU 卡死”，也不是“自动新建线程”。当前协程确实需要等待时，它可以把执行权交回事件循环，让其他就绪协程继续运行。它不保证 SQL 本身执行得更快。Python documentation

也不是所有数据库访问都经过网络。远程数据库可能有网络等待；SQLite 通常访问本地数据库，aiosqlite 提供与 asyncio 配合的接口。 是否使用 await，不能用“有没有网络”来判断。 SQLAlchemy 文档

async_sessionmaker 则是 配置好的 AsyncSession 创建器 。它用于创建多个独立 Session，不是一个让所有请求共用的 Session。SQLAlchemy 文档

并发任务为什么要使用各自的 AsyncSession？

一个 AsyncSession 不能被多个并发任务同时使用。它保存着当前事务、受管理对象和待写入变化；并发共用时，操作会互相干扰。一个任务 rollback 可能影响另一个任务的未提交工作，也可能直接遇到并发使用错误。SQLAlchemy 文档

决定是否分开 Session，要看数据库工作的关系：

看它们是不是独立并发的数据库工作，不是看函数处在第几层。

独立并发任务通常各自使用 Session。相反，几个函数按顺序合作完成同一笔事务，就应该使用同一个 Session，否则会把本来需要一起提交的操作拆开。

长期运行的 Agent 也不必一直持有同一 Session。可以在接收任务、记录中间状态、保存结果时，分别使用短小的数据库工作范围。

Alembic：让已有数据库按步骤改变结构

假设原来 runs 表只有 ID 和状态，后来代码需要增加 error_message。

修改 Python Model，不会让已有表自动增加这一列。create_all() 主要创建缺失的表，也不是自动修改已有表结构的工具。SQLAlchemy 文档

这里需要 migration，数据库迁移步骤 ：明确说明旧结构怎么变成新结构。

Alembic 管理这些迁移脚本及版本关系。upgrade 执行向新版本迁移的步骤；downgrade 执行向指定旧版本回退的步骤，不一定只退一个版本。自动生成的迁移也需要检查，不是改完 Model 就自动、安全地改变数据库。Alembic

用“数据库的 Git”来记住版本管理的作用时，要注意它的边界：

Alembic 不是业务数据的版本快照，也不是备份工具。

如果迁移删除了一列，回退时重新加回同名列，并不会自动找回原来的列数据。downgrade 能做什么，取决于回退脚本；不能把“有回退脚本”当成“数据一定可恢复”。Alembic

复习时容易混淆的地方

下面把容易混淆的说法放在一起，复习时可以用来检查自己的理解。

这张表是前面知识的压缩版。最值得反复区分的三组是： 对象与记录、flush 与 commit、原子性与幂等性。

把这些概念连起来

用户提交任务
      │
      ▼
Service／业务函数
决定做什么、按什么顺序做、哪些修改一起提交
      │
      ▼
Repository
把查询和保存写成 get_session、add_run 等方法
      │
      ▼
SQLAlchemy Session／AsyncSession
管理 ORM 对象，执行查询，组织 flush、commit、rollback
      │
      ├── 管理的对象遵循 Model 定义的映射
      │       Run 对象 ↔ runs 表中的记录
      │
      └── 通过 Engine 提供的连接访问数据库
                      │
                      ▼
                  数据库
          SQL 操作记录，事务控制提交
          UNIQUE 防重复，ForeignKey 检查引用

另一个方向：
Model 的表结构定义变了
      │
      ▼
Alembic 迁移脚本
      │
      ▼
修改已经存在的数据库结构

这张图表达的是本文的教学分工，不是要求项目必须照着建立同名文件或类。

十个常见问题速查

1. ORM、Model、Engine、Session 分别做什么？

ORM 让程序通过 Python 对象操作数据库记录。Model 定义对象与表的映射，Engine 提供数据库连接，Session 管理 ORM 对象和数据库工作。业务里的 Agent 会话是一条数据，和 SQLAlchemy Session 是两个概念。

2. 为什么修改 run.status 后能更新数据库？

通过当前 Session 查询到的 Run 对象受它管理，映射属性的变化会被跟踪。flush 时，SQLAlchemy 才把待写入变化执行成 SQL；事务提交成功后，这次修改才被确认。单纯创建对象或给属性赋值，都不等于已经保存。

3. flush 和 commit 有什么区别？

flush 在当前事务中执行待写入的新增、修改和删除，不会因此提交事务。commit 会先 flush 剩余变化，再提交。普通 Session 上下文退出只会关闭 Session；是否自动提交，要看有没有事务上下文或项目封装。

4. 为什么不把整个 Agent 任务放进一笔事务？

需要一起成立的数据库修改适合放在短事务中。LLM 和外部工具调用可能耗时很长，数据库回滚也撤销不了已经发生的文件写入或外部 API 操作。接收任务和保存结果可以分别提交，执行失败后再记录失败状态。

5. Repository 是必须的吗？

它是项目组织数据访问代码的一种方式，把查询细节封装成 get_session()、add_run() 等业务方法，不是 SQLAlchemy 的必需层。本文采用上层统一控制事务的设计，因此 Repository 使用传入的 Session，不在各方法里自行提交；其他项目要看具体实现。

6. ForeignKey 和 relationship 有什么区别？

ForeignKey 描述数据库外键约束；相应约束必须实际创建并启用，数据库才能检查非空引用是否合法。relationship 配置 Python 对象之间的关联访问，不会自动多存一个 ID 列表，也不保证关联数据已经加载。SQLite 还需要为连接开启外键检查。

7. 什么是 N+1，怎样避免？

先查询 N 个主对象，再为每个对象各查一次关联数据，就可能出现 1+N 次查询。确定会用到关联数据时，可以使用合适的 ORM 预加载策略。预加载可以通过 JOIN 或批量查询实现；普通 JOIN 不自动保证 relationship 已被填充。AsyncSession 还要避免隐式数据库 I/O，不能照搬同步懒加载写法。

8. 为什么重复请求不能只靠先查询？

查询与插入是两个操作，两个请求可能都先查到不存在。幂等键用于识别同一次提交，数据库唯一约束负责防重复；相关记录放进同一事务，是为了防止只提交一半。发生预期唯一约束冲突时，先正确回滚失败事务，再在可用的新事务中查询已有任务。数据库防重不会自动保证外部工具只执行一次。

9. 多个协程能共用一个 AsyncSession 吗？

多个并发任务不能同时共用一个 AsyncSession，因为它保存着事务和对象状态。每个独立并发任务应使用自己的 Session；几个函数顺序合作完成同一事务时，可以使用同一个传入的 Session。判断依据是工作的并发关系和事务边界，不是函数处在第几层。

10. Alembic 是数据库的备份工具吗？

Alembic 管理数据库结构迁移。修改 Model 不会自动改变已有表，迁移脚本需要检查后执行。downgrade 执行回退步骤，但不会自动恢复被删掉的业务数据，因此不能替代备份。

用四个问题检查理解

可以先不看后面的答案，用自己的话解释每个结果。

1. 对象修改有没有保存？

代码在一个普通的 async with SessionFactory() 中查询 Run，修改状态并成功执行了 flush。没有调用 commit，也没有任何自动提交封装。退出代码块后，这次修改会保留吗？

2. 任务失败，哪些事情仍然成立？

创建 Turn、Run、用户 Message 的事务已经提交。工具随后成功写了一个文件，但保存最终回复的事务失败并回滚。哪些结果仍然存在？哪些没有保存？

3. 查询是不是越少越好？

一个页面只显示最近 50 个 Run 的状态。另一页面还要读取这些 Run 的全部回复内容。两者应该采用相同的关联加载方式吗？

4. 防重和回退是不是“万能保险”？

数据库给提交编号加了 UNIQUE，是否就能保证工具永远只执行一次？Alembic 有 downgrade，是否就能保证删除过的数据一定找得回来？

参考答案

第 1 题：不会保留。

普通 Session 上下文退出不会自动提交。虽然 flush 已经执行 UPDATE，但事务仍未提交；Session 关闭时会清理未提交事务。SQLAlchemy 文档

第 2 题：接收记录和文件仍然存在，最终回复没有成功保存。

Turn、Run、用户 Message 已经由前一笔事务提交，后一笔事务的 rollback 不会撤销它们。文件属于数据库事务之外的操作，也不会因此消失。最终回复及同一失败事务中的状态修改未能提交，因此还需要后续处理来记录任务结果。PostgreSQL

第 3 题：不应该只为了统一写法而相同。

只看状态的页面不需要读取回复。确定需要全部回复的页面，可以考虑提前批量加载，避免逐个查询。查询次数少是目标之一，返回数据量是否合适也同样重要。SQLAlchemy 文档

第 4 题：两者都不能这样保证。

UNIQUE 约束对应的是数据库中的键，不是外部工具执行次数。Alembic 执行的是迁移脚本，不是业务数据备份；把被删的列重新建回来，也不代表原来的列数据随之恢复。SQLAlchemy 约束文档、Alembic 操作参考

回到项目里继续验证

把概念对应到 TARS-Agent 的实现

读源码时，可以带着三个问题：Model 实际定义了哪些字段与约束？事务封装在哪里提交和回滚？Repository 方法会不会自行提交？另外，SQLite 外键检查是否开启、幂等键采用什么唯一范围，也需要核对。

这些具体实现目前还没有在本文中确认。把实际代码位置和执行结果补进笔记后，才能区分通用原理与项目自己的选择。

做一次最小的数据库实验

可以在独立的测试数据库中依次观察四种情况：新建记录并 flush 后回滚；提交后重新查询；修改对象后不提交就关闭 Session；触发重复键冲突后正确回滚，再查询已有记录。

每一步都同时看两件事：日志里执行了什么 SQL，最后重新查询得到什么数据。先在测试数据上验证，不要用生产数据练习回滚或迁移。这些实验还需要实际运行，不能仅凭本文的说明就当作已经验证过。