Agent任务拆成可重试、可续跑的Pipeline节点链，长任务再也不怕中途失败

把一个稍复杂的任务直接丢给 Agent 一路跑到底，最难受的不是模型偶尔抽风，而是中途任何一步出故障之后的处理方式。我之前在做一个批量文档生成项目时，整条流程需要多个 Agent 依次协作：需求解析、资料检索、初稿生成、事实核查、风格改写、终审定稿。这就是典型的多 Agent 串行流水线。最初我把整条链路写成一个大的编排函数，结果只要中间某次模型服务超时、输出格式不合法，或者某个校验没通过，整个任务就废了，所有上游节点全部白跑，又得从头再来一遍，烧掉的 token 和时间都让人肉疼。

后来我下决心把任务彻底拆成可重试、可续跑的 Pipeline 节点链，才算真正解决了长任务可靠性的问题。这套方案的核心思想不复杂：把一次大任务划分成若干有前后依赖关系的节点，每个节点独立执行、独立校验、独立记录产物；节点失败可以局部重试；整个流程意外中断后，可以基于断点快照从失败处继续跑，而不是全盘重来。这篇文章就把这套节点链的设计思路、核心数据结构、重试与续跑的关键逻辑，以及我实际踩过的坑完整梳理一遍，适合正在做 Agent 自动化、想把任务编排做得更健壮的开发者参考。

1. 为什么要拆成节点链：长期跑 Agent 任务的人最懂这个痛点

1.1 “一步错、全盘重来”是长任务的第一大坑

先说说最初的做法有多蠢。早期我把整条 Agent 流水线写成一个几十行的编排函数，大概逻辑是这样的：调用 Agent A 做完需求解析，拿到结果后拼进 prompt，再调用 Agent B 做资料检索，然后再调用 Agent C 生成初稿……整条链路像一条长绳，任何一个环节抛异常，我只能在最外层用 try / except 兜底。

问题在于，异常一旦发生，这条长绳上的所有中间结果都丢了。比如第 3 个节点已经生成了一份两万多字的初稿，结果第 4 个节点的事实核查因为外部检索接口返回密集超时，导致整条流程失败。我重新跑一遍，第 1、2、3 个节点全部重新执行。单次重跑的成本不止是时间：一次完整的生成可能消耗十几万 token，如果重试三次，光 token 费用就很可观。更气人的是，重跑过程中初稿生成这类节点本来就是概率性输出，同一个输入重新调用模型，得到的内容往往不一样，前一次可能生成得很好，重跑反而变差了。

这就是长任务场景下最常见的一个坑：你真正想要的是“哪一步失败就补哪一步”，但粗放的编排方式逼着你“全盘重来”。

1.2 节点化之后换来的三个能力：重试、续跑、观测

把任务拆成节点链之后，最直接的变化是获得了三个能力。

第一个是局部重试。某个节点失败时，不需要重跑上游节点，只需要让这个节点自己再试几次。比如初稿生成节点调用模型服务时遭遇网络抖动，我可以对这个节点做指数退避重试，重试成功之后继续往下走。

第二个是断点续跑。这是比重试更硬核的能力。重试主要解决“进程还活着、短时间内恢复”的场景；续跑解决的是“整个任务中断了半小时，甚至服务器重启了”的场景。做法是在每个节点成功之后立刻把状态快照写到磁盘，进程恢复时加载快照，跳过已经成功的节点，从失败节点继续跑。这样不管任务是跑挂了、被 kill 掉，还是停电了，都能像打游戏存档一样接着来。

第三个是状态可观测。节点化之后，每个节点有明确的 id、状态、开始时间、结束时间、产物路径。你可以随时回答“现在跑到哪一步了、哪一步失败了、失败原因是什么”。这在排查问题时的价值远大于爽，尤其是多 Agent 协作的任务，光靠看日志猜流程是极其痛苦的。

1.3 什么样的任务适合串行流水线

并不是所有任务都适合拆成串行节点链。我自己总结了两类场景。

