Agent 工作流编排实战：多步骤任务的状态机设计与失败恢复避坑

第一个上生产的 Agent 任务是「竞品日报」：拉取竞品动态、汇总分析、生成报告、发送邮件，一共八个步骤。演示时一气呵成，上生产后第一周就暴露了两个致命问题——一次超时导致整条链路从第一步重跑，前面六步的成果全部作废；另一次是「生成报告」和「质量检查」两个步骤互相打回，Agent 在里面死循环了四十分钟，token 账单看得人心疼。问题的本质是：我们用「自由对话循环」的模式去跑「流程明确」的任务。解法是把编排方式升级为显式状态机加检查点持久化，这篇文章把完整设计与实现记录下来。

方案对比：三种编排模式的适用边界

模式	特点	适用场景
自由 Agent 循环（ReAct）	模型自主决定下一步，灵活	开放式探索，步骤不可预知
预定义工作流（状态机）	步骤与转移显式声明，可控可恢复	步骤明确的业务流程，可上生产
混合模式	状态机做骨架，单步内部允许 Agent 自由发挥	多数真实业务系统的最终形态

判断标准很简单：如果任务的步骤你能画在白纸上，就用状态机；画不出来，才考虑自由循环。而生产系统几乎总是混合模式——骨架负责可控与恢复，自由度收敛到单个步骤内部。自由循环跑关键业务流程，等于把系统正确性押注在模型的每次即兴发挥上。

补充一个我们踩过坑才理解的细节：状态机的「状态粒度」要按业务语义切，而不是按技术动作切。「调用大模型」不是状态，「完成风险分析」才是——前者是手段，后者是业务承诺。粒度切对了，检查点才有意义：恢复时你确信「风险分析已完成且可用」，而不是「某次模型调用返回过」。粒度切错的典型症状是恢复后状态自相矛盾，比如分析已经重跑了，报告却还是旧版本。

实操：一个可恢复的状态机实现

核心设计四点：状态与转移显式声明、每个成功步骤写检查点、失败按类型区分重试策略、全链路带护栏。完整代码如下：

import json, time, pathlib

class Workflow:
    """带检查点持久化的 Agent 工作流状态机：可恢复、可重试、有护栏"""

    def __init__(self, run_id: str, steps: list[tuple[str, callable]]):
        self.run_id = run_id
        self.steps = steps                                   # [(状态名, 处理函数)]
        self.state_file = pathlib.Path(f"./runs/{run_id}.json")
        # 启动时尝试从检查点恢复，没有则初始化
        if self.state_file.exists():
            self.ctx = json.loads(self.state_file.read_text(encoding="utf-8"))
        else:
            self.ctx = {"done": [], "current": None, "retries": {}}

    def _save(self):
        """每完成一步落盘一次，进程崩溃后从最后完成的步骤继续"""
        self.state_file.parent.mkdir(parents=True, exist_ok=True)
        self.state_file.write_text(json.dumps(self.ctx, ensure_ascii=False), encoding="utf-8")

    def run(self, max_steps: int = 30):
        """主循环：护栏 max_steps 防死循环，总步数超限立即终止"""
        executed = 0
        for name, fn in self.steps:
            if name in self.ctx["done"]:                     # 跳过已完成步骤，实现断点续跑
                continue
            if executed >= max_steps:
                raise RuntimeError(f"达到最大步数 {max_steps}，疑似死循环，run_id={self.run_id}")
            self.ctx["current"] = name
            self._save()
            result = self._exec_with_retry(name, fn)
            self.ctx["done"].append(name)
            self.ctx["results"][name] = result
            self._save()
            executed += 1
        return self.ctx.get("results", {})

    def _exec_with_retry(self, name: str, fn: callable):
        """重试策略：超时类错误指数退避重试，参数类错误不重试直接抛出转人工"""
        for attempt in range(3):
            try:
                return fn(self.ctx)
            except TimeoutError:
                wait = 2 ** attempt                          # 1s、2s、4s 指数退避
                print(f"[工作流] 步骤 {name} 超时，第 {attempt + 1} 次重试，等待 {wait}s")
                time.sleep(wait)
        raise RuntimeError(f"步骤 {name} 重试 3 次仍失败，需人工介入")

