Agent 调用“创建任务”工具后超时重试却生成了两条一样的任务。这类故障不一定是模型忘了自己做过什么可能第一次写入已经提交只是调用方没有收到结果。让模型在提示词里记住“不要重复操作”无法补上数据库提交与响应到达之间的空隙。下面把副作用与可回放结果写在同一个 SQLite 事务里验证重复调用、参数冲突、异常回滚和两连接竞争。一、重试要复用操作身份operation_key 应标识一次业务操作而非某次网络尝试。调度器应在执行前生成并持久保存它后续重试继续使用原键。模型重新生成的 tool call ID 是否稳定需要单独核实不能直接当作业务幂等键。本例用 tenant 与 operation_key 共同定位操作另存工具名、版本和参数。相同键且参数相同返回已有结果相同键但参数变化拒绝执行真正的新操作使用新键。输入只允许非空字符串没有提供通用 JSON 归一化算法。二、为什么两张表要一起提交jobs 是实际创建的本地任务receipts 保存参数和结果。先写 jobs 再单独提交 receipts中间失败仍然可能重复创建先提交成功标志再写 jobs则可能留下没有真实任务的“成功”。本例使用 BEGIN IMMEDIATE在同一个事务内检查旧回执、创建任务并保存结果。SQLite 同一时刻只允许一个写事务竞争可能等待或报 busy这不是无限并发保证。机制见 SQLite 事务文档。三、完整可运行示例保存为 idempotent_tool_demo.py执行python idempotent_tool_demo.py。需要 Python 3.12 或更新版本以使用 autocommit 参数本轮实际环境是 Python 3.13.13、SQLite 3.51.2。实验只创建临时数据库不调用模型或外部服务。这里显式使用 autocommitTrue 加 SQL 的 BEGIN、COMMIT、ROLLBACK。不要把它与其他事务模式下的 Connection.commit() 用法混在一起参考 Python sqlite3 事务控制。Local SQLite effects only; does not call an LLM or external service.importjsonimportplatformimportsqlite3importtempfileimportthreadingfromconcurrent.futuresimportThreadPoolExecutorfrompathlibimportPathdefconnect(path):returnsqlite3.connect(path,autocommitTrue,timeout5)defsetup(path):conconnect(path)try:con.executescript( CREATE TABLE jobs(id INTEGER PRIMARY KEY, tenant TEXT NOT NULL, title TEXT NOT NULL); CREATE TABLE receipts( tenant TEXT NOT NULL, operation_key TEXT NOT NULL, payload TEXT NOT NULL, result TEXT NOT NULL, PRIMARY KEY(tenant, operation_key) ); )finally:con.close()defcreate_job(path,tenant,operation_key,title,fail_atNone):ifnotall(type(x)isstrandxforxin(tenant,operation_key,title)):raiseValueError(non-empty strings required)payloadjson.dumps({tool:create_job,version:1,title:title},sort_keysTrue,ensure_asciiFalse,separators(,,:))conconnect(path)try:con.execute(BEGIN IMMEDIATE)oldcon.execute(SELECT payload, result FROM receipts WHERE tenant? AND operation_key?,(tenant,operation_key)).fetchone()ifoldisnotNone:ifold[0]!payload:raiseValueError(operation key reused with different arguments)con.execute(COMMIT)returnjson.loads(old[1])curcon.execute(INSERT INTO jobs(tenant, title) VALUES(?, ?),(tenant,title))result{job_id:cur.lastrowid,title:title}iffail_atbefore_receipt:raiseRuntimeError(injected failure before receipt)con.execute(INSERT INTO receipts VALUES(?, ?, ?, ?),(tenant,operation_key,payload,json.dumps(result,ensure_asciiFalse)))con.execute(COMMIT)iffail_atafter_commit:raiseTimeoutError(injected lost acknowledgement after commit)returnresultexceptBaseException:ifcon.in_transaction:con.execute(ROLLBACK)raisefinally:con.close()defcount_jobs(path):conconnect(path)try:returncon.execute(SELECT COUNT(*) FROM jobs).fetchone()[0]finally:con.close()defmust_raise(kind,call):try:call()exceptkind:returnraiseAssertionError(fexpected{kind.__name__})defmain():print(fPython{platform.python_version()}/ SQLite{sqlite3.sqlite_version})withtempfile.TemporaryDirectory()astmp:pathPath(tmp)/demo.sqlite3setup(path)firstcreate_job(path,team-A,op-1,整理采集报告)assertcount_jobs(path)1print(PASS first call: one job)assertcreate_job(path,team-A,op-1,整理采集报告)firstassertcount_jobs(path)1print(PASS retry through new connection: same receipt, still one job)must_raise(ValueError,lambda:create_job(path,team-A,op-1,改了内容))assertcount_jobs(path)1print(PASS same key with different arguments: rejected)must_raise(RuntimeError,lambda:create_job(path,team-A,op-2,第二个任务,before_receipt))assertcount_jobs(path)1print(PASS failure before receipt: effect rolled back)create_job(path,team-A,op-2,第二个任务)assertcount_jobs(path)2print(PASS retry after rollback: one new job)must_raise(TimeoutError,lambda:create_job(path,team-A,op-3,第三个任务,after_commit))assertcount_jobs(path)3recoveredcreate_job(path,team-A,op-3,第三个任务)assertrecovered[job_id]3andcount_jobs(path)3print(PASS lost acknowledgement: committed result replayed)create_job(path,team-B,op-1,整理采集报告)assertcount_jobs(path)4print(PASS separate tenant scope: independent operation)barrierthreading.Barrier(2)defcontender():barrier.wait(timeout5)returncreate_job(path,team-A,op-4,并发任务)withThreadPoolExecutor(max_workers2)aspool:futures[pool.submit(contender)for_inrange(2)]results[f.result(timeout10)forfinfutures]assertresults[0]results[1]andcount_jobs(path)5print(PASS two concurrent connections: one effect, same result)print(8 checks passed)if__name____main__:main()2026-09-12 的实际输出Python 3.13.13 / SQLite 3.51.2 PASS first call: one job PASS retry through new connection: same receipt, still one job PASS same key with different arguments: rejected PASS failure before receipt: effect rolled back PASS retry after rollback: one new job PASS lost acknowledgement: committed result replayed PASS separate tenant scope: independent operation PASS two concurrent connections: one effect, same result 8 checks passed四、从结果里看三种失败状态第一种是回执写入前发生异常。事务回滚后没有残留任务使用同一键重试才创建一条新记录。第二种是提交完成后调用方没有拿到结果。代码用提交后的 TimeoutError 模拟这种“确认丢失”下一次连接查到回执返回原 job_id任务总数不变。这是故障注入没有真的断网或杀进程。第三种是两个连接同时拿相同键执行。本轮用两线程、各自独立连接竞争得到相同结果最终只增加一条任务。这证明本轮小实验成立不是吞吐压测也没有覆盖所有锁超时场景。五、外部副作用不能被这段代码包住如果工具改为发送邮件、调用第三方接口或写另一套数据库本地 ROLLBACK 无法撤回这些动作。把外部请求塞进事务不会自动获得端到端的“恰好一次”。接外部系统时应检查对方是否接受稳定幂等键、是否能按操作 ID 查询结果以及保留多长时间。确认超时后先查已有状态只有能明确区分“未执行”和“已执行”时才能制定安全重试策略。无法确定的操作要保留待核对状态不能直接标成失败再创建一个新操作。收据清理也属于业务设计如果过早删除迟到重试就可能再次执行。多租户场景还必须从可信登录上下文取得 tenant并在结果回放前检查当前调用者的访问权示例没有实现鉴权系统。本文由 AI 辅助起草与校核8 项检查已在上述本地环境执行。示例验证同一 SQLite 数据库内的原子提交与结果回放不宣称解决了跨服务事务、断电恢复或真实模型工具调用的全部问题。