一类是前置依赖强、产出形态固定的任务，典型代表是内容生产流水线，比如“需求拆解 → 资料收集 → 初稿创作 → 事实核查 → 风格改写 → 终审定稿”。这种任务的前后顺序基本不能乱，每个节点的输出都有明确的下一步消费者，非常适合串行节点链。

另一类是涉及多次外部调用、单次运行时间长的任务，比如批量文档处理、报告自动生成、代码生成链。这类任务对稳定性要求高，因为跑一次需要好几分钟甚至几十分钟，一旦中途失败损失太大。

不太适合串行流水线的是那些需要自由探索、并行搜索的任务，比如让一堆 Agent 各跑各的然后汇总结果。那种场景更适合并行 Fan-out，而不是串行链。搞清楚自己的任务形态再做架构决策，比一上来就套框架重要得多。

2. 数据骨架设计：让“当前进度”变成可持久化、可恢复的状态

2.1 核心数据结构：NodeRecord 与 PipelineState

节点链的工程基础是一套能表达“任务走到哪一步了”的数据结构。我最终用的核心模型长这样：

# pipeline_models.py from dataclasses import dataclass, field from typing import Any, Dict, Optional @dataclass class NodeRecord: node_id: str # 节点唯一标识，例如 "draft_generation" name: str # 人类可读的名称 status: str = "pending" # pending / running / success / failed / skipped attempts: int = 0 # 已经尝试的次数 input_refs: Dict[str, str] = field(default_factory=dict) output_refs: Dict[str, str] = field(default_factory=dict) error: str = "" started_at: str = "" finished_at: str = "" @dataclass class PipelineState: run_id: str nodes: list[NodeRecord] current_index: int = 0 payload: Dict[str, Any] = field(default_factory=dict) created_at: str = "" updated_at: str = ""

每个节点的 input_refs 和 output_refs 存的是产物路径或上下文键名，而不是实际的大文本。比如初稿生成节点执行完后，output_refs 里可能存着{"draft_path": "artifacts/run_20250101_120000/draft_generation/output.md"}。这样状态快照文件体积很小，而且恢复执行时可以直接从磁盘路径读取产物，不需要把上万字的内容塞进 JSON。

PipelineState 里的 payload 用来存放跨节点传递的公共上下文，比如项目主题、目标用户、全局约束条件等。它同时也是一个轻量级的“黑板”，所有节点都可以往里面写键值，也可以从里面读键值。注意，我不建议在 payload 里存大文本或二进制数据，只存普通 JSON 可序列化的内容，大对象一律落盘后引用路径。

2.2 节点状态语义：五种状态定义清楚，恢复逻辑才不会乱

状态机的设计直接影响恢复逻辑的复杂度。我定义了五种状态：

恢复时最重要的一个判断是：如果一个节点的状态是 running 但整个流程已经中断，那么它必须被视为 failed。原因很简单，这个节点可能已经执行到一半，但我们没有拿到它的有效输出，最安全的做法就是重新执行。至于重新执行会不会产生副作用，我在后面幂等那一节会详细展开。

2.3 状态快照如何落地：先写临时文件再原子替换

状态快照的保存动作必须足够可靠。我的实现是每次节点状态变更后立即做一次快照保存，保存方式是先写到state.json.tmp，然后用os.replace原子替换state.json。这样做有一个好处：任何时刻磁盘上的 state.json 都是完整、合法的，不会被半个写入过程破坏。

# snapshot.py import json, os, tempfile def save_state(state: PipelineState, path: str): tmp_path = path + ".tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(asdict(state), f, ensure_ascii=False, indent=2) os.replace(tmp_path, path)

快照保存的时机也很关键：每个节点成功之后保存一次，节点失败并准备重试之前也保存一次。这样即使进程在节点执行过程中被打死，我们最多只会丢失当前节点未完成的产出，已完成节点的产物和记录都还在。

3. 可重试的细节：退避、幂等与节点校验，一个都不能少

3.1 三种失败类型，三种重试策略

