先问一个问题:你有没有遇到过这种情况——一个 Agent 任务跑了大几十分钟,眼看快出结果了,进程突然崩掉。日志里只剩一句agent execution terminated due to error。你第一反应是“重新跑一遍就行了”,于是你重试,结果发现它已经扣了两次款、创建了两份工单、把同名文件重复提交到了三个仓库。
这是我这两年看 Agent 项目落地时遇到最多的一类事故。比“任务失败”可怕得多的是“任务状态不可知”——你不知道它到底做完了没有、做到哪一步了、还能不能安全重试。相比普通软件的崩溃,Agent 的崩溃有一点本质区别:它背后接的是真实世界的操作,支付、发消息、改配置、提 PR、删数据。每一个操作都像泼出去的水,收不回来的。所以“Agent 崩溃之后怎么办”这件事,核心就一句话:不重复执行、不假装执行过。
这篇文章我想和你好好聊聊这件事。我会从幂等设计、执行确认、状态持久化几个维度展开,讲清楚 Agent 崩溃恢复时真正要面对的问题,也给出我们项目里实际在用的方案和踩过的坑。无论你是在做 AI Agent 开发,还是正准备把 Agent 接到自己的业务系统里,这篇内容都能帮你少走几个月的弯路。
1. 先想清楚:Agent 崩溃后到底丢掉了什么?
很多人把 Agent 崩溃理解成“程序挂了,重新拉起来就行”。但 Agent 崩溃真正丢掉的不是进程,而是对任务状态的记忆。这个差别很关键。
1.1 崩溃的本质是“状态丢失”,不只是“进程退出”
一个 Agent 执行任务的过程,本质上是一连串有先后依赖的操作:先读上下文、做规划,然后调用一个工具,拿到结果再决定下一步。每动一步,内存里都有一个“当前进行到哪”的状态。进程一退出,这个状态就全没了。
但如果 Agent 只和内存里的状态打交道,那还不可怕,重启重跑就行。真正可怕的是:它已经对外部世界产生了影响——钱已经扣了、邮件已经发了、文件已经写了。你重启后重跑一遍,这些副作用就会被再次触发,产生重复执行;如果你不重跑,Agent 又没有能力告诉你“我已经做到哪一步了”,那后续所有步骤都只能悬在那里。
所以,崩溃恢复的难点根本不是“进程能不能重启”,而是“进程重启以后,怎么让它知道外部世界的真实状态”。这也是为什么在 Agent 架构里,我建议把状态外置:把任务的进度、每一步的结果、工具的调用记录,都写到独立的存储里,而不是放在 Agent 模型的内存里。模型会忘,存储不会。
1.2 把崩溃分成三类:执行前、执行中、执行后
为了说清楚不同崩溃阶段的处理差异,我们先把“崩溃”这件事拆开来看。我习惯把 Agent 的一次任务执行分成三个阶段:
- 执行前崩溃:Agent 刚拿到任务,还没开始调用任何外部工具。这种情况最简单,重启后直接从头跑,不会产生任何副作用。
- 执行中崩溃:Agent 已经调用了部分工具,但任务没有完成。这里最棘手——有些工具调用成功了,有些失败了,有些可能是“执行了但响应超时”。重试之前,必须搞清楚:哪些步骤已经产生了副作用?
- 执行后崩溃:Agent 已经把结果算出来了,但在写回结果、提交状态、通知用户的环节挂了。这种情况其实很冤,因为活已经干完了,你只是没来得及把“干完了”这件事告诉别人。
这三种崩溃对应三种应对策略:执行前的直接重跑,执行后续接一个“状态修复”动作,执行中的最麻烦,必须靠幂等键和操作记录来区分“哪些步骤需要跳过、哪些需要重放”。
我见过好几个团队栽在执行中崩溃上。他们以为“每次重跑都是全新任务”,结果下游系统的日志里塞满了重复的调用记录。一定要建立起一个意识:Agent 的每一次执行,都应该运行在一个“可能被重放”的预设之下。
2. 不重复执行:给每一步操作都刻上“唯一身份”
想避免重复执行,核心手段就一个词:幂等。听起来挺玄,但其实是分布式系统里的老话题。很多人一看到“幂等”就联想到 C 语言里的幂运算,比如pow(2, 3),其实这是两码事。幂等表示的是“同一个操作执行一次和执行 N 次,对外部世界的影响完全相同”。读操作天然幂等,写操作就很危险——重复创建一个订单、重复发送一条通知,都会产生不同类型的后果。
Agent 的幂等设计,关键不在理论,而在落地。下面是我在项目里验证过比较有效的几个层次。
2.1 幂等键:最朴素的“唯一执行凭证”
幂等键(Idempotency Key)的核心思想特别朴素:每次要执行一个“会产生副作用”的操作时,先给它分配一个全局唯一的 ID,然后把“这个 ID 对应的操作是否已经执行过”记录下来。下游系统再收到同一个 ID 的请求时,直接返回第一次执行的结果,而不是重新执行。
以支付接口为例。前端用户手滑点了两次“确认支付”,如果后端接口不处理幂等,就会创建两笔订单。加上幂等键之后,第二次携带同样requestId的请求到达时,接口发现这个 ID 已经处理过了,于是直接返回第一笔订单的结果,不重复扣款。
在 Agent 场景里,幂等键通常长这样:
- 任务级:
projectA_task_20241216_001 - 步骤级:
task_20241216_001_step_3 - 工具调用级:
call_<uuid4>
键的粒度决定了你能恢复到什么程度。如果只有任务级幂等键,那么一个步骤重复执行了你也发现不了;如果有步骤级的键,就能精确锁定“第几步被重复调用了”。
注意:幂等键不是随便一个 ID 就行,它必须能稳定对应到同一个逻辑操作。如果一个操作的关键参数变了,比如用户从“购买 1 件商品”改成“购买 2 件商品”,那它应该用新的幂等键,否则会把两次不同的操作误判成一次。
2.2 状态机:从“爬起来不知道自己在哪”到“一目了然”
幂等键解决了“同一个操作会不会重复执行”的问题,但它没有回答“整个任务现在进展到哪一步”的问题。这个要靠任务状态机来解决。
我给 Agent 任务设计了这样一张状态流转图,每一步的推进都必须显式更新状态,不能靠“猜”:
pending -> running -> step_1_done -> step_2_done -> ... -> completed | +----> failed | +----> awaiting_confirmation为什么状态机这么重要?因为它给了 Agent 一个“外部锚点”。崩溃恢复时,Agent 不需要去问用户“我刚说到哪了”,直接读取状态存储里的字段,就知道自己是卡在step_3等待确认,还是已经推进到step_5准备写结果。
我见过很多失败的 Agent 项目,状态全靠模型“自觉”在 prompt 里维护。模型一旦忘了自己在哪,整个任务就会从开始重新编排。状态机不是用来限制 Agent 灵活性的,它其实是给“自由发挥”兜底用的——模型可以自由规划路径,但每走一步,都要在一个确定性的存储里“打卡签到”。
2.3 把外部操作也改成天然幂等
幂等键是“客户端主动去重”的方案,但还有一层更省心的思路:把目标操作本身就设计成幂等的。这样即使客户端忘了带幂等键,重复执行也不会造成重复影响。
举几个例子:
- 创建资源:不要用“新增一条数据”这种天然不幂等的接口,改成“如果已存在且参数相同则返回已有数据”的 upsert 语义。
- 状态更新:把“把状态设为 X”这种绝对赋值,设计成可重复执行的。同一个状态反复设置 10 次,结果都一样。
- 发送通知:这是最难幂等化的操作,因为外部用户真的会收到两封邮件。这种场景只能靠“发送记录表”来兜底,每次发之前查一下有没有发过。
所以,幂等设计不是“在调用外部接口时传个参数”那么简单,它需要从接口设计层面通盘考虑。如果你是在开发 Agent 要调用的下游系统,建议优先把接口做成天然幂等;如果你是 Agent 的调用方,那一定得靠幂等键和记录表来守住防线。两层都做,才叫稳妥。
3. 不假装执行过:从“我以为我做了”到“我证明我做了”
重复执行是“做了两遍”,而“假装执行过”是另一类更隐蔽的问题——Agent 以为它做了,但实际根本没有做,或者只做了一半就往下走了。这种问题比重复执行还难查,因为它不报错、不崩溃,甚至日志里可能都是“正常”的。
3.1 执行确认:工具调用结果要校验,不能只看“返回了”就算成功
如果你跑过 Agent 就知道,大多数 Agent 框架里“调用工具”这个动作长这样:模型决定调用某个工具,框架把请求发出去,收到响应,然后把这个响应塞回给模型继续推理。
问题就出在“收到响应”这个环节上。很多框架默认“收到了响应 = 工具执行成功了”,但这个等式在真实世界里几乎不成立。我排查过不少“Agent 假装干活”的案例,常见的原因包括:
- 工具返回了 HTTP 200,但业务状态码是失败(比如“余额不足”“库存不够”)。
- 工具调用因为超时被中断,但实际上在服务端已经执行了——Agent 把这当成“失败了,重试一次”,结果导致重复执行。
- 工具返回的内容是错误提示,但模型没意识到这是错误,把它当成正常结果继续往下编排。
所以,我现在在自己的项目里定了一条死规矩:Agent 每次调用工具,都必须显式校验三件事——调用是否成功、返回结果是否合法、业务语义是否成立。只有这三项全部通过,才允许模型把结果带入下一步。
这个校验动作,可以用一个小的“执行确认函数”包在工具外面,也可以靠 Agent 框架的中间件来实现。刚开始你可能觉得麻烦,但真的能救大命。
3.2 人工确认与自动确认的边界:别把“点确认”变成“假确认”
现在很多 Agent 产品都加了人工确认机制:Agent 想调用一个关键工具时,弹出一个确认框,用户点“允许”才继续执行。这个机制本意是好的,但我在实际体验中发现了一个很讽刺的现象:很多人点完确认之后,Agent 反而“假装执行过”了。
例如某些 Agent 工具,用户点了“确认执行”,但后续的自动化流程根本没有真正触发,页面只显示“已确认”的状态。用户以为任务在跑,实际上任务已经卡死了。这一现象的本质是:确认动作本身变成了一种“执行完成”的假象。
所以,设计人工确认机制时一定要想清楚:确认是“允许开始执行”,不是“执行已经完成”。确认之后,仍然要有独立的执行状态跟踪。用户点的确认,只能作为“授权信号”,不能替代“执行结果回执”。
在自动确认和手动确认之间做取舍时,我的建议是按风险等级分:
- 低风险操作(读日志、查数据、本地计算):可以直接自动执行。
- 中风险操作(创建草稿、生成文件、发起可撤回的请求):默认自动执行,但保留通知。
- 高风险操作(扣款、发送消息、删除数据、合入代码):强制人工确认,并且确认后还要等执行结果回执。
这个分层规则最好写死在 Agent 的编排层,而不是交给模型自己判断。模型在“是否该确认”这件事上,判断力并不可靠。
3.3 结果记录与记忆写回:崩溃后能回答“上次做到哪了”
“假装执行过”的另一个隐蔽场景是:Agent 其实做完了某一步,但它没把“做完了”这件事持久化下来,只是放在模型上下文里。然后上下文一丢,它就忘了。下次重启,它还以为这步没做,于是又做了一遍——这在用户看来,就是“执行了但像没执行过一样”。
解决这个问题的办法,是把“执行结果”也外置。我在项目里会把每个关键步骤的结果摘要,写回到状态存储和 Agent 的记忆模块里。状态存储是给系统恢复用的,记录结构化的字段,比如step_3_result: "product_list_exported, file_path: xxx.csv";记忆模块是给模型下次运行时参考的,让它不需要重新推理一遍也能知道前因后果。
这两个东西的分工非常重要:状态存储是“机器可读的真相”,记忆是“模型可用的上下文”。如果只写记忆不写状态,那系统层面还是不知道任务进展到哪;如果只写状态不写记忆,那模型下一次推理时还得靠 prompt 硬塞历史。
4. 一个可落地的方案:任务状态表 + 幂等中间件 + 确认回调
前面讲了不少理论,这一节直接上干货。我会以一个 Python 写的简单 Agent 任务管理器为例,把“不重复执行、不假装执行过”这套逻辑完整实现一遍。你可以在自己的 Agent 框架里参考着改。
4.1 表结构设计:先把“存储”搭起来
不引入太重的外部存储,用一张 SQLite 表就能说清楚核心设计。真实生产环境你可以换成 PostgreSQL 或者 Redis + DB 的组合。
任务表agent_task:
| 字段名 | 类型 | 说明 |
|---|---|---|
| task_id | TEXT PRIMARY KEY | 任务唯一 ID |
| idempotency_key | TEXT UNIQUE | 幂等键,防止重复提交 |
| status | TEXT | 当前状态:pending / running / step_done / awaiting_confirmation / failed / completed |
| current_step | INTEGER | 当前执行到第几步 |
| step_records | TEXT | 每步执行记录的 JSON 序列化 |
| result | TEXT | 最终结果摘要 |
| created_at / updated_at | DATETIME | 时间戳 |
工具调用记录表tool_call_log:
| 字段名 | 类型 | 说明 |
|---|---|---|
| call_id | TEXT PRIMARY KEY | 单次工具调用的唯一 ID |
| task_id | TEXT | 所属任务 |
| tool_name | TEXT | 工具名称 |
| request_hash | TEXT | 请求参数的哈希,用来检测重复调用 |
| status | TEXT | pending / success / failed / timeout |
| result_summary | TEXT | 执行结果摘要 |
| created_at | DATETIME | 时间戳 |
为什么要分开两张表?因为任务表记录“任务级”状态,工具调用表记录“操作级”状态。任务状态是给 Agent 编排用的,工具调用记录是给排查和审计用的。两者缺一不可。
4.2 核心流程:注册 -> 执行 -> 确认 -> 归档
整个执行流程我拆成了四个阶段,每个阶段都有明确的“崩溃恢复点”:
第一步,注册任务。拿到一个新任务后,先检查它的幂等键在不在表里。如果已经存在,直接返回已有结果,不再重复执行;如果不存在,插入一条status=pending的记录。
这一步是防重复的“第一道门”。我遇到过有人把幂等检查放在执行中,结果前一个任务还没执行完,后一个相同任务又进来了,两个任务同时跑,产生了竞态。所以检查一定要放在最前面,并且给幂等键加唯一索引。
第二步,执行步骤。Agent 每执行一个业务步骤,都先用自己的call_id查工具调用记录表。如果已经执行过了,拿历史结果;如果没有,执行后写回结果。
这里要注意:查记录、执行、写回,这三步要保证“看起来是原子的”。最简单的方式是用数据库事务,或者用一个带锁的分布式协调器。如果没有锁,两个实例同时查到“未执行”,然后同时执行,就又会重复。
第三步,确认结果。工具调用返回后,不要直接更新任务状态。先做结果校验,校验通过才把状态推进到下一步;校验不通过,标记failed并决定是否重试。
这一步是防“假装执行过”的关键。我会在确认函数里把“工具返回状态、业务状态码、结果合法性”三件事一次性检查完。
第四步,归档收尾。任务全部执行完,把最终结果写回任务表,同时更新记忆模块。这个阶段如果崩溃,恢复策略是“检查最终结果是否已写入”——如果已写入,说明任务其实已经完成了,只是通知没发出去,补发通知即可。
4.3 代码示例:一个极简的幂等执行器
下面是一个简化版的实现,核心思路可以直接抄到你的 Agent 项目里:
import hashlib import json import sqlite3 import uuid from datetime import datetime, timezone class IdempotentAgentExecutor: def __init__(self, db_path: str = "agent_tasks.db"): self.conn = sqlite3.connect(db_path) self._init_schema() def _init_schema(self): self.conn.execute(""" CREATE TABLE IF NOT EXISTS agent_task ( task_id TEXT PRIMARY KEY, idempotency_key TEXT UNIQUE, status TEXT, current_step INTEGER DEFAULT 0, step_records TEXT DEFAULT '[]', result TEXT, updated_at TEXT ) """) self.conn.execute(""" CREATE TABLE IF NOT EXISTS tool_call_log ( call_id TEXT PRIMARY KEY, task_id TEXT, tool_name TEXT, request_hash TEXT, status TEXT, result_summary TEXT, created_at TEXT ) """) self.conn.commit() def start_or_resume_task(self, task_key: str, task_payload: dict): # 第一步:检查幂等键是否已存在 row = self.conn.execute( "SELECT result FROM agent_task WHERE idempotency_key = ?", (task_key,) ).fetchone() if row: return {"status": "already_done", "result": json.loads(row[0])} # 新任务则创建记录 task_id = str(uuid.uuid4()) self.conn.execute( "INSERT INTO agent_task (task_id, idempotency_key, status, updated_at) VALUES (?, ?, 'pending', ?)", (task_id, task_key, datetime.now(timezone.utc).isoformat()) ) self.conn.commit() return {"status": "created", "task_id": task_id} def call_tool_once(self, task_id: str, tool_name: str, request: dict, executor): # 生成请求哈希,用于识别重复调用 req_str = json.dumps(request, sort_keys=True, ensure_ascii=False) req_hash = hashlib.sha256(req_str.encode()).hexdigest() existing = self.conn.execute( "SELECT status, result_summary FROM tool_call_log WHERE task_id = ? AND tool_name = ? AND request_hash = ?", (task_id, tool_name, req_hash) ).fetchone() if existing: # 已执行过,直接返回历史结果 return {"replayed": True, "status": existing[0], "result": json.loads(existing[1])} call_id = str(uuid.uuid4()) self.conn.execute( "INSERT INTO tool_call_log (call_id, task_id, tool_name, request_hash, status, created_at) VALUES (?, ?, ?, ?, 'pending', ?)", (call_id, task_id, tool_name, req_hash, datetime.now(timezone.utc).isoformat()) ) self.conn.commit() # 真正执行工具,这一步是外部调用的边界 try: result = executor(request) status = "success" summary = json.dumps(result, ensure_ascii=False) except Exception as e: status = "failed" summary = json.dumps({"error": str(e)}, ensure_ascii=False) self.conn.execute( "UPDATE tool_call_log SET status = ?, result_summary = ? WHERE call_id = ?", (status, summary, call_id) ) self.conn.commit() return {"replayed": False, "status": status, "result": json.loads(summary) if summary else None}这套代码的核心就两个点:
- 任务注册时用唯一幂等键挡重复。同一个
task_key不会被创建第二遍。 - 工具调用时用“请求哈希”识别重复。即使任务重启了,只要传入相同的
task_id + tool_name + request,就能命中历史记录,直接重放结果,而不是再执行一遍。
这是一个极简版本,生产环境的二哥(复杂度二哥)还可以加上分布式锁、重试退避策略、结果校验回调。但骨架就是这个意思。
5. 真实场景里的坑与排查实录
最后这一趴,我想分享几个我们在实际项目中真正遇到过的坑。有些问题光看理论根本想不到,只有跑过才知道有多痛。
5.1 常见问题速查表
| 现象 | 根本原因 | 处理方式 |
|---|---|---|
| 任务重跑后,外部系统出现重复扣款/工单 | 外部接口没有幂等设计,或 Agent 侧没有传幂等键 | 给外部调用统一加上幂等键;下游接口改造为 upsert 语义 |
| 工具返回 HTTP 200 但业务报错,Agent 照样继续跑 | 只校验了传输层,没校验业务层 | 增加“结果校验回调”,明确业务状态码必须为成功才允许继续 |
| Agent 超时后重试,结果外部系统其实已经执行成功了 | 超时并不代表未执行 | 超时先查状态,确认未执行再重试;能查询结果的接口先查询 |
| Agent 卡在“等待确认”,用户以为任务已运行 | 确认步骤只更新了 UI,没有触发实际执行 | 把“确认授权”和“执行任务”拆成两个独立状态,分别跟踪 |
| 两个 Agent 实例同时处理同一任务 | 没有分布式锁,幂等键检查竞态 | 加“先占锁再检查再执行”的流程,或对幂等键做唯一索引 + 冲突捕获 |
这五类问题几乎覆盖了 Agent 崩溃场景里 80% 的重复执行和执行不确定问题。我建议你把这张表贴在项目文档里,每次设计新 Agent 任务时逐条过一遍。
5.2 排查工具与日志设计:崩溃后靠什么还原现场?
崩溃后的排障,最怕的是日志里只有“Error: xxx”。Agent 的执行链路特别长,如果日志里没把“任务 ID、步骤号、调用 ID、请求哈希”这些关键字段串起来,排查工作基本等于考古。
我们项目的日志规范是:每一个 Agent 任务的每条日志,都带上task_id和call_id。不管你是用日志框架还是直接print,这两个字段一定要贯穿始终。具体格式类似:
ts=2024-12-16T10:23:45Z level=INFO task_id=xxx call_id=yyy event=call_start tool=create_order request_hash=abc123 ts=2024-12-16T10:23:47Z level=INFO task_id=xxx call_id=yyy event=call_success tool=create_order order_id=9527有了这种日志,崩溃后你能迅速回答三个问题:这个任务执行到第几步了?那一步的请求参数是什么?外部系统有没有返回成功?
另一个排查利器是“状态快照”。我会在任务表里把step_records维护成 JSON 数组,每完成一步就 append 一条记录,包含步骤名、工具调用 ID、结果摘要。这样即使格式化日志丢失了,只要数据库还在,就能还原出任务的完整执行轨迹。
5.3 避坑建议:从经验中总结的 5 条铁律
第一条铁律:永远不要在内存里保存“唯一的真相”。任何关于任务进展的信息,都必须落盘。模型上下文、局部变量、会话状态都不算数。
第二条铁律:不要在 Agent 崩溃后直接重试“整个任务”。先查状态、分析卡点,再决定是重放、跳过还是修复。盲目的整任务重试,是重复执行的温床。
第三条铁律:能查到的就先查,不要一上来就执行。很多接口设计得更好一些,让它提供了“查询是否已执行”的能力。例如支付订单可以先查订单状态,文件上传可以先查文件是否存在。先查后做,能避免一大半重复操作。
第四条铁律:模型的话不可全信,校验得靠代码。不要指望 Agent 模型自己判断“这个调用成功了”,这种判断在简单场景下还行,一旦涉及到外部系统的复杂状态,可靠性就会急剧下降。写一个确定性的校验函数,把“成功标准”固化成代码。
第五条铁律:给整个系统留一条“人工干预的后门”。无论 Agent 设计得多么完善,都要留一个操作员能手动修改任务状态、强制标记某一步为已完成或失败的入口。生产环境里,这个后门能救命。
最后,说点我自己的体会
这几个月做了好几个 Agent 项目以后,我发现一个有意思的现象:大家最开始都在拼谁的 Agent 更“聪明”,能处理更复杂的任务。但做了一段时间后,所有人的焦点几乎都会转到同一个问题上——怎么让一个不太聪明的 Agent 稳定地不出错。
幂等和执行确认,不是为了限制 Agent 的能力,而是为了让它的能力真正可以被信任。一个能写代码但无法确认自己是否提交成功的 Agent,和一个能力一般但每一步都可追溯、可重试、可回滚的 Agent,放在真实的业务里,后者价值高出一个数量级。
如果你正在做 Agent 项目,我强烈建议你把“崩溃恢复”当作一等公民来设计,而不是后期补丁。状态表、幂等键、执行确认这三个东西,从第一天就放进架构里,后边能少掉无数头发。如果之后再有人问你“Agent 崩溃了怎么办”,你可以反问他一句:“你敢让它不带幂等键地重跑一遍吗?”