AI 智能体导入数据怎样避免半批成功？用事务验证失败回滚

AI 智能体批量导入数据时，不能一部分失败却仍报“导入完成”。如果契约是全批成功或全批失败，数据库事务与失败后的实际行数核对要同时存在。

为AI智能体开发批量数据导入的读者。本文使用 Python 3.11 或以上版本的标准库，先验证一组人工构造的小样本。AI 负责提出实现或解释差异，结果由本地检查决定；这里没有把模型回答当成真实运行结果。

验证数据导入的全批回滚前，先定规则

示例表以id为主键，初始已有base。一批输入a与base，第二条冲突时应回滚第一条a。

异常发生后查询数据库，比只看异常提示更能证明没有残留。导入成功也应回读ID与行数。

示例使用SQLite内存库，无生产数据，不让模型提供任意SQL。字段映射与插入语句由应用固定，数据用参数传入。

给 AI 的输入要包含什么

把下面这份输入说明和你的实际样本一起交给可用的 AI 编程助手。示例只含虚构数据；对真实材料先去除账号凭据和个人信息。

实现固定表记录导入，参数化executemany，使用事务；批内任一主键冲突全批回滚。给第二条冲突的样本并查询证明第一条未写入。不得catch后返回成功。

要求模型保留检查条件，并把它认为缺少的业务定义列出来。若回答改变了输入字段、忽略异常分支或直接删除原材料，先要求修正，再运行。提示词的作用是缩小任务范围，验收仍以代码和数据为准。

保存并运行最小验证程序

下方程序把关键规则和验证用例放在同一个可复跑示例中，便于先理解输入如何变成输出，再用它检查 AI 给出的实现。

新建一个空目录，把下面代码保存为 check.py，在该目录打开终端，运行 python -X utf8 check.py。代码自带示例输入，不需要安装第三方库。

import sqlite3
db=sqlite3.connect(':memory:')
db.execute('create table records(id text primary key, value text not null)')
db.execute('insert into records values(?,?)',('base','old'));db.commit()
try:
    with db:
        db.executemany('insert into records values(?,?)',[('a','new'),('base','conflict')])
except sqlite3.IntegrityError: print('batch_failed',True)
assert db.execute('select id from records').fetchall()==[('base',)]
print('after_failure',db.execute('select id from records').fetchall())
with db: db.executemany('insert into records values(?,?)',[('a','new'),('b','new')])
assert db.execute('select count(*) from records').fetchone()[0]==3
print('after_valid_count',3)

怎样判断结果符合要求

失败后只有base，正常批次后共有3条记录。batch_failed是真实约束失败，不是模拟“返回错误但继续写”。

下方是这份最小示例在本地执行得到的输出。它验证示例程序与断言的关系，不代表任何 AI 模型一次就能生成同样代码，也不构成性能或生产可靠性结论。

batch_failed True
after_failure [('base',)]
after_valid_count 3

连接上下文退出不自动关闭连接，事务范围也要避开已有未提交业务操作。

不要为了避免冲突使用INSERT OR IGNORE后仍报全批导入成功，它改变了验收语义。

哪些失败必须停下来处理

业务可能允许逐行成功，此时必须返回每行状态与成功失败数量，而不是套用全批事务。

外部文件写入与数据库事务无法天然一起回滚，要另设暂存与提交流程。

接入自己的任务前再核对一次

对实际导入增加输入哈希、批次ID与重复请求口径，避免超时重试产生双份记录。

把冲突样本保留为回归，验证错误日志不会泄露原始个人字段。

资料与适用范围

sqlite3连接上下文可管理事务提交与回滚；约束失败需要保留异常。以下链接核对于 2026-10-03；运行环境及额外依赖按本文前述说明。

• https://docs.python.org/3.11/library/sqlite3.html

相关基础可阅读 智能体数据怎么增量同步？用 SQLite 核对版本、重跑与回滚。本文的重点是验证数据导入的全批回滚，可以把两项检查作为不同步骤保留。