★ FEATURED ARTICLE 多 Agent 串行流水线这种方案说白了就是把过去一个大 Prompt 丢给单 Agent 的粗暴玩法拆成一条可编排、可重试、可续跑的节点链。每个节点只干一件具体的事干完把结果塞给下一个节点链路挂掉之后还能从断点接着跑。这不是理论推演是实际搞过几轮之后觉得真正能落地的架构思路。先说一下我为什么盯上这个方向。早期做 Agent 应用最头大的就是单 Agent 干复杂活的时候要么答非所问要么中途情绪化罢工。后来试过并行多 Agent 群聊调度效果有好转但调试成本高得吓人Chain 一长经常出现上下文串味、节点结果互相污染。直到把串行流水线和任务快照结合在一起才真正确实解决了可重试、可续跑这两个核心痛点。这篇就把我实际搭建这套 Pipeline 节点链的过程、踩过的坑、还有关键节点的设计逻辑都记录下来。适合那些已经在做 Agent 应用、但被单 Agent 复杂任务稳定性搞到头疼的开发者参考。1. 为什么单 Agent 做复杂任务大概率会翻车很多人会觉得一个大模型就能把所有事干完为什么非要拆成多个 Agent 串起来跑。这个疑问我一开始也有直到被现实教育了几次才彻底想明白。复杂任务拆解不是“为了架构而架构”而是单 Agent 执行长链路任务时确实存在几个结构性的硬伤。1.1 上下文窗口的“安全容量”其实很窄单个 Agent 处理长任务时上下文窗口看起来很大但真正能稳定利用的有效信息密度是有限的。越到任务后期早期塞入的原始材料、中间推理过程就越容易在注意力机制里被“稀释”。一个真实的感受是让单个 Agent 从一段长文档里提取关键信息再进行结构化输出再基于结构化结果做二次分析它往往会在第二个、第三个环节就开始丢细节。不是模型变笨了而是它在同一个上下文里同时维护“原始数据、提取规则、中间结果、最终输出格式”这四层信息时注意力分配会失衡。串行流水线每个节点只维护自己需要的那一小段上下文反而让每个环节的专注度更高。另外单 Agent 一旦在任务中途接了用户的新输入或者发生了工具调用的异常返回整个上下文就开始“脏”了。之后的输出会带着前面错误的影子很难剥离。这种污染是累加的任务越长越明显。1.2 链路越长单点失败的影响越难排查我之前维护过一个相对复杂的单 Agent 自动化流程里面要顺序完成数据抓取、格式转换、内容生成、质量校验四个阶段。结果就是每次跑挂掉排查问题都特别痛苦。因为你根本不知道是数据抓取阶段返回了畸形结构还是格式转换阶段的 Prompt 让模型产生了误解还是质量校验阶段把合格内容误判成了不合格。单 Agent 的中间过程是一个“黑盒”它处理到哪一步、哪一步出了问题几乎无法精确观测。而串行流水线的每个节点是独立的哪一步挂了、返回了什么、重试了几次全部有日志有记录。这种可观测性是保证流程可维护的基础也是我愿意付出额外架构成本去换取的收益。1.3 串行 Pipeline 本质上是在“拆风险”把一个复杂任务拆成节点链相当于把一个大风险拆成多个小风险。每个节点只负责一个逻辑单一的子任务产生的结果是确定结构的。下游节点只需要对上游节点负责不需要关心初始数据长什么样也不用关心整个任务的全貌。打个比方单 Agent 就像一个人既要买菜、又要切菜、还要炒菜、还要摆盘忙起来容易手忙脚乱。串行流水线就像一个后厨流水线每个环节一个人只做一件事东西传过来就处理做完就往下一个环节送。单个环节出了问题重做一道菜的成本也远低于整桌菜全部推翻重来。这种“拆风险”的思路还有一个隐藏的好处可以针对每个节点的特点单独设计 Prompt 策略和工具集。比如某个节点需要严格 JSON 输出某个节点需要调用外部 API某个节点需要高强度推理。这些策略不再需要塞在一个巨型 Prompt 里互相打架而是各归各的位置。2. 流水线节点链的核心架构设计确定了要用串行流水线方案之后接下来的重点就是怎么设计这条链。好的架构设计能让你在节点增加、删除、重排时像搭积木一样轻松坏的架构设计会让流水线变成一坨纠缠不清的意大利面。2.1 三个核心设计决策前序驱动、任务快照、节点契约第一个决策是“前序驱动”模式。流水线不采用中心调度器统一控制所有节点而是每个节点只感知上游节点的输出。这样节点之间的耦合度降到最低。当你想要在中间插入一个新节点时只需要修改上游节点的输出契约和下游节点的输入适配其他节点完全不需要改动。第二个决策是“任务快照”。每个节点处理完自己的任务后会把完整的输入、输出、元数据打包成一份快照持久化到存储里。这份快照就是可续跑的基石。整个流水线如果在中途挂了重跑时只需要找到最后一个成功的快照位置从这个位置往后继续执行就行不需要从头再来。第三个决策是“节点契约”。每个节点严格定义输入格式和输出格式这个契约通常是一个 JSON Schema。节点收到输入先做格式校验不合法就直接报错不进入后续环节。这个设计保证了每个节点拿到的数据是可靠的避免了脏数据在流水线里越传越脏的情况。2.2 节点间的数据传递不该用“纯内存引用”流水线跑的时候很多人的自然反应是节点 A 返回一个结果直接作为变量传给节点 B内存里引用一下就完事了。这个做法在小规模试用时没问题一旦流水线跑长跑久就会暴露出三个问题无法恢复、无法观测、无法回溯。我的做法是把每个节点的输出都序列化成结构化数据写入本地存储或者对象存储。下游节点启动时从存储中读取上游的数据。这样做的代价是多了一次 IO但换来的是每个节点都可以独立重跑、独立调试、独立验证。你不需要把整条链路跑起来才能测某个节点的逻辑单独拿一份历史快照就能把它喂饱。这里有个实操细节快照的存储结构建议用“任务 ID 节点序号”作为主键。比如一个任务的 ID 是 task_20250115_001那么第三个节点的快照就存成 task_20250115_001/03_output.json。这样在 Web UI 或者日志系统里查看整条流水线的状态时一目了然。2.3 节点的状态机是重试和续跑的基础每个节点在流水线里的生命周期不是简单地“运行中”和“已完成”。我实际建模时用了一个五状态的状态机PENDING 待执行RUNNING 运行中SUCCEEDED 成功FAILED_RETRYABLE 失败但可重试FAILED_TERMINAL 失败且不可恢复节点开始执行前是 PENDING执行起来转 RUNNING正常结束转 SUCCEEDED执行过程中抛出可以被捕获的异常则转 FAILED_RETRYABLE只有确认这个错误再怎么重试也没用比如输入数据结构错误、上游数据缺失、配置不可用才转 FAILED_TERMINAL。这个状态机是整个流水线能够“可重试、可续跑”的核心。每一个状态迁移都有记录重跑的时候只需要从状态库里找到最后一个处于 PENDING 或者 FAILED_RETRYABLE 的节点从那里接着跑就好。3. 可重试机制的具体落地方式重试不是简单地“失败了就再跑一次”那样碰到某些类型的错误会陷入无限循环浪费大量资源。可重试机制的背后是对失败类型的精确分类、对重试策略的细化配置以及对幂等原则的严格遵守。3.1 分类处理可以重试的错误 vs 不能重试的错误我在设计阶段把 Agent 节点的错误类型分成了三类。第一类是基础设施类错误比如 LLM API 超时、限流、临时网络抖动。这类错误的重试价值非常高通常等待一段时间重试就能成功。第二类是模型输出格式类错误比如 Agent 返回了非法的 JSON、某个字段缺失。这类错误有时候值得重试因为调整一下参数或者温度设置模型可能就能输出正确格式。但重试次数不能过多我一般设置不超过 2 到 3 次。第三类是业务逻辑类错误比如输入数据根本不能推导出目标结果、Prompt 配置有逻辑冲突、工具依赖的外部系统返回了永久性错误。这类错误重试没有意义应该直接标记为 FAILED_TERMINAL进入人工排查流程。如果你把这三类错误混在一起统一重试轻则浪费资源重则把业务逻辑错误当成临时故障无限重试流水线卡死在一个注定失败的任务上。3.2 指数退避 最大重试次数防止“重试风暴”不加控制的即时重试在并发场景下就是灾难。多个节点如果都失败一起即时重试瞬间就能把 LLM API 的限流阈值冲垮然后引发更多失败再触发更多重试形成恶性循环。我采用的策略是指数退避重试。第一次重试前等待 2 秒第二次等待 4 秒第三次等待 8 秒每次翻倍同时设置上限比如最大等待 60 秒超过就不再等待而是直接报错。加上一个随机抖动jitter防止同一批节点在同一时刻发起重试。这样做的核心逻辑很简单如果调用一个接口失败下次重试的紧迫程度应该降低而不是升高。服务端大概率是过载了你越是疯狂重试它越恢复不过来。3.3 每个节点都必须做到“幂等”“幂等”这个词听起来很学术但落到实际就是一句话同一个节点用同一个输入跑一次和跑一百次效果是一样的。为什么这个原则对流水线很重要因为重试意味着同一段代码可能会被多次执行。如果节点在执行过程中有副作用——比如写数据库、发通知、扣费——就必须保证这些副作用不会因为重试而重复执行。一个实际场景我的流水线里有一个节点负责调用某内容生成服务的计费 API。如果这个节点超时了客户端其实不知道服务端到底有没有处理成功。直接重试可能导致二次扣费。解决办法是引入幂等令牌每次节点执行生成一个唯一的请求 ID服务端通过这个 ID 识别是否为重复请求是重复就直接返回上次的结果。如果你的流水线没有这样的设计重试机制越强大你给自己挖的坑就越深。3.4 实际配置参数分享重试策略的参考值我目前使用的默认重试策略是最大重试次数 3 次基础退避时间是 2 秒退避倍率是 2最大退避时间是 30 秒启用随机抖动。在这个参数下即使每分钟有几十个节点同时失败也不会把接口打爆。针对不同的节点类型我又做了部分定制LLM 调用类节点的最大重试次数调整到 5 次因为这类失误大概率是临时限流工具调用类节点保持 3 次因为外部工具服务的不确定性更高业务校验类节点干脆不允许重试因为校验逻辑是纯计算输入一样结果就一定一样重试没有任何意义。这个参数可以先抄作业跑一段时间再看日志调整。重点不是参数本身而是你清楚每个节点的失败类型和合理的重试次数。4. 可续跑机制的核心玩法如果说可重试解决的是“单点失败之后怎么办”那可续跑解决的就是“链路跑一半挂了能不能别从头再来”。续跑的价值在长耗时任务里体现得尤其明显。4.1 检查点持久化续跑的物理基础续跑的核心是检查点。每完成一个节点就把整个节点链的当前状态、所有已完成节点的输出、还在排队节点所需的前置数据全部打包保存下来。这个过程我通常放在节点状态从 RUNNING 迁移到 SUCCEEDED 之后和写入日志同步进行。保存的粒度和格式要确保即便你跑到第 10 个节点时第 3 个节点的输出数据丢了也能通过检查点恢复到第 10 个节点执行前的状态重新加载第 3 个节点的快照数据作为上下文。检查点的存储我用过本地文件系统和对象存储两种方案。小规模单机跑本地文件就够了一个任务目录、几个 JSON 文件、一个状态索引逻辑简单透明度高。上到分布式多机跑就换成对象存储加一层分布式锁保证并发安全性。建议先本地跑通再考虑分布式否则排查问题的复杂度瞬间倍增。4.2 从断点续跑的恢复流程续跑不像重新启动那么简单它需要回答三个问题当前跑到哪了这个节点的输入数据还在吗后续节点的上下文是不是完整的我的恢复流程是首先从状态库里读取当前任务的状态索引找到最后一个 SUCCEEDED 的节点序号加 1 就是需要重新执行的起点节点。然后检查该节点的输入快照是否存在且完整不完整就回溯找上游最近一个完整快照从这个节点重新执行。最后把已经成功节点的输出数据按原有结构加载到内存的数据注册表里后续节点像正常流程一样从注册表里取数据。这三步走完流水线就能无缝衔接。整个过程对用户来说感知不到看起来就像是任务“停下来休息了一会儿又接着干活了”。4.3 上下文重建让续跑后的节点“不迷路”续跑最容易翻车的地方不在机制而在上下文。你从第 6 个节点开始重新执行但第 6 个节点的 Agent 需要知道前面 5 个节点处理后的结果如果它看到的上下文是支离破碎的输出质量必定大打折扣。我采用的方案是为每个节点注入一份“上下文摘要文件”。这份文件在节点执行前生成内容包含两个部分一部分是前面所有成功节点的输出摘要另一部分是当前节点的任务背景说明。这就等于给每个节点的 Agent 配了一张“任务地图”它不需要知道整条链路的细节也知道自己手里这份活儿在整个任务里的相对位置。实际测试下来的效果很明显。加了上下文摘要之后流水线中后段节点的输出质量明显比裸跑要稳定因为 Agent 的推理起点是扎实的。4.4 续跑时点的选择不是任何时候都能续有一点必须提醒续跑的适用场景是有边界的。它不是万能工具在设计节点链路的时候就要区分哪些节点的操作是可逆的、哪些是不可逆的。比如一个节点已经调用了第三方发送确认邮件的接口且发送成功这个动作就是不可逆的。任务在这里失败之后续跑时不能简单地把这个节点标记为未执行然后重新跑一遍否则用户会收到两封确认邮件。我的处理方式是给每个节点的输出快照里加一个“副作用是否已完成”的标记。节点执行成功后立即标记续跑时如果发现标记为真就直接跳过该节点的业务执行只把输出数据传递给下游。这样既保证续跑能继续又避免了重复副作用。5. 实操演示一个真实的 Pipeline 节点链 Demo理论讲了这么多来点真正能跑的。我按照上面这套设计做了一个文档智能处理流水线的简化版 Demo演示串行节点的创建、执行、重试和续跑。5.1 Demo 场景与节点划分模拟任务给一份本地 Markdown 文档自动生成结构化摘要、提取关键指标、翻译成英文摘要然后执行质量校验。整个任务拆成了四个节点节点 01文档解析与预处理节点 02结构化摘要生成调用 LLM节点 03关键指标提取与校验调用 LLM 规则校验节点 04英文翻译审核调用 LLM每个节点的输入就是上一个节点的输出输出都写成标准 JSON。5.2 核心代码实现直接贴核心实现代码主要集中在节点基类、流水线编排器、重试与续跑这几个部分。节点基类定义from abc import ABC, abstractmethod import json from pathlib import Path import time class BaseNode(ABC): def __init__(self, node_id: str, input_key: str, output_key: str): self.node_id node_id self.input_key input_key self.output_key output_key abstractmethod def execute(self, input_data: dict) - dict: 子类实现具体的业务逻辑 pass def run(self, context: dict) - dict: # 从上下文数据注册表读取输入 if self.input_key not in context: raise KeyError(fNode {self.node_id}: missing input key {self.input_key}) input_data context[self.input_key] # 执行节点逻辑 output_data self.execute(input_data) # 校验输出是否为合法 JSON if not isinstance(output_data, dict): raise ValueError(fNode {self.node_id}: output must be dict) # 把输出写入上下文注册表 context[self.output_key] output_data return output_data流水线编排器class Pipeline: def __init__(self, task_id: str, checkpoint_dir: Path): self.task_id task_id self.checkpoint_dir checkpoint_dir self.checkpoint_dir.mkdir(parentsTrue, exist_okTrue) self.nodes [] self.context {} self.status {} def add_node(self, node: BaseNode): self.nodes.append(node) self.status[node.node_id] PENDING def _save_checkpoint(self, node_id: str): checkpoint { task_id: self.task_id, current_node: node_id, context: self.context, node_status: self.status } checkpoint_path self.checkpoint_dir / f{node_id}.json checkpoint_path.write_text( json.dumps(checkpoint, ensure_asciiFalse, indent2) ) def _load_checkpoint(self) - str: # 找到最后一个成功节点的快照 checkpoints sorted(self.checkpoint_dir.glob(*.json)) if not checkpoints: return None last_cp json.loads(checkpoints[-1].read_text()) return last_cp[current_node] def run_from_checkpoint(self): last_node self._load_checkpoint() start_idx 0 if last_node: start_idx self._find_node_index(last_node) 1 for idx in range(start_idx, len(self.nodes)): node self.nodes[idx] self.status[node.node_id] RUNNING success False retry_count 0 max_retries 3 base_delay 2 while not success and retry_count max_retries: try: node.run(self.context) success True except Exception as e: retry_count 1 if retry_count max_retries: self.status[node.node_id] FAILED_TERMINAL raise self.status[node.node_id] FAILED_RETRYABLE delay base_delay * (2 ** retry_count) time.sleep(delay) self.status[node.node_id] SUCCEEDED self._save_checkpoint(node.node_id) def _find_node_index(self, node_id: str) - int: for idx, node in enumerate(self.nodes): if node.node_id node_id: return idx return None具体节点实现class DocParsingNode(BaseNode): def __init__(self): super().__init__(node_01, raw_doc, parsed_doc) def execute(self, input_data: dict) - dict: content input_data[content] paragraphs [p for p in content.split(\n\n) if p.strip()] return {paragraphs: paragraphs, total_paragraphs: len(paragraphs)} class SummaryNode(BaseNode): def __init__(self): super().__init__(node_02, parsed_doc, summary) def execute(self, input_data: dict) - dict: # 这里实际调用 LLMDemo 里直接用本地逻辑模拟 paragraphs input_data[paragraphs] summary f共 {len(paragraphs)} 段首段主要内容{paragraphs[0][:50]} return {summary: summary}调用方式checkpoint_dir Path(./checkpoints) pipeline Pipeline(task_demo_001, checkpoint_dir) pipeline.add_node(DocParsingNode()) pipeline.add_node(SummaryNode()) source_doc {content: 第一段内容。\n\n第二段内容。\n\n第三段内容。} pipeline.context[raw_doc] source_doc pipeline.run_from_checkpoint()5.3 断点续跑的实际效果我在 Demo 里人为模拟了一次故障让 SummaryNode 第一次执行时抛异常重试两次后再成功。整个流水线的输出结果是正确的日志里清楚记录了两个失败重试的时间点和原因。更关键的是续跑效果我单独跑完了前两个节点并生成了检查点然后修改了 SummaryNode 的实现逻辑重新执行了流水线。结果流水线只从 node_02 开始运行node_01 的结果直接通过快照加载到 context没有重新解析文档。对于长篇文档或耗时任务这个收益非常直接。续跑的并联效果也很直观。如果你有 100 个任务都要跑同一个六节点流水线其中 30 个在第 4 个节点失败修复之后只需要重跑第 4 个节点及后续节点而不是整个链路重新执行。这省下的不只是时间还有 LLM 的调用成本。5.4 实现过程中的三个隐性细节第一个隐形细节是快照文件名的排序问题。文件系统里node_10.json会排在node_2.json前面如果节点数量超过 9 个字符串排序会导致恢复顺序错误。我的解决方法是节点 ID 用零填充格式化node_02、node_10或者直接按照时间戳而不是节点序号来选择最后一个检查点。第二个细节是上下文资源释放。如果你的节点链路很长而每个节点的输出都保留在 context 里内存开销会持续增长。我在真实场景里会对 context 做内存回收只保留最近两个节点的输出更早的快照数据用完即弃需要时再从检查点加载。第三个细节是并发场景下的检查点写入冲突。如果同一个任务被意外并发触发两次两个进程同时写检查点文件就会互相覆盖产生脏数据。我在检查点写入时采用先写临时文件写入成功后用原子操作替换正式文件的方式保证任何时刻读取到的文件都是完整的。6. 常见问题与排查技巧实录这部分内容基本都是我从真实实践里摸爬滚打总结出来的经验希望对你有用。6.1 节点重试风暴引发的雪崩有一段时间我流水线跑得特别频繁然后发现 LLM API 经常报 429 限流错误。一开始以为只是调用量太大后来仔细查日志才发现部分失败的节点触发了重试重试又触发更多节点失败形成级联效应。排查思路是为每个节点记录失败原因和重试次数然后按“失败类型 节点类型”聚合分析。一次日志分析就能看出大部分 429 错误集中在相同两三个节点上原因是这两个节点的 Prompt 引导模型输出了格式不稳定的数据导致下游节点不断校验失败。解决方案有两个方向一是优化这两个节点的 Prompt 和输出格式约束从源头减少失败二是把重试上限从 3 次降低到 1 次并且对下游节点增加输入格式预校验格式不对直接丢弃而不触发重试。6.2 Resumed 之后出现上下文错位续跑功能刚做出来的时候有一个很诡异的问题整个流程跑完结果看起来也正常但仔细对比发现第 5 个节点的部分输出逻辑引用的是旧数据。查了很久才定位到原因续跑时加载的检查点里context 是整个任务所有节点的输出全量快照。但我的某个节点实现里用了一个内部的全局缓存变量这个变量存了上一次执行时的残留数据。续跑时节点代码拿到的输入对了但内部缓存没有清空导致用了旧数据。修复方式很直接所有节点类在run()方法最开始处强制清空节点内部状态确保每次执行都是干净的环境。6.3 中间节点输出量过大把存储打爆我设计过一段时间比较密集的向量化摘要节点这些节点会把上游文本向量化后的结果全部存进快照文件。结果跑完任务后一看一个任务的检查点就占了几个 GB 的存储。解决方案是分级存储策略原始数据存冷存储中间阶段的临时数据定期清理保留最近两个节点的快照作为热数据其余归档到冷存储。同时给每个节点的输出额外配置一条“最大体积限制”超过限制就截断并记录告警由人工判断是否影响后续节点的执行。6.4 重试成功但返回了重复副作用这个问题前面提到了幂等设计这里讲一个实例。我有一次给某个节点配置了一个 webhook 通知动作表示“节点执行成功”。因为网络超时触发了这个节点的重试第二次执行成功了但第一次执行其实已经在服务端生效了结果通知重复发送了三次。排查后发现我当时只对 LLM 调用部分做了重试保护忽略了通知发送这个副作用操作。修复方式是给通知动作引入请求唯一 ID服务端根据 ID 去重同时把通知动作从节点的业务逻辑里拆出来放到一个专门的后置处理器里只有节点真正标记为 SUCCEEDED 之后才会触发。这系列问题的共性教训是重试和续跑机制必须在设计节点的时候就考虑到并发安全、幂等性和副作用控制。这些不是事后能补完善的东西最好前置到架构设计阶段。如果你准备把系统切到这种多 Agent 串行流水线的架构上我建议最开始不要把所有节点都做成“可续跑”先选一条核心链路跑通把检查点、重试策略、状态机这些基础能力验证稳定了再逐步扩展。我自己就是从单条链路开始踩完所有坑才把这套方法论复制到其他应用场景里的。有一点想特别强调串行流水线不意味着所有任务都必须串行。有些步骤之间没有强依赖该并行就并行。但如果你把任务拆成了严格的线性流水线那“可重试、可续跑”就是这条链必须具备的生命力。一个不支持断点恢复的流水线本质上跟一个更复杂的单 Agent 没有太大区别反而徒增了系统复杂度。
阅读完成 · 觉得有帮助？
