做多Agent应用最头疼的问题不是把每个Agent本身写好而是多个Agent串起来之后任务跑到一半卡住了。我今天想聊的“多Agent串行流水线”本质上就是把这个复杂任务拆成一条Pipeline节点链每个节点由一个Agent专职处理节点之间用共享上下文传递结果整条链支持单节点重试、断点续跑。我第一回把这套东西落地是因为一个多步骤数据处理任务总在最后一步超时每次从头重跑要花二十分钟改成流水线之后哪一步挂了就重跑哪一步进程重启也能从失败节点继续往下走整体耗时基本被压到原来的三分之一。这篇文章会从设计思路、状态管理、重试与续跑实现一路讲到完整的可落地Demo适合已经在用大模型接口做应用、但被任务编排搞得头疼的开发者也适合刚接触Agent开发、想找一套稳定执行框架的朋友。1. 为什么把任务拆成多Agent串行流水线1.1 单次大Prompt的三大坑先聊聊我为什么不直接用一个大Prompt把所有事情都交给模型做。单个Agent处理复杂任务最明显的三个问题第一结果不稳定。任务一旦超过两三个子步骤模型很容易在中间逻辑上出错。比如你先让它做实体识别再做情感判断它可能把两步的动作搅在一起输出一个不伦不类的混合结果。你反复调Prompt能好一阵子但换一批数据又翻车这种随机性非常消耗耐心。第二失败成本高。一次请求从发出到返回可能要好几分钟如果中间某一步格式错了整次输出作废重新提交时又要完整跑一遍。任务越重重跑的时间和经济成本就越不可控。我遇到过一次批处理任务在倒数第二步格式错误重跑时又碰上限流前后浪费了小半天。第三不透明。你只能看到最终输出完全不知道任务到底卡在哪一步。日志里只有一次Prompt和一次Response出了问题只能靠猜猜来猜去最后还是得开打印一个个试。当时我做一个模拟项目X输入是大批量的文本片段需要先做实体识别再做情感判断最后拼装成结构化表格。一次性Prompt跑得又慢又烂后来拆成三个独立Agent反而稳定很多。这件事让我意识到任务拆解本身就是一种可靠性设计拆得合理后续的所有故障处理都会简单一个量级。1.2 串行流水线解决了什么串行流水线的思路很直接把任务拆成有先后依赖的多个步骤每个步骤对应一个Agent前一个节点输出结构化结果后一个节点只消费它需要的那部分数据。节点之间靠一份共享的TaskContext传数据而不是把整个对话历史滚动塞给下一个节点。这样做的好处非常实际定位问题精准哪一步失败看节点状态就知道不用再从最终结果反推。单点重试成本低失败节点单独重跑已成功的节点一点都不动。可续跑进程崩了也没关系只要把节点状态和结果落盘下次启动直接从失败节点继续走。上下文可控每个Agent只看到它需要的那一小块数据Prompt长度稳定模型不容易“迷失”在无关信息里。这里得先记住一句话多Agent的价值不是“拆得越多越好”而是“每个Agent只做最小必要粒度的事同时链路状态能被完整记录和恢复”。串行只是当前最实用的起步方式但它已经把一个LLM应用里的可靠性地基打好了。之后哪怕你想演变成并行或动态调度这个状态管理的思路也一样能复用。2. 流水线引擎的整体设计与核心抽象2.1 三个核心角色上下文、节点、引擎动手写代码前先把抽象定下来。我的流水线里只有三个核心角色TaskContext贯穿整个任务链的上下文对象记录任务ID、原始输入、每个节点的输出、节点状态和重试次数。PipelineNode一个可执行的Agent节点只暴露execute(ctx)接口内部可以是模型调用、规则处理或外部服务请求。PipelineEngine负责编排节点顺序、执行重试、保存和恢复状态。一开始别急着上复杂框架把这三个东西定义清楚就够了。用Python的dataclass实现TaskContext非常顺手from dataclasses import dataclass, field, asdict from typing import Dict, Any, Optional import json import time import os dataclass class TaskContext: task_id: str payload: Dict[str, Any] results: Dict[str, Any] field(default_factorydict) node_status: Dict[str, str] field(default_factorydict) retry_counts: Dict[str, int] field(default_factorydict) current_node: str status: str pending created_at: float field(default_factorytime.time) updated_at: float field(default_factorytime.time)字段为什么要这么设计node_status里存每个节点的当前状态是为了让引擎在重启后能快速判断“哪个节点还没成功”。retry_counts单独记录重试次数方便排查时看到底是哪个节点在反复失败而不是和节点输出混在一起。results存的是节点已完成输出续跑时成功节点不必重新调用模型直接从results里取就行。节点定义更简单它只做一层很薄的封装class PipelineNode: def __init__(self, name: str, agent_func): self.name name self.agent_func agent_func def execute(self, ctx: TaskContext): return self.agent_func(ctx)这里的agent_func就是真正会调用大模型接口的函数。你也可以把它包成一个类方法内部处理Prompt拼接、输出解析和结果校验。重点是节点只接收TaskContext不直接访问数据库或上游服务所有依赖都通过Context传递这样引擎才能统一管理生命周期也才能把重试和续跑做成通用能力。2.2 为什么先做串行而不是并行或DAG很多人会问任务都拆成节点了为什么不直接并行或者做成有向无环图DAG我的回答是先做串行因为串行把可靠性问题压缩到了最简单。并行意味着多个节点同时执行那你就要处理并发写Context、部分失败回滚、结果合并顺序等等一堆问题DAG则要设计依赖解析和拓扑排序。对大多数LLM应用来说任务天然是分步的先抽取信息再清洗再格式化每个步骤都强依赖前一步的输出。这时候串行不仅够用还是最容易调试的。串行还有一个隐藏优势检查点天然清晰。每个节点跑完就是一个稳定的中间快照。如果是并行A节点成功了、B节点失败了快照里会出现“半边成功”的尴尬状态续跑时还得判断A的结果到底能不能复用。串行完全没这个问题整条链的状态要么往前一步要么停在原地。串行的短板也很明显总耗时等于所有节点耗时之和。如果某一步模型响应要30秒十个节点就是五分钟。我的建议是先把正确性和可恢复性跑通再按热点节点做局部并行或批处理不要一上来就追求架构上的“酷炫”。2.3 节点之间传什么不传什么这是最容易翻车的地方。一开始我喜欢把整个Context JSON塞给下一个Agent让它“根据上下文自由发挥”结果模型被大量无关字段干扰输出质量迅速下降调用成本也跟着涨。现在我的原则是显式传输最小必要数据。每个节点的输入Schema单独定义只包含上一步输出里它真正需要的字段。比如实体识别节点只需要文本字段不需要上一步的情感分数。如果硬塞模型很可能把情感词也当成实体或者在判断时被无关字段带偏。这个原则看起来很朴素但真正影响的是稳定性和成本越到后期越明显。同样的道理节点返回给Context的结果也应该是结构化dict别让Agent自由发挥输出一整段话。强烈建议为每个节点定义一个输出Schema执行完立刻做合法性校验。校验不通过就直接抛RetryableError让重试机制兜底而不是等到下一个节点解析时才报“字段缺失”。字段缺失这个错在流水线里会一路传导最后变成一个让人摸不着头脑的深层异常。3. 重试机制让每个节点都具备抗失败能力3.1 节点状态机的生命周期每个节点会经历四种状态pending、running、success、failed。加上因为续跑被跳过的节点我用skipped表示。具体含义如下pending还没开始。running正在执行如果此时进程崩溃恢复时可以继续处理。success执行成功结果已经持久化。failed超过最大重试次数流水线在这里停止。引擎每次执行节点前先把状态置为running并落盘执行成功后再置为success并落盘。这个“先标注再执行”的顺序非常重要它保证了一个节点至少在语义上不会出现“未开始却已有结果”的矛盾。换句话说看到running它可能是执行中也可能已经崩溃需要继续看到success它一定已经有可用结果。实际代码里状态更新和结果写入最好放成同一个原子操作。用本地文件存储时我习惯先写临时文件再rename避免进程中途崩溃留下半份坏JSON如果业务量再大一点可以换成一个支持事务的小型数据库把整份Context作为一行记录更新的原子性会更强。状态写入的时间点越早越安全千万不要等Agent执行完再一次性写那样崩溃恢复时根本不知道节点跑到哪里。3.2 重试策略与退避算法重试不是简单地循环调用。大模型接口最常见的错误有两类一类是限流、超时属于可重试错误另一类是Prompt格式错误、参数非法属于不可重试错误。可重试错误我单独建立一个异常类型和普通业务异常区分开class RetryableError(Exception): pass class FatalError(Exception): pass执行节点时把Agent函数里的异常分类捕获。如果是RetryableError按指数退避重试如果是FatalError直接中断不再浪费请求。分类这件事看起来小但能省一大笔成本不然遇到参数错误还会反复撞墙白白消耗额度。重试参数我一般这样设置max_retries3base_delay2秒指数退避。也就是第一次失败后等2秒第二次等4秒第三次等8秒超过3次就不再自动继续。具体数值要根据上游服务的限流窗口调整如果窗口是60秒退避基数可能要调到10秒甚至更长。另外我还会加一点随机抖动避免多个任务同时失败后一起重试结果把限流打得更严重。下面是带重试的节点执行函数def run_node_with_retry(node: PipelineNode, ctx: TaskContext, max_retries: int 3, base_delay: float 2.0): for attempt in range(max_retries): try: result node.execute(ctx) except RetryableError as e: if attempt max_retries - 1: ctx.node_status[node.name] failed save_context(ctx) raise delay base_delay * (2 ** attempt) ctx.retry_counts[node.name] attempt 1 save_context(ctx) time.sleep(delay) continue except FatalError: ctx.node_status[node.name] failed save_context(ctx) raise ctx.results[node.name] result ctx.node_status[node.name] success save_context(ctx) return result注意每次重试前都要save_context把retry_counts更新。这能让你事后从状态文件里看到“这个节点到底试了几次”排障时非常有用。保存状态这个动作看起来多点几次但在续跑场景里它就是唯一能依赖的“账本”。3.3 幂等控制别让重试变成重复执行重试机制最大的隐性坑是副作用。比如节点里做了“写入一条数据库记录”或“发送一封邮件”第一次请求其实已经成功但因为响应超时被判定为失败重试后又会执行一次造成重复写库、重复发送。这个问题处理不好重试机制反而变成数据污染的来源。解决办法是给每个节点执行加上请求唯一ID或者把操作设计成幂等的。更通用的做法是节点内部不要直接执行有副作用的操作而是先输出一个“动作描述”由引擎统一去执行执行前根据动作ID查一下是不是已经执行过。如果你只是想快速规避也可以把写操作放在流水线的最后一步并保证最后一步不做自动重试主流程失败后人工确认再补跑。我自己在模拟项目里是这样处理的所有外部写操作都收口到一个叫“输出节点”的Agent里这个节点不参与自动重试只做状态登记。虽然牺牲了一点自动化但至少不会产生脏数据。简单可靠比花里胡哨的分布式事务方案值钱得多。4. 续跑机制从断点恢复而不是从头再来4.1 续跑的前提全量状态落盘续跑不是魔法它的前提只有一个每次状态变化都要持久化。进程可以丢状态不能丢。所以save_context函数要保存完整的内容包括任务ID、原始输入、每个节点的输出、节点状态、重试次数以及当前正在处理的节点。用本地JSON文件就够起步阶段使用。文件名直接用task_id更新时整体覆盖但一定要采用原子写。我第一次实现时直接覆盖原文件结果有次进程被kill留下一个写完一半的JSON恢复逻辑彻底废掉。后来改成先写临时文件再替换代码是这样的def save_context(ctx: TaskContext, path: str tasks): os.makedirs(path, exist_okTrue) tmp_path f{path}/{ctx.task_id}.json.tmp final_path f{path}/{ctx.task_id}.json with open(tmp_path, w, encodingutf-8) as f: json.dump(asdict(ctx), f, ensure_asciiFalse, indent2) os.replace(tmp_path, final_path) def load_context(task_id: str, path: str tasks) - TaskContext: with open(f{path}/{task_id}.json, encodingutf-8) as f: data json.load(f) return TaskContext(**data)os.replace是原子操作比“直接覆盖原文件”安全得多。这里有个隐含要求TaskContext里所有字段都要能JSON序列化payload和results里别放bytes、datetime这类对象放进Context之前先转成基础类型。否则恢复时你会看到一堆诡异的反序列化异常而且异常已经脱离节点上下文很难定位。4.2 续跑主循环跳过已成功节点有了持久化状态续跑主循环就变得很自然加载Context找到第一个状态不是success的节点从它开始继续执行。前面的成功节点直接跳过因为它们的输出已经在results里。def run_pipeline(ctx: TaskContext, nodes: list[PipelineNode]): start_idx 0 if ctx.current_node: for i, node in enumerate(nodes): if node.name ctx.current_node: start_idx i break for node in nodes[start_idx:]: ctx.current_node node.name if ctx.node_status.get(node.name) success: continue ctx.node_status[node.name] running save_context(ctx) run_node_with_retry(node, ctx) ctx.current_node ctx.status done save_context(ctx)注意current_node的意义。它在失败时指向失败节点在续跑时告诉引擎“从哪个节点接着来”。当整个任务跑完后我把它清空并置status为done表示任务已经收尾。这套逻辑的简单之处在于你不需要记录“最后一个成功的节点”只需要记录“当前正在处理的节点”。“当前节点失败”和“当前节点未完成”在启动恢复时会指向同一个位置语义很干净。4.3 强制重跑与跳过修复默认续跑策略是“成功节点不重跑”但调试时你会有例外需求。比如某个节点虽然返回了success但业务校验发现结果质量很差你想单独重跑它或者你改了某个节点的Prompt希望该节点和它之后的所有节点重新执行再比如节点代码本身有Bug修复后想从该节点重放。这些情况都需要一个显式的强制重跑入口。所以我都会给引擎加一个mode参数resume是默认续跑rerun_from指定从某节点开始强制重跑该节点及之后的节点忽略已有success状态。实现时只需要在run_pipeline开头把目标节点及之后的node_status和results清掉再走同样的循环即可。实际实现里要按顺序标记让节点状态和结果保持同步避免出现results里少一个字段但状态却是success的脏状态。续跑能力的价值不只是机器崩了能恢复它其实给你提供了一套“定点重放”的调试工具。哪个节点不顺眼就只重跑哪个节点不需要整条链陪跑。做Agent应用调试这个能力几乎每天都会用到尤其是遇到模型输出不稳定的时候。5. 实操Demo三段式节点链跑通“重试续跑”5.1 定义三个节点用一个具体例子把前后逻辑串起来。模拟一个文本分析任务输入一段原始文本经过三步处理抽取节点从文本里抽标题和关键词。清洗节点把抽取结果里的空字段、格式问题修掉。格式化节点把结果整理成固定JSON结构。为了演示不依赖真实模型我用确定性函数模拟Agent并且让抽取节点第一次执行时故意抛一个RetryableError用来展示重试生效再让格式化节点支持一个修复开关用来展示续跑。def extraction_agent(ctx: TaskContext): raw ctx.payload[raw_text] if ctx.retry_counts.get(extract, 0) 0: raise RetryableError(模拟模型限流) return {title: raw.split(\n)[0], keywords: [流水线, Agent]} def cleaning_agent(ctx: TaskContext): extracted ctx.results[extract] title extracted[title].strip() keywords list(dict.fromkeys([kw.strip() for kw in extracted[keywords] if kw.strip()])) return {title: title, keywords: keywords} def formatting_agent(ctx: TaskContext): if not ctx.payload.get(format_repaired, False): raise FatalError(格式化节点遇到不可重试错误) cleaned ctx.results[clean] return {document: {title: cleaned[title], keywords: cleaned[keywords]}}抽取节点故意在第一次失败后重试成功正好触发重试机制。真实场景下这里就是“大模型偶发限流”的典型表现把它单独拎出来演示比看十页理论都直观。然后组装节点链nodes [ PipelineNode(extract, extraction_agent), PipelineNode(clean, cleaning_agent), PipelineNode(format, formatting_agent), ]节点命名要稳定我在前面说过这个坑所以这里直接用extract、clean、format这种不会轻易改动的名字。5.2 模拟一次完整执行有了节点链就可以构造TaskContext并执行主流程代码如下ctx TaskContext( task_iddemo-001, payload{raw_text: 多Agent流水线实践\n这是一段用于测试的文本。}, ) run_pipeline(ctx, nodes) print(ctx.results[format])正常情况输出{document: {title: 多Agent流水线实践, keywords: [流水线, Agent]}}同时tasks/demo-001.json里应该能看到三个节点都是successextract的retry_counts是1。这说明状态记录不只在进程内存里存在而是已经留给了下一次启动恢复。这时候你可以把这个文件当作一个完整的任务账本里面记录了每一步谁做过、做成了什么结果、失败了几次。5.3 模拟崩溃后续跑再演示续跑。假如格式化节点第一次执行失败了任务停止在format节点上。这时tasks/demo-001.json里extract和clean都是successformat是failed。我们手动做一次“修复操作”把payload里的format_repaired字段设为true然后重新加载Context调用run_pipeline看看extract和clean会不会被重跑。ctx load_context(demo-001) ctx.payload[format_repaired] True save_context(ctx) run_pipeline(ctx, nodes) print(ctx.results[format])这里的关键是extract和clean这两个节点的agent_func不会被调用。你可以自己在两个模拟函数里加print验证运行日志只会出现类似下面这样的内容[engine] start from node: format [engine] node format completed不会有extract和clean的执行记录。这就是续跑的核心体验前面成功的节点原样保留只从失败点继续推进整个过程不需要用户重新提交原始输入也不需要为已经做成功的步骤再次付费。这个Demo里没有引入真实模型调用但核心机制完全通用。你只需要把agent_func换成真正的模型请求函数并让它按分类抛出我们定义的异常重试和续跑就会按同样的路径工作。6. 常见问题与排查技巧实录6.1 问题速查表这里整理了一份我实际使用中频率最高的问题对照表排查时可以直接对着看现象可能原因处理方式下游节点拿不到上游字段Agent输出格式不稳定给每个节点配置输出Schema执行后立即校验每个节点Prompt越来越长把完整Context塞给模型只传当前节点需要的字段Context只做存储重试后出现重复数据节点内有写库、发消息等副作用副作用收口到最终节点或使用请求唯一ID去重进程崩溃后恢复读不了状态直接覆盖写JSON文件改成先写临时文件再os.replace节点明明成功却要重跑结果缓存没复用检查run_pipeline的success跳过逻辑确认节点命名稳定无限重试拖死服务未区分可重试和不可重试错误异常分类FatalError不重试这张表不是我临时编的是踩完坑以后抄在便签上的。你看一遍基本能知道哪些问题值得提前预防尤其“结果缓存没复用”这一点特别容易在流水线里被忽略。6.2 我踩过的几个坑第一个坑是节点名不稳定。一开始我用“1_抽取”这样的字符串当节点名后来想调整执行顺序改成“node_1_extract”结果续跑时current_node对不上后面所有节点状态全部作废。从那以后节点名定义好就不再改动顺序调整只靠列表顺序不靠名字。这个约定看起来简单但能避免一次非常难查的状态错乱问题。第二个坑是把重试逻辑放在Agent内部。很多Demo里的Agent自己会重试导致引擎完全不知道失败过日志里也看不到重试次数。后来我把重试统一收归引擎让Agent只负责单次执行并抛出分类异常。这样重试次数才可靠续跑时也才能从真正失败的位置接上。如果重试被隐藏在Agent深处你会看到整条链几乎不会失败但成本却莫名其妙高出一大截。第三个坑是保存Context时序列化失败。结果dict里混进了模型返回的非JSON对象一执行save_context就崩。我的解决办法是定义一个统一的输出数据结构Agent返回前先做一次强制转换保证一切可序列化。不要偷懒把转换丢给引擎引擎处理时错误信息早就脱离了节点上下文排查起来非常费劲。6.3 给新手的配置建议如果你准备照着这套思路做一个最小可用版本我建议从这些配置开始节点数量先控制在3到5个不要一开始就拆十几个max_retries先设为2跑通后再调大文件存储路径放在临时目录避免污染正式工程每个Agent的执行超时时间必须设置没有超时的LLM调用比显式失败更可怕。我的经验是先不要追求全自动而是把“重试续跑”这条主链路在人工可控的范围内跑出节奏感再逐步放开。稳定性是靠监控和异常处理堆出来的不是靠把代码写得看起来无所不能堆出来的。你先把状态文件打开看几次真正理解了“每个节点什么时候写状态、什么时候写结果”这套流水线就算入门了。最后再分享一个我用着很顺手的小技巧在每个节点执行前往Context里塞一条begin_time执行完后塞一条end_time。别小看这个时间戳续跑发生时你把整份Context打出来就能看到每个节点实际跑了多久、重试间隔多久。定位慢节点或者排查限流问题这个信息比什么日志都直观。我现在的所有Pipeline节点都默认带上这个字段算是个人习惯推荐你也试试。