一个AI任务可能包含资料准备、内容提取、模型生成、人工审核、结果保存和通知等多个步骤。最简单的异常处理方式是：任何一步失败，就从头重新运行整条流程。
这种做法在演示阶段容易实现，进入长期运行后却会产生新问题。已经完成的模型调用可能再次计费，已经写入的结果可能重复保存，已经发送的通知可能再次触发；更危险的是，业务校验失败也被当成网络错误反复重试，系统只是在更快地重复同一个错误。
更稳妥的设计是把失败分成三类：瞬时错误允许有限重试，业务错误直接进入人工处理，已产生副作用的步骤使用补偿动作回到可控状态。云工作流负责步骤编排、Retry和Catch，函数计算负责返回清晰错误类型，应用自己的补偿台账负责保证撤销动作只执行一次。
一、先回答：什么时候可以局部恢复
局部恢复成立需要三个前提：
- 每个步骤有清晰输入和输出；
- 已完成步骤有可查询的完成记录；
- 具有副作用的步骤支持幂等或补偿。
如果工作流只是一个大函数，内部连续完成下载、生成、保存和通知，失败时只能知道“函数报错”，很难确认哪一步已经产生影响。把流程拆成状态，不是为了增加节点数量，而是为了让恢复边界可见。
阿里云云工作流以状态机描述流程，可通过Task状态集成函数计算。官方文档说明，云工作流可以调用函数计算函数，并将函数资源、调用方式和输入参数写入流程定义。云工作流集成函数计算
二、不要对所有错误使用同一重试规则
可以先建立三类错误：
1. 瞬时错误
例如临时网络失败、服务短暂限流或偶发内部错误。这类问题在等待后可能自行恢复，适合有限次数、指数退避的重试。
2. 业务错误
例如输入缺少必要字段、文档没有处理权限、审核不通过或模型输出违反业务规则。重复执行不会让输入自动变正确，应直接停止并进入人工检查。
3. 不确定错误
例如调用超时后，不清楚下游是否已经成功。此时不能直接重试副作用步骤，应先根据幂等键查询结果，再决定继续、补偿或人工介入。
云工作流的错误处理包含Retry和Catch；对于支持错误处理的Task、Parallel和Map状态，会先执行Retry，重试仍失败后再执行Catch。Retry可以配置最大次数、间隔、退避倍率和最大退避时间。云工作流错误处理
产品提供的是错误匹配和流程跳转能力，业务仍需定义哪些错误可重试。不能把FnF.ALL统一配置成多次重试后就认为系统具备正确恢复能力。
三、一个本地可运行的恢复核心
下面的Python示例模拟三种状态：步骤完成、瞬时错误重试、业务错误转人工。
from dataclasses import dataclass, field
class TransientError(Exception):
pass
class BusinessError(Exception):
pass
@dataclass
class Ledger:
completed: set[str] = field(default_factory=set)
compensated: set[str] = field(default_factory=set)
Ledger是最小补偿台账，分别记录完成步骤和已经补偿的步骤。生产环境应将其保存到持久化存储，并使用任务ID、步骤名和执行版本组成唯一键。
步骤执行器如下：
def run_step(name, action, ledger, retries=2):
if name in ledger.completed:
return "already_completed"
for attempt in range(retries + 1):
try:
action()
ledger.completed.add(name)
return f"completed_at_{attempt + 1}"
except TransientError:
if attempt == retries:
return "retry_exhausted"
except BusinessError:
return "manual_review"
这里有两个关键点。
第一，步骤开始前检查完成台账，避免流程恢复后再次执行已经完成的动作。
第二，只捕获明确的TransientError进行重试。BusinessError不进入循环，直接返回人工处理状态。
四、补偿动作为什么也必须幂等
补偿不是数据库事务的自动回滚，而是业务定义的反向动作。例如：
- 已创建草稿，补偿动作可以把草稿标记为废弃；
- 已预留额度，补偿动作可以释放额度；
- 已写入临时对象，补偿动作可以加入清理队列；
- 已发出不可撤回通知，则不能假装补偿成功，只能记录影响并人工处理。
一个最小补偿函数：
def compensate(name, undo, ledger):
if name not in ledger.completed:
return "skip"
if name in ledger.compensated:
return "skip"
undo()
ledger.compensated.add(name)
return "compensated"
补偿动作本身也可能因为网络问题被重复调用，因此必须检查compensated状态。不能只让正向步骤幂等，却让回滚重复执行。
五、用一次瞬时失败和一次业务失败验证
测试场景如下：
ledger = Ledger()
calls = {
"prepare": 0, "notify": 0}
def prepare():
calls["prepare"] += 1
if calls["prepare"] == 1:
raise TransientError()
def notify():
calls["notify"] += 1
raise BusinessError()
print("prepare", run_step("prepare", prepare, ledger))
print("notify", run_step("notify", notify, ledger))
print("rollback", compensate(
"prepare", lambda: None, ledger
))
print("rollback_again", compensate(
"prepare", lambda: None, ledger
))
实际运行输出：
prepare completed_at_2
notify manual_review
rollback compensated
rollback_again skip
结果说明：
prepare第一次遇到瞬时错误，第二次完成；notify属于业务错误，只调用一次并进入人工检查；prepare的补偿第一次成功；- 同一补偿第二次执行被幂等跳过。
这组输出只验证本地分类与台账逻辑，不代表已经在云工作流中部署。
六、怎样映射到云工作流
可以将AI内容任务拆成以下状态：
ValidateInput
↓
PrepareMaterial
↓
GenerateDraft
↓
HumanReview
↓
SaveResult
↓
Notify
函数对不同问题返回或抛出明确错误类型：
TemporaryUpstreamError
InvalidInput
ReviewRejected
ResultConflict
流程策略可以设计为：
TemporaryUpstreamError：最多重试两次，间隔逐步增加；InvalidInput：Catch后进入NeedInput；ReviewRejected：进入Rejected终态；ResultConflict：进入CheckExistingResult；- 重试耗尽：进入
ManualReview并保存错误上下文。
云工作流文档说明，Retry失败后可由Catch捕获并跳转到指定状态，还可构造输出交给下一个状态处理。Retry与Catch错误处理
对于函数计算异步任务，阿里云也提供结合云工作流进行顺序、分支和并行编排的方案，并跟踪任务状态转换及执行自定义重试逻辑。函数计算异步任务编排
七、流程参数不要携带越来越大的中间结果
工作流步骤之间适合传递小型控制信息：
{
"task_id": "t-001",
"input_ref": "oss://.../input.json",
"result_ref": "oss://.../draft.json",
"processor_version": "2026-07-30",
"attempt": 2
}
较大的文档、模型输出和审核附件应保存到OSS等外部存储，流程只传引用。云工作流官方文档指出，步骤输入、输出和本地变量存在总大小限制，因此更不应把大段模型文本在每一步之间复制。云工作流输入和输出
这里不在正文中固化具体限制数值作为永久事实；配置和限制可能变化，发布前应重新查看官方文档。
八、补偿顺序应与正向步骤相反
若流程已经完成：
创建临时资源 → 写入草稿 → 预留额度
后续失败时，补偿通常按相反顺序执行：
释放额度 → 标记草稿废弃 → 清理临时资源
原因是后一步往往依赖前一步创建的资源。如果先删除底层资源，后续补偿可能失去必要信息。
但不是所有动作都可以补偿。模型调用已经发生，通常无法“撤回计算”；外部通知已经到达，也无法保证收件人没有看到。对此应记录不可逆副作用，而不是把补偿状态伪装成完全回到原点。
九、人工审核不是一种异常
AI工作流经常把人工审核设计成“函数失败后让人处理”，这会让审核状态与系统故障混在一起。
更好的做法是将人工审核作为正常状态：
pending_review
approved
rejected
expired
审核超时、拒绝和要求补充资料都有明确流转。它们不是服务异常，不应套用网络重试。
对于等待外部回调或人工确认的任务，云工作流提供相应的异步集成模式；使用时需要按照当前官方文档配置权限、回调和超时。云工作流集成函数计算异步调用
十、恢复执行前必须检查版本
长时间暂停的流程恢复时，函数代码、提示词和业务规则可能已经更新。
不能默认用“最新版本”继续旧任务，否则同一条流程前后可能执行不同规则。任务开始时应记录：
workflow_version
function_version
prompt_version
input_version
恢复时要么继续使用原版本，要么显式创建迁移任务，并记录为什么切换。阿里云官方也提供云工作流调用函数固定版本或别名的实践，用于控制流程执行中的版本一致性和回滚。云工作流调度固定版本函数
十一、可观测性应该围绕状态变化
仅记录“函数调用成功”不足以说明业务完成。建议每次状态变化记录：
task_id
workflow_version
step_name
from_status
to_status
attempt
error_type
side_effect_id
timestamp
需要关注的指标包括：
- 各步骤首次成功率；
- 瞬时错误重试后成功数量；
- 业务错误进入人工队列数量；
- 重试耗尽数量；
- 补偿成功、失败和重复跳过数量；
- 长时间停留在同一状态的任务数量。
不要只追求整体成功率。若大量任务经过多次重试才完成，可能说明上游服务、超时参数或并发控制存在问题。
十二、适合OPC一人公司的最小方案
一个人运营时，可以先从四个步骤开始：
- 输入检查；
- 生成草稿；
- 人工审核；
- 保存结果。
为每一步定义完成条件和错误类型。只有明确的瞬时错误可以重试；输入问题和审核拒绝直接停下；保存结果使用任务ID作为幂等键；已经产生的临时对象加入补偿台账。
在“智能体来了”的内容实践中，AI大模型工具深度运用不等于追求完全无人化，而是让自动步骤、人工判断、失败恢复和版本记录各有边界。OPC中国只表示中国语境下的一人公司议题，不表示官方机构或标准。
结语
AI工作流中间失败后，不应该默认整条重跑，也不能默认所有错误都适合重试。
可靠恢复需要四层设计：函数返回可分类错误，云工作流用Retry和Catch编排流转，业务台账记录已完成与已补偿步骤，版本信息保证恢复前后规则一致。
本文原型验证了瞬时错误有限重试、业务错误转人工以及补偿幂等；映射到阿里云时，仍需结合具体函数版本、流程定义、权限、超时、外部存储和不可逆副作用进行验证。
局部恢复的目标不是让失败消失，而是让系统明确知道失败发生在哪里、哪些动作已经完成、下一步应自动重试还是交给人处理。
说明：本文使用AI工具辅助进行结构整理和语言优化，架构判断、示例代码及正文内容已由发布者人工审核。文中本地输出只验证示例逻辑，不代表云端部署结果、性能、费用或平台审核结果。