不是所有失败都值得无脑重试。我在实践里把节点失败分成三类，分别采用不同策略：

调用类失败用指数退避是最稳的。我用的退避公式是sleep_time = base * (2 ** attempt) + random_jitter，其中 base 通常设为 1 秒，jitter 是 0 到 1 秒之间的随机值。这样既不会在服务刚恢复时把自己重试到触发限流，也能避免所有节点同时重试造成“惊群”。

最简单的一个实现长这样：

# retry.py import random, time def retry_with_backoff(func, max_retries=4, base_delay=1.0): for attempt in range(max_retries): try: return func() except Exception as e: if attempt == max_retries - 1: raise sleep_time = base_delay * (2 ** attempt) + random.uniform(0, 1) time.sleep(sleep_time)

3.2 幂等是重试的底线：不能让节点重跑产生副作用

调用类失败重试起来比较自然，但校验类失败和“进程中断后恢复”的场景里，一个更隐蔽的问题是：节点重跑会不会产生副作用？

模型调用本身通常没有副作用，你调一次模型和调两次模型，最多只是多花点 token。但节点里如果包含了外部系统的写入操作，问题就大了。比如某个节点会把生成的内容通过接口发布到内部系统，第一次调用实际上成功了，只是返回超时；重试时又提交了一次，结果就产生了重复数据。这是重试机制最大的坑：没有幂等保护的节点，不配被重试。

我在代码里应对这个问题的办法有两个。第一，每个节点在执行开始时生成一个 request_id，并把这个 request_id 写入上下文；凡是涉及外部写入的调用，一律携带这个 request_id 作为幂等键。第二，节点的中间产物按“运行 ID + 节点 ID + 尝试次数”分目录保存，杜绝重试覆盖首次尝试产物的问题。

# node_with_request_id_example.py import uuid def execute_with_idempotency(node_id: str, request_id: str): # 假设这里调用某个外部服务写入内容 # 外部服务应该根据 request_id 去重 payload = { "request_id": request_id, "content": "some generated content", "node_id": node_id, } response = external_service.submit(payload) return response

在实际项目里，你要么保证外部系统支持幂等键，要么就把写入动作拆成一个独立的人工确认节点，绝不放在自动重试链路上。这是我从一次线上事故里换来的教训：某次节点重试导致内部系统产生了三份重复内容，清理数据的代价比重新生成内容高得多。

3.3 校验器设计：节点输出只有通过校验才算成功

重试策略里最重要的一环其实是校验器。一个节点执行完之后，不能直接拍板“成功了”，必须先过校验器。校验器负责回答这么几个问题：输出格式对不对？关键字段齐不齐？内容质量有没有明显问题？比如我要求初稿生成节点输出一个包含title、summary、body和sources四个字段的 JSON，那么校验器就要检查这四件事：

# validators.py def validate_draft_output(raw_text: str) -> tuple[bool, str]: try: data = json.loads(raw_text) except json.JSONDecodeError as e: return False, f"输出不是合法 JSON: {e}" required_fields = ["title", "summary", "body", "sources"] for field in required_fields: if field not in data or not data[field]: return False, f"缺少字段: {field}" if len(data["body"]) < 500: return False, "正文长度过短" return True, ""

校验不通过和调用抛异常是两个完全不同的信号。调用抛异常说明大概率是外部问题，重试可能有效；校验不通过说明模型返回了内容但不合规格，重试的效果也不一定好，所以我通常给校验失败设置更少的重试次数，并且重试时会在 prompt 里带上错误信息，明确告诉模型“你上次的输出格式有问题，具体是：……”。这种“带着反馈重试”的方式，比单纯静默重调模型有效得多。

4. 可续跑的实现：断点快照、状态机与恢复执行逻辑

4.1 续跑的核心不是重跑，而是“只补缺口”

断点续跑听起来高深，本质其实很简单：恢复的时候找出哪些节点没有成功，只执行这些节点。已经成功的节点统统跳过，直接复用它们的产物。

