把一个大任务直接丢给单个 Agent 处理，是目前很多 AI 自动化项目翻车的根源。我最近在跑一个内容生产项目时，需求说起来很简单：输入一个主题，自动产出一篇结构完整、有观点、有依据的文章。最初我用一个 Agent 把规划、调研、写作、校对全包了，提示词写得又长又绕，结果翻车率居高不下。不是漏掉关键信息，就是写到一半上下文撑爆，偶尔还会出现前后观点打架的情况。后来我把整个任务拆成了一条多 Agent 串行流水线，每个环节交给一个专职 Agent，节点之间用固定数据契约传递结果，并且给整条 Pipeline 节点链加上了可重试、可续跑的能力。折腾完这版之后，任务的稳定性和可控性直接上了一个档次。
这篇文章就是那次重构的完整复盘。重点讲清楚三件事：为什么单个 Agent 搞不定复杂任务，串行流水线该怎么设计，以及“可重试、可续跑”这两个能力到底怎么落地。如果你正在做 Agent 应用的开发，或者想用大模型跑稍微复杂一点的自动化流程，这篇应该能帮你少踩几个坑。
1. 为什么要把单个 Agent 拆成一条流水线
1.1 单体 Agent 的三个死穴
先说结论：一个 Agent 包打天下，在任务足够复杂的时候基本必炸。原因有三个。
第一个是上下文窗口的物理限制。大模型对话上下文是有限的，哪怕是最新一代模型，塞进去的内容也有天花板。一个 Agent 既要记住原始需求，又要处理调研资料，还要维护文章大纲、草稿、修改意见，几轮交互下来上下文就堆满了。这时候模型的表现力会急剧下降，开始出现“前面讲的事实后面忘了”“重点被稀释”这类问题。
第二个是错误放大效应。复杂任务中间一定会出错，但单体 Agent 的错误没有边界。规划出错会影响写作，写作出错会影响结论，而故障点在哪个环节、错误是什么时候引入的，往往要翻完整段对话才能定位。这会浪费大量排查时间，而且很多时候你明知道它错了，也只能通过改提示词整段重来。
第三个是不可恢复。单体 Agent 一旦中途失败，比如超时、接口报错、输出格式崩了，通常只能重开对话从零开始。哪怕前面 90% 的工作都是对的，也全部作废。对真实项目来说，这不仅是体验问题，还是成本问题——Token 和时间都烧不起。
类比一下就很好理解：一个工厂如果只有一个全能工人，从采购原料到质检出厂全让他一个人干，分工一旦复杂起来，他一定顾此失彼。流水线之所以存在，就是因为每个工位只做一件事，出了问题只影响当前工位。
1.2 流水线模式解决了什么问题
流水线模式的核心思路，是把一个复杂任务拆成多个有先后依赖关系的子任务，每个子任务由一个独立 Agent 处理，前一个节点的输出经过校验后，作为后一个节点的输入。
这样做的好处非常直接。
首先是上下文专一。每个 Agent 只需要关注自己那一段任务，比如“调研 Agent”只需要读资料、列要点，不需要操心文章结构；“写作 Agent”只需要拿到调研要点和标题，专心把段落写漂亮。上下文小了，模型注意力自然更集中，输出质量会稳定很多。
其次是错误边界清晰。哪个节点失败了，就直接定位到那个节点，不用再从头翻对话。你可以只重跑失败的节点，也可以只针对它调整参数，其他节点完全不受影响。
第三是可续跑。因为每个节点完成后都做了状态持久化，就算整个任务因为停电、进程被杀、接口连续失败等原因中断，下次启动时只要读取已保存的状态，找到最后一个完成的节点，从它的下游继续执行即可。这等于给流水线加了断点续传的能力。
一句话总结：流水线不是为了让流程变复杂，而是为了让“复杂”本身可控。
1.3 哪些场景适合串行流水线
串行流水线不是万能的，它解决的是“任务内部存在明确顺序依赖”的场景。我用一张表列一下自己总结的判断标准：
| 适合串行流水线 | 不适合串行流水线 |
|---|---|
| 任务有明显的前后依赖，比如“先调研，后写作” | 子任务之间互相独立，改 A 不影响 B |
| 每个子任务需要不同角色或不同侧重点 | 需要多个 Agent 实时讨论、来回博弈才能出结果 |
| 中间产物需要被后续环节反复引用 | 单次调用就能完成的轻量任务 |
| 对出错率、可审计性、可恢复性有要求 | 对响应延迟极敏感，比如实时对话场景 |
我在实际项目里用得最多的是内容生产流水线：调研、写大纲、写初稿、审核、修复、润色。也见过有人拿它做数据处理流程，比如数据清洗、格式转换、质量校验、入库，逻辑上完全一致。如果你是做简单问答或者单轮工具调用，直接一个 Agent 就够了，拆流水线属于过度设计。
2. 流水线设计：节点、数据契约与状态管理
2.1 节点类型怎么设计
流水线的最小单位是节点。我习惯把节点分成四类：执行节点、校验节点、修复节点、人工审核节点。
执行节点是最常见的，它封装一个具体的 Agent 调用。比如“调研 Agent”“写作 Agent”，负责干活的。
校验节点本身不调用大模型，或者调用一个轻量模型，专门检查上一个执行节点的输出是否符合要求。比如检查文章标题是否存在、输出 JSON 格式对不对、关键字段有没有缺失。它的存在是为了让“可重试”有的放矢——没有校验，你就不知道失败是因为格式问题还是内容问题。
修复节点是校验节点的下游。如果校验失败，修复节点会拿着校验反馈信息和之前失败的输出，打包成一条新的提示词让 Agent 重新生成。它比简单重试更聪明，因为它给了 Agent 一个“你为什么不行”的反馈。
人工审核节点则是让流程暂停下来，把中间产物交给人来判断。比如涉及金额、法律条款、品牌声誉的内容，卡一道人工审核很必要。这个节点的加入会打破全自动流程，但在真实业务里，它往往才是放心上生产线的关键。
每个节点内部需要配置的东西也很固定：提示词模板、输入输出的字段定义、最大重试次数、单次超时时间、重试退避间隔。
2.2 数据契约：节点间的接口协议
流水线能不能稳定工作，真正起决定作用的不是 Agent 有多聪明，而是节点之间的数据契约设计得有多严谨。
什么叫数据契约？就是每个节点的输入和输出格式。我会给每个节点定义一个 JSON Schema-like 的描述，明确必须包含哪些字段、字段类型是什么、可选的字段有哪些。
举个例子，“调研节点”的输出契约长这样：
{ "type": "object", "properties": { "topic": {"type": "string", "description": "调研主题"}, "key_points": { "type": "array", "items": {"type": "string"}, "description": "3-5条核心资料要点" }, "sources": { "type": "array", "items": {"type": "string"}, "description": "数据来源或参考文章" } }, "required": ["topic", "key_points", "sources"] }为什么要花心思做这件事？因为大模型输出天然不稳定，同一个要求，它这次可能给你规范的 JSON，下次就多包一层 Markdown 代码块。如果你在下游节点直接拿字符串硬解析，一个格式抖动就能让整条流水线崩掉。有了数据契约，每个节点执行完先做一次字段校验，不合法就进入重试或修复流程，下游拿到的数据永远是干净的。
这跟前后端联调必须定接口文档是一个道理。没有接口文档也能跑通，但一旦出问题，谁都不承认是自己那边坏了。
2.3 全局状态与局部状态
流水线在运行过程中会积累大量中间数据，我把它分成两层：全局状态和局部状态。
全局状态是整条流水线都需要访问的东西，比如用户最初输入的原始需求、当前 run_id、整体的配置参数。它贯穿所有节点，但内容通常不会太大。
局部状态是每个节点自己的产出，包括节点的输出结果、重试次数、耗时、状态、校验反馈等。每个节点只需要关心自己的局部状态和上一节点传给它的数据，不需要知道整条链上所有节点的细节。
状态管理上我最想强调的是“别把全部历史都丢给下一个节点”。很多人在实现流水线时偷懒，直接把前面所有节点历次运行的完整对话记录拼进下一个节点的上下文。这样做短期能跑，但随着节点变多，上下文爆炸是必然的。正确做法是每个节点只接收它需要的最小字段集，比如“写作节点”只需要收到“调研节点”提取出来的 key_points，而不需要收到调研 Agent 和用户之间的完整聊天记录。
状态快照也要考虑。我通常在节点成功结束后立刻保存一次快照，内容包括当前节点名、输入、输出、时间戳。这样即使后面所有节点全部失败，你也能知道失败发生时的完整现场。
3. 可重试与可续跑的实现要点
3.1 重试不是把同一份请求再发一遍
很多人理解的“可重试”就是失败了重新调用一次同样的接口。这个思路在纯计算场景下没问题，但在 Agent 流水线里，重试至少要注意三件事：幂等性、重试策略和失败分类。
幂等性是非常容易被忽视的。大模型调用本身是幂等的——你发同样的话，它生成的结果略有不同，但不会对外部世界造成副作用。问题出在那些“带动作”的节点上，比如一个节点会调用外部 API 发送邮件、提交订单、写入数据库。如果这个节点失败了，你不知道请求到底有没有成功到达外部系统，盲目重试很可能会产生重复操作。这种节点必须在数据契约里加入幂等键（比如生成一个 request_id），外部系统通过幂等键去重，重试才安全。
重试策略也很有讲究。最简单的方案是固定间隔重试，比如失败后等 5 秒再试。但更好的做法是指数退避，把重试延迟设为2 的 n 次方秒，第一次失败等 2 秒，第二次失败等 4 秒，第三次等 8 秒。这样做的原因是多数临时性错误都跟限流、服务过载有关，给服务留出恢复时间反而比疯狂重试更容易成功。
失败分类我平时会区分三种：
| 失败类型 | 典型案例 | 处理方式 |
|---|---|---|
| 可重试的临时错误 | 超时、限流、上游服务抖动 | 按退避策略重试 |
| 可修复的错误 | 输出格式不对、字段缺失、内容不达标 | 带反馈信息重试，或交给修复节点 |
| 不可恢复的错误 | 提示词写错、输入数据本身有误 | 直接终止，人工介入 |
一个非常容易踩的坑是把“内容不达标”和“格式错误”混在一起处理。格式错误可以通过修复提示词解决，但“内容不达标”比如调研结果太浅、文章逻辑不通，简单重试往往是浪费钱。这种真要重试，也应该在重试提示词里带上校验节点给出的具体反馈，比如“你漏了行业现状分析，请补充”，而不是原封不动地再跑一遍。
3.2 续跑：把状态固化到断点
可续跑的底层逻辑，是“每完成一个节点，就保存一次可恢复的检查点”。
我用的状态结构很简单，一个 run_id 对应一次完整流水线实例，内部用字典记录每个节点的运行结果。检查点文件的内容大致长这样：
{ "run_id": "8f3a9c2e", "pipeline_name": "article_pipeline", "status": "paused", "current_node": "writing_agent", "nodes": { "research_agent": { "status": "completed", "input": {"topic": "边缘计算在制造业的应用"}, "output": {"key_points": ["...", "..."]}, "attempts": 1, "finished_at": "2025-01-15T10:12:00Z" }, "writing_agent": { "status": "failed", "error": "output validation failed: missing field 'title'", "attempts": 3 } } }有了检查点文件，续跑的逻辑就非常简单了。启动流水线时先检查是否存在该 run_id 的检查点，如果存在，就只从current_node开始继续执行；如果不存在，就从第一个节点跑起。已经标记为completed的节点直接跳过，不重复调用，不重复消耗 Token。
这里要提醒一个细节：检查点文件一定要原子写入。直接覆盖写有进程崩溃后文件损坏的风险，我曾经就碰到过写了一半断电，状态文件只剩半个 JSON，导致整条流水线无法恢复。后来统一改成“先写临时文件，再用 rename 替换”的方式，这个问题彻底解决。
3.3 超时、熔断与成本控制
可重试和可续跑解决的是“能不能恢复”的问题，超时和熔断解决的是“别在恢复上烧太多钱”的问题。
每个节点必须设置单次超时时间。大模型接口经常出现长时间不返回的情况，如果你没有超时控制，一个卡死的节点会把整条流水线堵住。我通常把单次 Agent 调用超时设置在 60-120 秒之间，超出就按可重试错误处理。
熔断是另一个容易被忽略的机制。连续重试 5 次仍然失败的节点，继续重试没有意义。更好的做法是让流水线进入暂停状态，保存好现场，然后通知人来看。你可以设定一个规则：单个节点连续失败达到 3 次，自动暂停整个流水线。这样既不会无限烧钱，也保留了继续诊断的空间。
成本控制上我还有一个实测有效的技巧：所有节点的输出长度全部用参数限制死。有些 Agent 会在结果里附带大量解释和废话，这些内容既不进数据契约又不进下游上下文，完全是浪费。通过提示词加输出约束，能省下不少 Token 费用。
4. 从零搭建一个可续跑的多 Agent 流水线
4.1 最小实现：不依赖重框架能跑吗
先说结论：能跑，而且很稳。很多开源的 Agent 编排框架确实好用，但对一个固定结构的流水线来说，用 Python 原生逻辑自己写一个 Runner 反而更可控。框架帮你解决的问题是复杂状态管理、可视化调度、并行分支，但如果你只需要“按顺序跑几个节点”，自己写一个两百行的 Runner 完全够用，而且坑都踩在自己脚下，排查起来更快。
我用的结构是这样的：用dataclass定义节点配置和运行状态，用简单的循环遍历节点列表执行任务，每次节点执行完立刻保存检查点。
核心数据结构长这样：
from dataclasses import dataclass, field from typing import Callable, Dict, Any, Optional @dataclass class NodeConfig: name: str run_func: Callable[[Dict[str, Any]], Dict[str, Any]] output_schema: Optional[Dict[str, Any]] = None validator: Optional[Callable[[Dict[str, Any]], str]] = None max_retries: int = 3 timeout: int = 90 retry_base_delay: float = 2.0 @dataclass class NodeResult: status: str = "pending" input: Dict[str, Any] = field(default_factory=dict) output: Dict[str, Any] = field(default_factory=dict) error: str = "" attempts: int = 0 finished_at: str = "" @dataclass class RunState: run_id: str pipeline_name: str status: str = "running" current_node: str = "" nodes: Dict[str, NodeResult] = field(default_factory=dict)run_func是一个可调用对象，它内部封装了实际的 Agent 调用、提示词拼接、模型请求。校验逻辑用validator字段单独放，它返回空字符串表示通过，返回错误信息表示不通过。
4.2 Runner 的主循环逻辑
Runner 的思路很简单：遍历所有节点配置，跳过已完成的，执行未完成的，每次执行完保存检查点。核心代码示意如下：
import json, os, time def load_state(run_id: str, base_dir: str = "checkpoints") -> Optional[RunState]: path = os.path.join(base_dir, f"{run_id}.json") if not os.path.exists(path): return None with open(path, "r", encoding="utf-8") as f: data = json.load(f) return RunState(**data) def save_state(state: RunState, base_dir: str = "checkpoints"): os.makedirs(base_dir, exist_ok=True) path = os.path.join(base_dir, f"{state.run_id}.json") tmp_path = path + ".tmp" with open(tmp_path, "w", encoding="utf-8") as f: json.dump(state.__dict__, f, ensure_ascii=False, indent=2) os.replace(tmp_path, path) def run_node(node: NodeConfig, node_input: dict) -> NodeResult: result = NodeResult(input=node_input) for attempt in range(1, node.max_retries + 1): result.attempts = attempt try: output = node.run_func(node_input) if node.output_schema: validate_output(output, node.output_schema) if node.validator: error_msg = node.validator(output) if error_msg: raise ValueError(error_msg) result.status = "completed" result.output = output result.finished_at = time.time() return result except Exception as e: result.error = str(e) if attempt < node.max_retries: delay = node.retry_base_delay * (2 ** (attempt - 1)) time.sleep(delay) result.status = "failed" return result def run_pipeline(pipeline: list[NodeConfig], run_id: str, initial_input: dict): state = load_state(run_id) if state is None: state = RunState(run_id=run_id, pipeline_name="article_pipeline") for node in pipeline: state.current_node = node.name node_result = state.nodes.get(node.name) if node_result and node_result.status == "completed": continue node_input = build_node_input(node, state, initial_input) result = run_node(node, node_input) state.nodes[node.name] = result save_state(state) if result.status == "failed": state.status = "failed" save_state(state) break else: state.status = "completed" save_state(state) return state这段代码里最关键的几个点值得展开说。
第一个是build_node_input。这个函数决定了每个节点到底能看到多少上下文，而不是简单地把initial_input全部传下去。我一般会传入initial_input加上前置节点输出里的必要字段，但会过滤掉大的历史文本，只保留提炼后的结果。
第二个是“跳过已完成节点”的判断。没有这个判断，续跑功能就名存实亡。有了它，中断后重新启动，所有已完成的节点都不会重复执行，只会从current_node继续。
第三个是执行完一个节点立刻save_state。这样即使下一个节点崩溃，上一个节点的成果也已经保住了。
4.3 一条真实流水线的运行现场
我在内容项目里跑通的流水线叫“某技术趋势文章生产流水线”，一共 6 个节点：
research_agent：根据主题做资料调研，输出核心要点列表planning_agent：根据要点生成文章大纲writing_agent：按照大纲撰写初稿review_agent：检查初稿的问题，输出修改意见列表fixing_agent：根据修改意见修复内容polish_agent：做最后的语言润色和格式整理
我特意把review_agent和fixing_agent分开，就是为了让可重试更有价值。因为写作节点经常“自我感觉良好”，让它自己检查自己根本找不出问题。独立评审节点给出的反馈更客观，修复节点再拿着这些具体意见去重写，效果比盲改好得多。
运行日志大概长这样：
[1/6] research_agent -> completed (attempts: 1, 耗时 18s) [2/6] planning_agent -> completed (attempts: 1, 耗时 9s) [3/6] writing_agent -> failed (validator: missing field 'title') [3/6] writing_agent retry 2 with feedback: 缺少文章标题，请补充 [3/6] writing_agent -> completed (attempts: 2, 耗时 42s) [4/6] review_agent -> completed (提出 3 条修改意见) [5/6] fixing_agent -> completed (attempts: 1, 耗时 25s) [6/6] polish_agent -> completed (attempts: 1, 耗时 12s) run_id=8f3a9c2e status=completed中间如果进程意外退出，我只需要重新执行一次 Runner，传入同一个run_id，它就会自动加载checkpoints/8f3a9c2e.json，跳过前两个已完成节点，从writing_agent继续。这个体验跟下载软件断点续传差不多，谁用谁知道。
4.4 状态查看与故障定位的土办法
在没接可视化面板之前，我排查流水线问题主要靠两个土办法：看检查点文件，看日志。
检查点文件能告诉我每个节点的状态码、耗时、错误信息，基本能满足 80% 的排障需求。我会在每次重试时把错误信息记录到NodeResult.error里，多次重试的错误也会累积，方便看出“是不是每次错得不一样”。
日志方面，我建议每个节点开始前和结束后各打一行日志，格式固定为“节点名 + 状态 + 耗时 + 关键信息”。不要嫌日志多，Agent 流水线出问题时，没有日志你只能靠猜。
5. 常见问题与排查实录
5.1 重试了好几次输出还是不合法
这个情况我遇到过很多次，第一次直觉是“模型太笨”，后来排查发现大部分原因出在提示词和校验器的配合上。
最常见的问题有两个：一是校验器要求太严，比如要求输出严格 JSON，但提示词里没给例子，Agent 一直按自己的习惯输出 Markdown 格式，校验永远失败。二是重试时没有把失败原因反馈给 Agent，它根本不知道要改什么。
解决办法是给校验失败的情况专门设计一条反馈分支。捕获到格式错误时，把具体的错误信息拼进重试提示词，比如“你上一轮输出被格式校验拦截：缺少字段 title，请严格按以下 JSON 结构输出”。实测这一招能把重试成功率拉高很多。
如果带反馈重试 3 次还是不行，就别继续烧钱了。这时候问题往往出在任务本身，比如输入主题太模糊、节点提示词写反了。让修复节点介入，或者干脆人工看一眼，比无限重试强。
5.2 检查点文件损坏导致续跑失败
检查点文件损坏是续跑功能最大的敌人。我踩过最惨的一次坑，是进程在写状态文件的瞬间被强制杀掉，文件只剩半个 JSON 结构，第二次启动直接解析失败。那一次整条流水线不得从头跑，气的我在项目里加了一条硬性规则：所有状态文件必须原子写入。
技术实现不复杂，核心就是“先写临时文件再改名”，临时文件写完整后os.replace原子替换旧文件。这样即使进程崩溃，最多丢失本次未写完的临时文件，已完成的旧文件是完整的。
另外建议定期备份检查点文件，尤其是长任务。我通常是每次任务彻底跑完以后，手动把检查点目录复制一份到带日期的子目录，以备回滚。
5.3 节点有外部副作用时重复重试造成重复执行
流水线节点不一定只是调用大模型，很多节点会带着外部动作。最典型的是“发送通知邮件”和“写入业务数据库”。
问题在于，如果节点调用外部 API 时超时，你无法确定外部到底是处理成功了还是失败了。此时盲目重试，可能造成邮件发两封、数据库多插入一条记录。
解决方案是给这类节点设计幂等键。每次生成一个全局唯一的 request_id，传给外部系统，外部系统记录已处理过的 request_id，重复请求直接返回第一次的结果。这个思路跟支付系统的幂等机制是一样的。
如果外部系统不支持幂等键，那就反着来：把“发送动作”和“确认结果”拆成两个节点。第一个节点只负责生成内容并保存到本地，第二个节点才实际推送给外部。重试只重跑第一个节点，外力推送节点失败后走人工确认，从设计上避免重复。
5.4 上下文爆炸和 Token 超预算
流水线节点多起来以后，最常见的隐性问题是“每个节点都在偷偷长胖”。明明数据契约里只需要传 5 个字段，但如果你在实现时把整个历史记录都塞进run_func的上下文，前面节点的输出会随着流水线长度一路带到最下游。
排查方法很直接：在每个节点执行完以后，打印一下实际传给大模型的消息字节数。如果发现某个节点输入远超预期，检查build_node_input是不是没做字段裁剪。
节省 Token 还有一个技巧：对调研节点这类输出大量的中间环节，不要直接把它全部输出传给下游，而是先做一个“摘要节点”。这个节点用轻量模型，把上游输出提炼成下游真正需要的要点。多花一次调用，省下的是后面多次调用的上下文空间。
5.5 限流错误处理欠妥导致重试越等越久
大模型接口限流是家常便饭。如果你用的退避策略是固定间隔，比如每次都等 5 秒，一旦上游限流持续超过几秒，你的重试会密密麻麻地撞在枪口上。
比较好的做法是根据错误类型区分等待时长。限流类错误通常带有Retry-After响应头，或者错误信息里明确写了“请 X 秒后重试”。那就按它说的来，而不是用统一的退避公式。
如果错误信息里没给建议，就用指数退避，而且把基准时间调大一点，比如第一次失败等 5 秒，第二次等 10 秒，第三次等 20 秒。宁可等待时间拉长，也不要连续高频重试触发更严厉的限流规则。
6. 经验总结与后续扩展
6.1 项目里最有价值的三个认知
整套机制跑完，我最大的体会是：多 Agent 流水线的核心瓶颈，从来不是模型智商，而是工程化能力。
第一个认知是数据契约比 Agent 数量重要。哪怕你只有两个 Agent，只要它们之间的数据格式松松垮垮，流程就一定不稳。反过来，节点再多，只要每个节点输入输出都像接口一样规范，整条链就可以稳定运行。
第二个认知是状态管理优先级高于一切功能。可重试、可续跑、可视化调度，这些听起来高大上的能力，底层全是状态管理。先把检查点、恢复逻辑做扎实，再谈其他优化，顺序不能反。
第三个认知是串行不等于低效。很多人一听串行就觉得慢，但串行的确定性、可调试性和低上下文开销，在复杂度上来之后带来的价值远超那点时间损耗。并行该加的时候再加，但它应该是稳定之后的放大器，而不是稳定本身。
6.2 后续可以继续扩展的方向
这套基础框架跑通之后，可以扩展的玩法挺多。
可视化面板值得做。检查点文件已经记录了每个节点的状态和耗时，接一个 Web 界面，把节点画成卡片流，跑起来的时候实时刷新状态，对团队协作和汇报都很有帮助。
并行分支适合用在“同一节点需要处理多个独立子任务”的场景。比如调研阶段需要同时查多个信源，每个信源开一个并行的 Agent 调研，最后汇总。这个升级需要在 Runner 里引入“扇出-汇聚”逻辑，但整体架构不需要推倒重建。
动态规划是另一个方向。传统流水线节点和执行顺序是提前定死的，升级版的思路是让一个“调度 Agent”根据前序节点的输出，动态决定下一个节点是谁。举个例子，审核节点发现文章表格数据有误，可以直接把任务路由到“数据处理 Agent”，而不是按固定顺序进“润色 Agent”。
最后再分享一个小技巧：我在做检查点时，会把当前节点的提示词版本也一并存进去。这样如果后续改了提示词导致结果变差，直接回滚检查点对应的提示词版本就能快速定位问题。这个小习惯，在长流水线迭代时真的能救命。
