把一个大任务直接丢给单个 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_factorydict) output: Dict[str, Any] field(default_factorydict) 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_factorydict)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, encodingutf-8) as f: data json.load(f) return RunState(**data) def save_state(state: RunState, base_dir: str checkpoints): os.makedirs(base_dir, exist_okTrue) path os.path.join(base_dir, f{state.run_id}.json) tmp_path path .tmp with open(tmp_path, w, encodingutf-8) as f: json.dump(state.__dict__, f, ensure_asciiFalse, indent2) os.replace(tmp_path, path) def run_node(node: NodeConfig, node_input: dict) - NodeResult: result NodeResult(inputnode_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_idrun_id, pipeline_namearticle_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_id8f3a9c2e statuscompleted中间如果进程意外退出我只需要重新执行一次 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”。最后再分享一个小技巧我在做检查点时会把当前节点的提示词版本也一并存进去。这样如果后续改了提示词导致结果变差直接回滚检查点对应的提示词版本就能快速定位问题。这个小习惯在长流水线迭代时真的能救命。