这里有一个容易想错的地方：恢复执行时，失败节点后面还没跑过的节点该怎么办？答案是继续按顺序执行，不需要任何特殊处理。比如一个 6 节点流水线，第 4 个节点失败，恢复时第 1、2、3 个节点直接跳过，第 4 个节点重新执行，执行成功后继续执行第 5、6 个节点。恢复执行本质上就是“从失败节点开始重新执行剩余节点”。

4.2 失败节点的上下文依赖怎么重建

我刚开始做恢复功能时遇到一个很实际的问题：第 4 个节点重新执行时，它的输入要从哪里拿？

如果整个任务只有一份全局上下文 payload，那很简单，因为所有节点的输入都在 payload 里，失败节点重新执行时读 payload 就好了。但实际项目中，节点可能依赖上游节点的产物文件，而这些产物路径是在上游节点执行时才写入 output_refs 的。

所以正解是：恢复执行前，先把所有成功节点的 output_refs 收集起来，合并回上下文中。我在具体实现里是这样的：加载快照后，遍历所有状态为 success 的节点，把它们的 output_refs 灌入 payload；然后从第一个状态不是 success 的节点开始继续执行。这样第 4 个节点重跑时，就能从 payload 里读到draft_path、research_notes_path这些键，顺利拿到上游产物。

4.3 恢复执行的主循环代码

下面是一个简化版的主循环实现，核心逻辑就是“扫描节点，跳过成功，执行失败，写快照”：

# pipeline_executor.py class PipelineExecutor: def __init__(self, state: PipelineState, artifact_root: str): self.state = state self.artifact_root = artifact_root def restore(self): # 将已完成节点的产物引用灌回 payload for node in self.state.nodes: if node.status == "success": self.state.payload.update(node.output_refs) def run(self): self.restore() idx = 0 while idx < len(self.state.nodes): node = self.state.nodes[idx] if node.status == "success" or node.status == "skipped": idx += 1 continue # 进程中断后遗留的 running 节点，视为失败重新执行 if node.status == "running": node.status = "failed" node.error = "interrupted before completion" node.status = "running" node.attempts += 1 try: output, output_refs = self.execute_node(node) node.output_refs = output_refs node.status = "success" self.state.payload.update(output_refs) save_state(self.state, self.state_path) except Exception as e: node.error = str(e) node.status = "failed" save_state(self.state, self.state_path) # 这里可以按失败类型决定是否继续重试, 否则抛出 raise idx += 1 save_state(self.state, self.state_path)

主循环的逻辑不复杂，但工程细节都在 execute_node 内部和状态保存的时机上。每成功一个节点就保存一次快照，是我最推荐的频率；如果你合并多个节点再保存，一旦中断，丢失的进度会变多，恢复的开销也变大。

5. 一次真实故障的抢救复盘：断在第 4 个节点的完整恢复过程

5.1 实例背景：6 节点内容生产流水线的节点划分

为了让上面的设计更具体，我讲一次实际的故障抢救过程。当时我在跑一条 6 节点的内容生产流水线，节点划分如下：

流水线的上下文是串行传递的，每个节点都依赖上一个节点的产物。N3 生成的初稿大约两万字，N4 需要读这份初稿，逐条核对事实。

5.2 故障现场与排查链路

那天 N4 跑挂了。日志里显示它调用外部检索接口时，连续五次超时，指数退避重试也耗尽了。节点状态变成 failed，错误信息是external service timeout after 5 retries。更麻烦的是，整个进程在 N4 重试耗尽后被人为中断了，因为我想先看看是不是检索服务的问题。

排查过程我分了三步走。

第一步是看快照，打开 state.json，确认 N1、N2、N3 都是 success，N4 是 failed，N5、N6 是 pending。这说明前三个节点的产物都还在磁盘上，恢复时不该再浪费 token 去重跑它们。

第二步是检查产物文件是否完整。我写了一个小脚本遍历 artifacts 目录，确认 outline.json、research_notes.md、draft.md 都存在且文件大小正常。这一步非常关键：因为系统可能在进程被强杀时，文件已经写完但快照还没来得及更新。如果出现这种情况，要以磁盘上的实际文件为准，快照只能作为辅助参考。

