1. 为什么AI Agent的“出错处理”比模型能力更值得关注
很多人第一次搭AI Agent,注意力全在模型选型上——用哪个大模型、上下文窗口多大、推理能力多强。但真正把Agent放到生产环境跑上一周,你会发现一个扎心的事实:决定Agent能不能用的,往往不是它有多聪明,而是它出错之后你怎么收拾。
我自己的经历很典型。早期做的一个Agent负责自动整理工单、调用内部接口改状态、再回写数据库。测试环境跑得挺顺,上线第三天就出事了:模型把一条“待审核”的工单误判成“已关闭”,直接调了写接口。等发现的时候,几十条记录已经被改。没有校验、没有暂停、没有回滚,只能人工一条条对着日志往回捞。那次之后我才真正理解,Agent和普通程序最大的区别在于——它的执行路径是概率性的,不是确定性的。普通代码你写if a then b,它就一定走b;Agent是“大概率走b,偶尔抽风走c,极端情况下走一个你根本没想过的d”。
所以这篇东西我想聊的不是“怎么让Agent更聪明”,而是怎么让它在犯错时可控。核心就四件事:校验(Validation)、暂停(Pause)、回滚(Rollback)、人工接管(Human Takeover)。这四个词听起来像运维术语,但放到Agent语境里,含义和实现方式都不太一样。适合谁看?如果你正在从0到1搭Agent,或者已经把Agent跑起来了但总担心它闯祸,那这篇基本就是给你写的。我会把每个环节的原理、为什么这么设计、具体怎么落地、踩过哪些坑都摊开讲,代码和配置尽量给到能直接抄的程度。
先说一个贯穿全文的判断:Agent的可靠性不来自模型,来自外围的约束层。模型是发动机,校验、暂停、回滚、接管是刹车、安全带、气囊和备用方向盘。发动机再强,没有这四样,你不敢让它上高速。
2. 校验层:在Agent动手之前拦住它
2.1 校验到底校验什么:三层校验模型
很多人一提“校验”就想到表单校验规则那种东西——检查输入是不是邮箱、是不是数字。但Agent场景下的校验要复杂得多,因为它校验的不只是数据格式,还有意图、权限和后果。我一般把它拆成三层:
- 输入校验:Agent拿到的任务描述、参数、上下文是否合法、完整、无歧义。
- 决策校验:Agent打算执行的动作,是否符合预设的规则和边界。
- 输出校验:Agent执行完的结果,是否在可接受范围内,是否和预期一致。
这三层的顺序不能乱。输入校验是入口关,决策校验是核心关,输出校验是兜底关。任何一层没过,都应该触发后续的暂停或接管流程,而不是硬着头皮往下走。
举个具体例子。假设Agent的任务是“把库存低于10的商品下架”。输入校验要确认:商品ID存在吗?库存数据来源可信吗?决策校验要确认:这个Agent有没有下架权限?下架这个动作是不是高危操作、需不需要二次确认?输出校验要确认:下架后商品状态真的变了吗?有没有误伤其他商品?
2.2 规则校验的落地:从硬编码到可配置
最朴素的校验就是硬编码if判断,但Agent的动作空间一大,硬编码会爆炸。我的做法是把校验规则抽成独立的规则表,用配置驱动。规则表大概长这样:
| 规则ID | 适用动作 | 校验类型 | 条件 | 失败处理 |
|---|---|---|---|---|
| R001 | 修改订单状态 | 决策校验 | 目标状态 ∈ {待审核,已关闭} | 暂停并转人工 |
| R002 | 调用支付接口 | 决策校验 | 金额 ≤ 5000 | 拒绝执行 |
| R003 | 删除数据 | 决策校验 | 永远禁止 | 直接拦截 |
| R004 | 任意写操作 | 输出校验 | 影响行数 ≤ 100 | 回滚并告警 |
这样设计的好处是,规则可以热更新,不用改代码重新部署。新增一条“禁止在凌晨2点到4点执行批量操作”的规则,运维同学自己就能加。
规则引擎的选型上,轻量场景我推荐直接用JSON/YAML配置加一个简单的表达式求值器(比如用Python的asteval或者JS的expr-eval),没必要上Drools那种重型规则引擎。Agent的校验规则通常不复杂,重引擎反而增加维护成本。
2.3 数据完整性校验:别忽略那些“看起来没问题”的输入
Agent经常要处理外部数据——从数据库读、从接口拉、从文件解析。这些数据在格式上可能完全合法,但内容上是错的。比如一个金额字段是-100,格式没问题,但业务上不可能。一个时间戳是0,能解析,但那是1970年。
这类问题靠格式校验抓不住,得靠业务语义校验。我的经验是维护一份“字段语义约束表”,对关键字段定义合理范围:
FIELD_CONSTRAINTS = { "amount": {"min": 0, "max": 1_000_000, "type": "decimal"}, "timestamp": {"min": 1_600_000_000, "max": None}, # 不早于2020年 "status": {"enum": ["待审核", "处理中", "已关闭", "已取消"]}, "user_id": {"pattern": r"^U\d{8}$"}, }Agent在读取数据后、做决策前,先过一遍这张表。任何字段越界,直接标记为“数据可疑”,暂停流程。这一步能拦掉相当一部分因为脏数据导致的误操作。
注意:校验规则本身也会出错。我见过有人把
max写成1_000_000结果单位是“分”而实际数据是“元”,导致所有正常订单都被拦。规则上线前一定要用真实数据跑一遍回归,别拍脑袋定阈值。
2.4 校验失败的降级策略:不是所有失败都要停
校验失败不等于世界末日。根据严重程度,我一般分三档处理:
- 硬失败:违反安全红线(如删除数据、越权操作),直接拦截,记录审计日志,通知负责人。
- 软失败:数据可疑但动作可逆(如金额略超阈值),暂停并请求人工确认,确认后可继续。
- 警告:轻微异常(如字段格式不规范但可自动修正),记录警告,自动修正后继续。
这个分级很重要。如果所有校验失败都触发人工,人会累死,Agent也会因为频繁暂停而失去自动化价值。关键是把“必须人管”和“可以自动兜”分开。
3. 暂停机制:让Agent在关键时刻“踩刹车”
3.1 暂停的三种触发方式
Agent的暂停不是简单地把进程kill掉,那样状态全丢。真正的暂停要保留上下文,支持后续恢复。触发方式主要有三种:
- 规则触发:校验层判定需要暂停,主动调用暂停接口。
- 人工触发:监控人员发现异常,手动按下暂停按钮。
- 系统触发:外部依赖不可用(如数据库连接断开)、资源超限(如Token消耗超预算),自动暂停。
这三种触发最终都汇聚到同一个暂停管理器。暂停管理器负责:冻结当前执行状态、保存上下文快照、通知相关方、等待恢复指令。
3.2 状态快照:暂停的核心是“可恢复”
暂停如果不可恢复,那和终止没区别。所以暂停时必须做状态快照。快照要包含什么?我的清单是:
- 当前任务ID和任务描述
- 已执行的步骤序列(每一步的输入、输出、时间戳)
- 当前待执行的步骤及其参数
- 模型的对话历史(如果Agent是基于对话的)
- 外部调用的中间状态(如已发起的请求、已获取的锁)
快照存哪里?小规模用Redis就行,HSET存结构化字段,RPUSH存步骤序列。大规模或者需要长期保留的,落库到PostgreSQL,用JSONB字段存上下文。关键是快照要原子写入,不能写一半崩了。
def pause_agent(task_id, reason): snapshot = { "task_id": task_id, "reason": reason, "timestamp": time.time(), "steps": get_executed_steps(task_id), "pending": get_pending_step(task_id), "context": get_agent_context(task_id), } # 原子写入,先写临时key再rename redis_client.set(f"snapshot:{task_id}:tmp", json.dumps(snapshot)) redis_client.rename(f"snapshot:{task_id}:tmp", f"snapshot:{task_id}") redis_client.set(f"status:{task_id}", "PAUSED") notify_operator(task_id, reason)3.3 暂停的粒度:步骤级还是动作级
暂停粒度是个容易纠结的点。步骤级暂停是“执行完当前步骤后停”,动作级暂停是“当前动作执行到一半也能停”。后者实现难度大得多,因为很多外部调用不是原子的。
我的建议是:默认用步骤级暂停,只对高危动作做动作级暂停。比如“调用支付接口”这种,要么完整成功要么完整失败,中间状态没法暂停。而“批量更新100条记录”可以做成每10条检查一次暂停信号,支持中途停。
实现上,步骤级暂停就是在每个步骤执行前检查一个pause_flag:
def execute_step(step): if is_paused(step.task_id): save_snapshot(step.task_id) return StepResult(status="PAUSED") # 执行步骤...动作级暂停则需要在动作内部埋检查点,复杂度高,只在必要场景用。
3.4 暂停后的通知与等待:别让人干等
暂停之后,得让人知道。通知渠道看团队习惯,邮件、IM、工单系统都行。通知内容要包含:哪个任务停了、为什么停、当前状态、需要人做什么决策。
等待恢复的机制有两种:轮询和回调。轮询简单,Agent暂停后定期查status字段,变成RESUMED就继续。回调更实时,人工在管理后台点“恢复”后,后台直接调用Agent的恢复接口。我倾向回调,因为轮询有延迟,而且Agent进程一直挂着也占资源。
实操心得:暂停状态一定要设超时。我遇到过人工忘了处理,任务暂停了三天,恢复的时候外部数据早就变了,快照里的上下文全失效。现在我的做法是暂停超过2小时自动转“待人工接管”,超过24小时自动标记为“废弃”,释放资源。
4. 回滚:把Agent闯的祸收回来
4.1 回滚的前提:可逆性设计
回滚能不能做,取决于动作本身可不可逆。发出去的邮件收不回来,调用第三方接口改了对方数据也未必能改回来。所以回滚能力是在设计Agent动作时就要考虑的,不是出事之后才想。
我的原则是:能设计成可逆的,绝不设计成不可逆。具体做法:
- 写操作优先用“软删除”或“状态标记”,而不是物理删除。
- 批量操作前先备份原数据,或者记录反向操作。
- 外部调用尽量选幂等的接口,或者自己维护一个“补偿操作”映射。
比如“下架商品”这个动作,不要直接DELETE,而是UPDATE status = 'offline'。回滚就是UPDATE status = 'online'。简单、可靠、可审计。
4.2 回滚的三种实现模式
根据场景不同,回滚有三种模式:
模式一:反向操作(Compensating Action)每个正向操作配一个反向操作。正向是“扣库存”,反向是“加库存”。正向是“发通知”,反向是“发更正通知”。这种模式适合操作序列明确、每步都有对应逆操作的场景。
模式二:快照恢复(Snapshot Restore)操作前对受影响的数据做完整快照,回滚时用快照覆盖。适合批量修改、数据结构复杂的场景。缺点是快照占空间,大表快照成本高。
模式三:事务回滚(Transaction Rollback)如果所有操作都在同一个数据库事务里,直接ROLLBACK就行。但Agent的操作往往跨系统、跨接口,很难包在一个事务里。所以这种模式只适合单库单事务的简单场景。
实际项目中,我通常是模式一为主,模式二为辅。关键数据操作前做快照,日常操作靠反向操作。
4.3 回滚的触发与执行流程
回滚不是随便就能触发的。我的流程是:
- 检测异常:输出校验失败、人工发现错误、监控告警。
- 评估影响范围:哪些数据被改了?改了多久?有没有下游依赖?
- 决定回滚策略:全量回滚还是部分回滚?用反向操作还是快照恢复?
- 执行回滚:按操作序列的逆序执行,每步校验。
- 验证回滚结果:确认数据恢复到预期状态。
- 记录审计:回滚的原因、范围、结果全部留痕。
第4步的“逆序”很关键。Agent的操作是有依赖的,先创建订单再扣库存,回滚就得先恢复库存再删订单。顺序错了会出问题。
def rollback(task_id): steps = get_executed_steps(task_id) for step in reversed(steps): if step.reversible: inverse = get_inverse_action(step) result = execute_inverse(inverse) if not result.success: # 回滚失败,升级为人工接管 escalate_to_human(task_id, f"回滚步骤{step.id}失败") return mark_task_rolled_back(task_id)4.4 回滚不干净怎么办:那些“回滚了但没完全回滚”的坑
回滚最怕的是“回滚了但没回干净”。我踩过的坑包括:
- 部分回滚:批量操作回滚到一半失败,数据处于半新半旧状态。
- 级联遗漏:改了主表忘了改关联表,或者改了数据库忘了清缓存。
- 外部系统不同步:内部数据回滚了,但已经推送给外部系统的数据没撤回。
- 副作用残留:操作触发了通知、日志、计费等副作用,回滚时没处理。
应对这些,我的经验是:
- 回滚操作本身也要有校验和重试,不能假设一次成功。
- 维护一份“副作用清单”,回滚时逐项检查。
- 对于无法回滚的外部副作用,生成“更正说明”而不是假装没发生。
- 回滚后跑一遍数据一致性检查,别信“应该没问题”。
注意:回滚不是万能的。有些操作就是不可逆的,比如已经发给客户的邮件、已经执行的支付。这种情况下,回滚的目标不是“当没发生过”,而是“把状态修正到可接受”。这个心态要摆正,不然会在不可逆操作上浪费大量精力。
5. 人工接管:最后的防线怎么设计
5.1 什么情况下必须人工接管
不是所有暂停都需要人工接管。我划的线是:
- 涉及不可逆操作:删除、支付、对外发送,必须人工确认。
- 涉及高价值数据:金额超过阈值、影响核心业务,必须人工确认。
- Agent置信度低:模型自己都拿不准的决策,转人工。
- 连续失败:同一任务重试超过N次仍失败,转人工。
- 规则明确要求:某些业务场景合规要求必须人工复核。
反过来,低风险、可逆、Agent置信度高的操作,暂停后可以自动恢复,不用惊动人。
5.2 接管界面的核心要素
人工接管不是让人去看日志猜发生了什么。接管界面要让人在最短时间内理解现状并做出决策。核心要素:
- 任务概览:任务是什么、谁发起的、跑了多久。
- 执行轨迹:每一步做了什么、结果如何、耗时多少。
- 暂停原因:为什么停、哪条规则触发的。
- 待决策项:需要人做什么选择,选项有哪些。
- 影响预览:每个选项会导致什么后果。
- 操作按钮:继续、回滚、修改参数后继续、终止。
我见过太多接管界面只给一个“发生了什么”的日志,然后让人自己判断。这是偷懒。好的接管界面应该把决策所需的信息都摆好,人只需要点按钮。
5.3 接管后的状态同步:别让Agent“失忆”
人工接管后,Agent恢复执行时,必须知道人工做了什么。比如人工修改了某个参数,Agent得用新参数继续;人工回滚了某步,Agent得从回滚后的状态重新规划。
实现上,接管操作要写回Agent的上下文。我的做法是维护一个“人工干预记录”,Agent恢复时先读这个记录,把人工的修改合并进当前状态。
def resume_after_human(task_id, human_action): intervention = { "task_id": task_id, "action": human_action.type, # continue / rollback / modify / abort "modifications": human_action.modifications, "timestamp": time.time(), } save_intervention(intervention) context = load_context(task_id) context = apply_intervention(context, intervention) save_context(task_id, context) set_status(task_id, "RUNNING")5.4 人机协作的边界:哪些交给Agent,哪些留给人
长期来看,人工接管的比例应该越来越低,但不是降到零。我的经验是,把“判断”留给人,把“执行”交给Agent。人负责决定“要不要做”“做到什么程度”,Agent负责“怎么做”“做多少”。
具体分工上:
| 决策类型 | 交给Agent | 留给人 |
|---|---|---|
| 常规数据查询 | 是 | 否 |
| 低风险状态更新 | 是 | 否 |
| 批量数据修改 | 校验后执行 | 超阈值时确认 |
| 对外发送 | 否 | 是 |
| 资金相关 | 否 | 是 |
| 删除操作 | 否 | 是 |
| 异常处理 | 尝试自动恢复 | 恢复失败时接管 |
这个边界不是一成不变的。随着Agent可靠性提升和校验规则完善,可以把更多决策下放给Agent。但下放的前提是有对应的校验和回滚能力兜底。
6. 把四件事串起来:一个完整的容错流程
6.1 从任务开始到结束的完整链路
把校验、暂停、回滚、接管串成一条线,一个任务的生命周期大概是这样:
- 任务进入:输入校验,确认任务合法、参数完整。
- 规划阶段:Agent生成执行计划,决策校验逐条检查计划中的动作。
- 执行阶段:每步执行前检查暂停信号,执行后做输出校验。
- 异常分支:
- 校验失败 → 按严重程度决定拦截/暂停/警告。
- 暂停 → 保存快照,通知人工。
- 人工决策 → 继续/回滚/修改/终止。
- 回滚 → 逆序执行反向操作,验证结果。
- 任务结束:记录完整审计日志,更新任务状态。
这条链路里,每个环节都要有日志。Agent的审计日志不是“出了事才看”,而是“随时能看”。我习惯把日志结构化,每条包含:时间、任务ID、步骤ID、动作类型、输入摘要、输出摘要、校验结果、耗时。
6.2 一个可复用的容错框架设计
如果每个Agent都从头实现这四件事,重复劳动太多。我一般会抽一个容错框架,Agent开发者只需要定义“动作”和“校验规则”,框架负责暂停、回滚、接管。
框架的核心接口:
class FaultTolerantAgent: def register_action(self, name, forward, inverse=None, reversible=True): """注册动作及其反向操作""" def register_rule(self, action_name, rule): """注册校验规则""" def execute(self, task): """执行任务,内置校验、暂停、回滚""" def pause(self, task_id, reason): """暂停任务""" def resume(self, task_id, intervention=None): """恢复任务""" def rollback(self, task_id): """回滚任务"""这样Agent开发者写业务逻辑,框架管可靠性。新Agent接入成本低,容错能力也统一。
6.3 监控与告警:让问题在爆发前被发现
容错机制再完善,也是事后处理。更好的做法是提前发现苗头。我关注的监控指标:
- 校验失败率:突然升高说明输入数据或规则有问题。
- 暂停次数:频繁暂停说明Agent决策质量下降。
- 回滚率:回滚率高说明校验不够前置。
- 人工接管率:接管率是Agent自主性的反向指标。
- 平均恢复时间:从暂停到恢复的耗时,反映人工响应效率。
这些指标设阈值告警。比如校验失败率超过5%就告警,暂停次数一小时超过10次就告警。告警不是目的,目的是让人在问题变大之前介入。
6.4 常见问题速查表
| 问题现象 | 可能原因 | 排查方向 | 处理建议 |
|---|---|---|---|
| Agent执行了不该执行的动作 | 决策校验缺失或规则未覆盖 | 检查规则表是否覆盖该动作 | 补充规则,回滚已执行动作 |
| 暂停后无法恢复 | 快照不完整或上下文丢失 | 检查快照字段是否齐全 | 完善快照,必要时重跑任务 |
| 回滚后数据不一致 | 反向操作遗漏或顺序错误 | 对比回滚前后数据快照 | 补做遗漏的反向操作 |
| 人工接管后Agent行为异常 | 干预记录未正确合并 | 检查上下文合并逻辑 | 修正合并逻辑,重跑 |
| 频繁暂停影响效率 | 校验规则过严 | 分析暂停原因分布 | 放宽低风险规则,改为警告 |
| 回滚耗时过长 | 批量操作数据量大 | 评估回滚性能瓶颈 | 分批回滚,或改用快照恢复 |
实操心得:这套机制刚上线时,暂停和接管会特别频繁,别慌。这是正常的,说明校验在起作用。随着规则调优和Agent决策质量提升,暂停率会降下来。我第一个Agent上线第一周接管率30%,一个月后降到5%以下。关键是每次接管后都要复盘,看是规则问题还是Agent问题,针对性优化。
7. 一些不那么显然的经验
7.1 校验规则要“可解释”
校验失败时,不能只告诉人“失败了”,要告诉人“为什么失败”。规则引擎返回的结果要包含:触发了哪条规则、规则的条件是什么、实际值是什么、期望值是什么。这样人工接管时能快速判断是数据问题还是规则问题。
我见过有人把校验结果写成{"passed": false},然后人工一脸懵。改成{"passed": false, "rule": "R002", "condition": "amount <= 5000", "actual": 8000, "expected": "<= 5000"},排查效率天差地别。
7.2 回滚要考虑“回滚的回滚”
回滚操作本身也可能失败,或者回滚之后发现回滚错了。所以回滚也要有审计,也要能“再回滚”。虽然这种情况很少,但设计时留个口子,比事后抓瞎强。
7.3 人工接管不是越多越好
接管率高不代表安全,可能代表Agent太笨或者规则太严。目标是让Agent处理绝大多数常规情况,人只处理真正的例外。如果接管率长期高于10%,得回头看看是Agent能力问题还是流程设计问题。
7.4 容错机制本身要轻量
我见过有人为了做容错,引入了一堆中间件、消息队列、分布式事务框架,结果系统复杂度爆炸,容错机制自己成了故障源。我的建议是:能用简单方案就别上重型框架。Redis存快照、数据库存审计、一个管理后台做接管,对大多数Agent场景足够了。等规模真的上来了再考虑升级。
7.5 测试容错机制比测试Agent更难
测试Agent的正常路径容易,构造异常路径难。我的做法是故意注入故障:在测试环境随机让校验失败、随机暂停、随机让回滚失败,看整个流程能不能正确处理。这种“混沌测试”能发现很多正常测试发现不了的问题。
最后分享一个我自己的习惯:每次Agent上线新功能,我都会先问三个问题——这个动作可逆吗?校验覆盖了吗?出错了人怎么接管?三个问题有一个答不上来,就不上线。这个习惯帮我省了无数次半夜爬起来救火。Agent的能力可以慢慢加,但容错的底线一开始就得守住。