使用侧只需声明步骤，每步是一个纯函数，接收上下文返回结果：

def fetch_data(ctx):   # 拉取竞品动态（可重试的 IO 操作）
    return {"items": ["竞品A发布新版本", "竞品B融资"]}

def analyze(ctx):      # 调用 LLM 汇总分析，读取上一步结果
    items = ctx["results"]["fetch"]["items"]
    return {"summary": f"共 {len(items)} 条动态，主题集中在版本发布"}

# 声明式编排：步骤顺序即业务流程
flow = Workflow("daily-20261012", [
    ("fetch", fetch_data),
    ("analyze", analyze),
    ("report", make_report),   # 生成报告
    ("notify", send_email),    # 发送（幂等：以 run_id+step 作为幂等键去重）
])
flow.run()

进程在第 6 步崩溃后重启，run_id 不变即可从上次完成处继续；notify 这类外部副作用步骤以 run_id + 步骤名 为幂等键，重跑不会重复发邮件。

步骤间的上下文传递还要防止「内存膨胀」：八个步骤的中间产物全塞进上下文，跑到后面提示词长度爆炸、成本与错误率同步上升。我们的约定是每步只向上下文写入下游需要的「结论字段」，原始明细（如抓取的完整网页）写对象存储，上下文里只留引用路径。这一条同时服务于恢复效率——检查点文件始终维持在百 KB 量级，序列化与恢复都是毫秒级。

生产化还有两块拼图：每步观测与单步降级。观测上，把状态机的每次状态转移打点（步骤名、耗时、重试次数、结果状态），与可观测性体系打通，长流程的问题定位从「翻日志猜」变成「看时间线」。降级上，给每个 LLM 步骤准备一个保底实现——分析步骤失败时降级为模板拼接报告，保证日报「不那么聪明」也能准时发出，Agent 的可用性上限往往由降级方案决定，而不是模型本身。灰度则建议按「影子运行」起步：新流程先跑在只记录不发送的模式下两周，比对人工产出与 Agent 产出的差异清单，再逐步放开发送权限。

避坑总结

1. 状态只在内存里：进程重启、部署发版都会让长流程从零开始。每个成功步骤写检查点，恢复逻辑和执行逻辑同等重要——没有落盘的状态机只是「碰巧没挂」。
2. 工具调用不幂等：发送邮件、创建工单、扣减配额这类有外部副作用的步骤，重试或恢复重跑会造成重复执行。统一以业务幂等键去重，把幂等性做进工具层而不是祈祷不重试。
3. 没有步数与预算护栏：两个互相打回的步骤就能烧掉整月 token 预算。max_steps 硬上限之外，建议再加单次运行的成本上限，超限熔断并告警。
4. 失败一律盲目重试：超时、限流值得重试；参数错误、权限不足重试一百次也不会好，反而放大混乱。按异常类型分流——可重试的走退避重试，不可重试的直接转人工队列。
5. 把整条流程塞进一个巨型提示词：「你是一个助手，先做 A 再做 B 然后 C……」的写法，任何一步出错都无法定位和替换。单步职责单一、提示词独立，才能单独评测、单独换模型、单独重跑。

这套骨架还有一个隐性收益：步骤是显式声明的，替换模型就成了局部决策。哪个环节换更强的模型、哪个环节用小模型降本、哪个环节干脆改回传统规则引擎，都只改步骤注册表里的一行，不用动整个流程。编排清晰之后，优化空间才真正打开。

Agent 从演示走向生产，分水岭不在模型能力，而在工程骨架：状态机提供确定性，检查点提供恢复力，护栏提供止损线。把这三样做进去，长流程自动化才敢在没人盯着的时候跑。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。