第三步是修复 N4 的实现后手动恢复。那次的根因是检索接口对单个请求的处理时间设置太短，我在 N4 内部把单次请求的超时时间从 10 秒调到了 30 秒，然后重新执行整个流水线。恢复时我看到日志很快打出了skip N1..N3的提示，N4 重新执行成功，N5、N6 继续跑，整个过程只花了正常全量跑的三分之一时间，token 消耗更是只有全量的四成。

5.3 全量跑 vs 断点恢复跑的实测数据

那次故障恢复后我做了一个简单的对照组，结果很说明问题：

从那次之后，我把“尽量别让上游节点陪跑”变成了团队里的基本共识。这种收益越是在长任务里越明显：节点越多、单节点越重，断点续跑省下的成本就越可观。

6. 从串行链扩展：条件分支、人工审核与可观测性

6.1 加上条件分支：根据节点结果决定下游走哪条链路

串行链路做到稳定之后，下一个自然不过的扩展就是条件分支。比如在事实核查节点跑完后，可以判断核查结果：如果存疑项超过阈值，走“人工介入”分支；如果一切正常，直接走风格改写分支。

实现条件分支不需要推翻已有的节点链模型，我用的办法是给 NodeRecord 增加一个可选的next_node_id字段，或者更通用一点，用一张边表来表达节点之间的依赖关系：

# edges_example.py pipeline_edges = [ ("requirement_parsing", "info_retrieval"), ("info_retrieval", "draft_generation"), ("draft_generation", "fact_checking"), ("fact_checking", "style_rewriting", {"if": "check_score >= 0.8"}), ("fact_checking", "human_review", {"if": "check_score < 0.8"}), ]

这套东西本质上是把串行链演进到有向无环图的雏形。不过我的建议是，如果需求真的复杂到需要任意 DAG，就不要自己维护边表了，直接换成熟的流程引擎，把精力放在节点本身的设计上。

6.2 插入人工审核节点：把重试和续跑的概念延伸到人机协同

另一个对我很有用的扩展是人工审核节点。有些内容即使模型自己觉得没问题，也需要真人过目一遍才能放行。这种节点在流水线里是一个特殊存在：执行到这里时，系统生成一份审核工单，然后把节点状态置为 running，等待人工在后台操作后回调接口，把状态改为 success，流水线继续往下跑。

人工审核节点的设计让“可续跑”这个词的意义从“进程恢复”扩展到了“任务暂停”。比如周五晚上跑到人工审核节点停下，周一早上审核人确认通过，流水线继续跑完剩下节点。状态快照机制完全不用改，只需要给审核节点增加一个等待回调的状态迁移逻辑。

6.3 节点状态变更的事件钩子

最后强烈建议做一件事：节点状态变更时触发事件钩子，把状态推送给观测系统。我自己的实现很简单，在每次 save_state 之后调用一个事件回调：

# events.py def emit_state_change(state: PipelineState, node: NodeRecord): report_json = { "run_id": state.run_id, "node_id": node.node_id, "status": node.status, "attempts": node.attempts, "error": node.error, } metrics_client.send("pipeline.node_state", report_json)

有了这些事件流，你就可以在内部看板上实时看到每个任务跑到哪个节点、哪个节点重试了几次、哪条流水线最近失败率偏高。这些数据反过来会指导你调整重试策略和节点划分粒度。

回到开头那个问题：为什么一定要把 Agent 任务拆成可重试、可续跑的 Pipeline 节点链？我的答案已经写在前面了——拆的不是代码，而是你对“失败”这件事的态度。节点化之后，失败变成了流程里的普通状态，可以重试、可以跳过、可以转人工、可以断点续传，而不是让整条流水线陪着一起翻车。这套思路我用了很久，越用越觉得：先把任务拆清楚，再考虑 Agent 怎么发挥，才是做多 Agent 协作最值得花时间的环节。