多 Agent 系统做多了你会发现一个很反直觉的现象并行编排看起来很美真正跑起来却经常让你想砸键盘。大家一提到多 Agent第一反应就是“让好几个 Agent 一起干活”好像越并行就越高级。但我在实际项目中踩过几次坑之后反而越来越倾向于一种被很多人忽视的编排方式——多 Agent 串行流水线。核心思路很简单把一个大任务拆成一串 Pipeline 节点链每个节点由独立 Agent 负责前一个节点的输出作为后一个节点的输入节点之间严格按顺序执行而且每个节点都支持可重试、可续跑。这篇文章不是讲概念而是记录我如何把一个真实的复杂任务拆成一条可靠的节点链以及中途遇到的各种奇怪问题。如果你正在做 Agent 编排、工作流设计或者发现自己写的多 Agent 代码一旦跑起来就失控那这篇内容值得你花十分钟看看。1. 为什么需要串行流水线单 Agent 的边界与多 Agent 协作的痛点1.1 单 Agent 处理复杂任务时的三个典型困境先聊一个基础问题一个 Agent 到底能处理多复杂的任务我在很长一段时间里习惯把任务一股脑丢给一个大模型 Agent让它自己规划、自己执行、自己检查。结果是简单任务很好用复杂任务经常翻车。举一个很常见的业务例子让一个 Agent 完成“从一篇文章里提取数据、清洗数据、生成报表、用自然语言总结结论”的完整流程。单 Agent 模式是这样运行的Prompt 里塞满所有要求模型自己决定先做什么后做什么。看起来聪明实际运行中你会发现三个典型问题。第一个问题是上下文污染。Agent 一旦执行到后期步骤它的上下文里已经塞满了前面的中间结果、临时输出、甚至失败重试的报错信息。到了最后生成结论时它可能被前期的脏数据或无关细节干扰输出的结论偏离原始事实。我遇到过一个实例前期提取出的数字有异常值模型在总结时不但没有识别反而把异常值也当作正常现象写进了结论因为它已经分不清哪些是待处理数据、哪些是已处理数据。第二个问题是不可局部纠正。单 Agent 任务中如果最后发现报表格式不对你只能让整个任务从头再来。前面已经跑通的提取、清洗逻辑也会跟着重新执行一遍浪费时间和令牌。更糟糕的是大模型本身有随机性重新跑一遍的结果可能与第一次不完全一致越迭代越不稳定。第三个问题是不可观测。单 Agent 内部到底先做了什么、后做了什么你很难从外部判断它的运行路径。出了问题想排查只能翻一整段对话记录定位成本很高。一个 Agent 内部十几个步骤混在一起你根本不知道是哪一步造成了结果偏差。所以当你发现“一个 Agent 干到底”的思路撑不住复杂业务时就该考虑把任务拆开。拆开不一定是并行走串行反而是最稳的起点。1.2 多 Agent 并行协作的另一面死锁、资源争抢与不可观测并行编排的诱惑力很强任务拆成几个独立子任务Agent 同时跑理论上总耗时能大幅缩短。我在一个模拟项目里也试过并行编排结果并不像想象中那么美好。先碰到的是共享资源争抢。多个 Agent 同时需要访问同一个数据源或第三方服务时很容易触发接口限流。我做过一个实验三个 Agent 同时抓取同一批网页目标站点的反爬策略立刻生效三个 Agent 全部失败。如果换成串行同一个站点在时间上错开的访问反而不会触发限流。后半段我们通过某个开源网关限流测试并行请求的失败率一度达到 40%串行方式则几乎不受影响。其次是结果合并的一致性问题。你设计了几个并行分支每个分支输出一份结果片段最终需要一个聚合 Agent 来合并。但各个分支的结果之间可能存在冲突、重复或遗漏。聚合 Agent 的 Prompt 很难写得完美它面对的是几个不同来源、不同格式、不同侧重点的信息块合并结果经常出现内容重叠和逻辑断裂。协作过程中各个分支之间没有中间信息交换往往要到聚合阶段才能发现彼此的假设不一致。还有一点很少人提并行编排排错成本高。多个 Agent 同时失败日志交织在一起你很难区分一个异常到底是由哪个分支的哪一步引起的。有些框架会打出一大堆并发 Trace但实际排查时你多半是在对着时间戳猜。我自己后来的选择是核心链路用串行只有确定互不依赖、各自独立的子任务才敢并行。对于绝大多数业务流程串行的确定性远比并行带来的那点加速更值钱。1.3 串行流水线为什么是“最朴素但最可靠”的编排方式串行流水线的理念其实非常古老跟工厂里的装配线一个逻辑一个工位只做一道工序做完交付给下一个工位。它不追求每个工位的速度而是追求整条线稳定、可预测地运转。把它映射到多 Agent 系统里就是一个 Pipeline 节点链。每个节点完成一个边界清晰、动作明确的子任务节点之间通过标准数据结构传递结果。这样做的好处很直接每一步的输入输出都明确出问题能立刻定位到具体节点。节点是天然的检查点只要每个节点都做了结果记录整个任务就可以随时从某一步续跑。单个节点失败后可以单独重试不影响前面已经完成的工作。每个节点的 Prompt 可以针对单个子任务做极致优化不需要一个巨型 Prompt 兼顾所有步骤。我并不是说串行优于并行而是说串行拥有“准入门槛最低、心智负担最小、可控性最强”这三个优势。你甚至不需要引入任何复杂编排框架用 Python 里的一段顺序调用代码就能实现一个最小可用版本。等跑通了再慢慢往里面加重试、续跑、状态持久化这些工程能力可靠性就建立起来了。复杂系统从来不是一开始就复杂而是在简单骨架之上逐步加固的。2. 流水线节点设计任务拆分的粒度与节点职责边界2.1 节点粒度的判断标准可独立验证、可重试、可续跑节点拆分是整个流水线设计的灵魂。拆得太大跟单 Agent 直接干没有区别拆得太小节点之间的衔接开销会让整个流水线变得笨重每一次节点跳转都要解析、传递、验证数据反而降低了效率。我常用的判断标准有三个。第一个是可独立验证一个节点完成后你是否能通过某种校验规则判断它的输出是否合格如果输出是一份数据表能不能用脚本检查行数、字段完整性、数值范围如果输出是自然语言文本有没有一套规则或二次确认机制判断它是否符合要求能验证才能作为独立节点存在。第二个是可重试如果这个节点失败重新执行它是否安全它的执行过程会不会产生无法撤销的外部副作用比如发送邮件这个动作天然就不应该成为一个普通节点因为重试可能导致收件人收到两封一模一样的邮件。而一个纯计算、纯转换、纯生成文本的节点重试是安全的因为它只消费输入、产生输出不触碰外部系统。第三个是支持续跑节点内部是否有明确的完成点一个节点要么产生最终输出并进入已完成状态要么中途失败进入待重试状态不存在悬挂的中间态。如果你拆出来的一个 Agent 内部还有长期循环或隐含的等待那说明它还不够细需要继续拆。对于很多业务任务节点数量在四个到八个之间是比较舒服的区间。太少了说明拆分不彻底太多了说明你把一个自然的 Agent 环节强行切碎了。判断标准只有一个每个节点都能被一个独立的 Prompt 清晰描述并且有独立的输入、输出和验收方式。2.2 节点的标准接口约定输入、输出、状态、异常节点之间靠什么连接不是靠 Prompt 里的模糊指引而是靠一套严格的接口约定。我在流水线里定义节点时每个节点至少具备四个要素输入 Schema、输出 Schema、运行状态和异常信息。输入输出 Schema 是整个 Pipeline 的地基。我强烈建议用结构化格式约束 Agent 的输出而不是让它返回一坨自由文本。实际项目中我们常用 JSON 作为节点的数据交换格式具体设计如下{ node_id: extract_entities, run_id: e2f8a91c-3d94-4b0a-bcfe-83a1a44f2d3a, input: { source_text: ……待处理的原文……, params: { max_entities: 50 } }, output: { entities: [ { name: 某公司, type: org, confidence: 0.93 } ] }, status: completed, attempts: 1, error: null, created_at: 2025-01-12T10:30:00Z, updated_at: 2025-01-12T10:32:15Z }这里的关键是每个节点的运行记录都是独立的包含自己的输入快照、输出快照和状态信息。当流水线执行到第 N 个节点时前面 N-1 个节点的输出记录都已经持久化这就为实现续跑打下了基础。状态字段我一般只允许四种值pending、running、completed、failed。不再额外引入什么“部分完成”“已跳过”的中间状态因为它们会显著增加编排器的判断复杂度。如果一个节点被前序节点跳过那它的状态就被直接标记为completed但输出里带一个skipped: true标记这样既不阻断主流程也能记录真实执行情况。异常信息字段也不仅是字符串。我更建议保存结构化的异常对象包括错误类型、错误消息、可选的堆栈信息、重试策略建议。未来做诊断时能直接看到“这是超时异常建议提高超时上限”而不是一条模糊的“调用失败”。2.3 拆分案例一个实际业务的 Pipeline 节点示例只讲抽象概念不够我拿一个实际做过的模拟项目来说明拆分过程。假设任务是给定一批 PDF 类型的合同文件自动提取关键条款、做合规检查、生成摘要、输出结构化报告。如果交给单 Agent它能跑但效果不稳定。我最终拆成了六个串行节点节点职责输入输出可验证性1. 文档解析PDF 转文本、按页切分PDF 文件路径原始文本块列表检查文本块数、非空比例2. 信息抽取抽取合同方、金额、日期、期限文本块列表结构化字段 JSON字段存在性、类型校验3. 数据校验检查必填字段、金额一致性结构化字段校验结果、修复建议规则引擎自动校验4. 合规检查判断条款是否符合给定规范校验后的字段合规性结论列表关键词规则 二次评审5. 摘要生成基于字段和结论生成摘要字段与结论摘要文本长度限制、关键词覆盖率6. 报告组装汇总所有结果生成最终文档以上全部输出结构化报告模板字段完整性校验这个拆法有几个明显好处。第一每个节点可以独立测试比如“信息抽取”节点我可以准备 50 份合同原文反复调 Prompt 和模型参数直到准确率达标不会影响其他节点。第二某个节点失败后我可以只重跑这个节点前面解析结果、抽取结果都还在不需要重新花钱重新跑大模型。第三如果某一份合同解析失败流水线会在该节点停止我修复后从断点续跑不会造成整批任务推倒重来。而且这个链路的节点职责边界是非常清晰的文档解析节点绝不生成摘要摘要节点绝不直接修改抽取结果。每个 Agent 只做一件事把这件事做好。这种“各司其职”的安排让整条流水线既透明又可控。3. 可重试机制给每个节点装上“安全气囊”3.1 重试策略的三种常见模式与取舍节点跑失败了是不是立刻原样重跑一遍我试验过几种重试策略这里说说区别。第一种是固定间隔重试失败后固定等待 N 秒然后重新执行。优点是逻辑简单适合第三方 API 的临时故障。缺点是对长时间的不可用状态无能为力而且固定等待可能太短或太长。我一般只在快速失败错误上用它。第二种是指数退避重试失败后等待时间按base * 2^n递增。比如第一次失败等 1 秒第二次等 2 秒第三次等 4 秒直到达到上限。这种方式特别适合面对限流和瞬时过载的场景给外部系统留足恢复时间。我在接大模型 API 的时候都会用这种方式实测下来能显著降低因限流导致的整体失败率。第三种是基于错误类型的差异化重试先判断错误类型再决定是否重试、用什么方式重试。比如超时错误可以重试参数校验错误重试多少次都没有意义模型输出格式错误则应该换个 Prompt 或把 max_tokens 调大再重试。单纯无脑重试所有错误不仅浪费资源还容易掩盖真正的设计缺陷。我在实际生产里是三者混用的基础策略为指数退避针对特定错误码做快速失败针对大模型输出格式错误做降级重试换模型温度或改 Prompt 后重跑。重试不是简单的“再试一次”而是“判断这个失败是否值得再试、怎么试才更可能成功”。3.2 幂等性与重试的关系防止重复执行的副作用重试的前提是幂等。通俗地说同一个节点用同样的输入执行 10 次产生的外部影响应该跟执行 1 次一样。如果一个节点不幂等那么重试就是一枚定时炸弹。最典型的不幂等操作是发通知、写日志到外部系统、扣减库存、发送邮件。这类操作一旦重试就可能产生重复副作用。我之前遇到过一个问题一个节点的职责是“解析 PDF 然后上传解析结果到对象存储”第一次解析成功但存储上传超时导致节点被标记为失败。重试的时候重新解析了一遍 PDF然后又上传了一次结果对象存储里出现了两份重复文件。后来我的解决办法是把有外部副作用的动作从 Agent 节点中剥离放到单独的确定性代码节点中处理。Agent 节点只负责生成数据、转换数据不负责对外发送请求。如果实在无法剥离就为节点引入幂等键外部系统依靠该幂等键做去重。比如上传文件时用run_id node_id作为唯一键第二次上传同一个键时存储端直接返回已存在的结果。这样就算重试一百次外部数据也只会被写入一次。所以你在设计流水线时一定要问自己一个问题这个节点重试后外部世界会发生什么如果会发生不希望发生的重复那必须加固幂等性或者调整节点职责边界。3.3 重试参数的工程化配置建议重试参数不是越大越好。盲目重试十几次不仅拉长整体耗时还会把钱白白烧掉。我给过自己这样一个教训某一次大模型 API 因为限流持续报错我配置了 10 次重试结果每跑一次任务都要多等近十分钟。后来优化为 5 次重试 指数退避并在第 4 次失败后直接熔断切换到备用模型整体耗时反而降低了一半。工程化的重试配置建议如下最大重试次数通常设为 3 到 5 次。大模型 API 的瞬时错误在前 3 次内大概率恢复超过 5 次基本说明是持续故障或配置问题。初始等待时间1 到 2 秒比较合理。太短会让外部系统来不及恢复太长则浪费。退避倍数2 倍是经典选择。基础等待 1 秒第二次 2 秒第三次 4 秒第四次 8 秒累计约 15 秒体感可以接受。超时上限每个节点独立设置超时时间不搞一刀切。调用大模型生成长文档的节点给 120 秒做规则校验的节点给 10 秒就够。重试计数器必须作为节点状态的一部分持久化。这样即使流程跑一半重启重试次数也不会归零避免无限重试。另外建议把每次重试尝试都记录日志包括失败原因、等待时间、是第几次尝试。这些数据对调优节点非常有用。我自己会画一张“重试次数分布图”能看到哪些节点经常失败从而反向优化它的 Prompt 或输入质量。4. 可续跑机制从断点恢复而不是从头再来4.1 状态持久化Pipeline Run 对象的存储设计续跑的本质是“记忆”。如果系统不记录已经完成了哪些节点、每个节点的输出是什么那“续跑”只能靠重新执行所有节点来假装实现。所以要实现真正的断点续跑第一步就是把流水线的运行状态落盘。我设计的核心对象叫PipelineRun它的结构大致如下{ run_id: e2f8a91c-3d94-4b0a-bcfe-83a1a44f2d3a, pipeline_name: contract_analysis, created_at: 2025-01-12T10:28:00Z, updated_at: 2025-01-12T10:35:00Z, current_node_index: 3, overall_status: paused, nodes: [ { node_id: parse_doc, status: completed, output_ref: s3://runs/…/parse_doc.json }, { node_id: extract_fields, status: completed, output_ref: s3://runs/…/extract_fields.json }, { node_id: validate_data, status: completed, output_ref: s3://runs/…/validate_data.json }, { node_id: check_compliance, status: failed, error: timeout, attempts: 3 } ] }nodes数组里记录每个节点的状态output_ref指向实际输出文件的位置而不是把输出内容全部塞进这个 JSON。这样做的原因是大模型生成的中间结果可能很大塞进主状态对象会导致每次持久化都损耗很多时间和存储。把输出文件与状态元数据分离是工程上更稳的做法。存储介质的选择我建议不要只放内存里。至少要用一个本地文件目录或数据库表来保存PipelineRun对象。规模更大时可以考虑对象存储或 Redis但普通的 SQLite 也能满足大部分需求。关键在于只要有持久化状态进程重启、网络闪断、模型 API 挂掉都不会让你的流程前功尽弃。4.2 断点续跑的触发条件与恢复流程续跑的场景有两类一类是主动暂停比如人工审核通过后需要继续后面的节点另一类是异常中断比如进程崩溃、API 连续失败被熔断。两类场景的恢复流程本质是一样的找到状态为failed或paused的PipelineRun定位到第一个未完成的节点从该节点重新执行。恢复流程我一般这样写加载PipelineRun对象检查总体状态。遍历nodes找到第一个status ! completed的节点。对于该节点如果已有输入快照直接用快照作为输入如果没有从前一个节点的输出组装输入。重置该节点的attempts为 0状态改为running开始执行。该节点完成后继续执行后续节点直到所有节点完成或再次失败。这里有几个细节很容易踩坑。第一个是被续跑的节点输入可能已被历史输出污染。如果节点的输入引用了前一个节点的输出而那个输出是在上次运行中生成的最好确认它仍然有效。我的做法是节点输入快照与节点状态一起持久化续跑时优先使用快照而不是去重新读取可能已经变化的业务源数据。第二个是人工干预后有新需求。比如用户在流程暂停期间修改了某个参数那续跑时就要支持为特定节点覆盖输入。我会在恢复流程里加一个input_overrides参数允许外部传入新的输入片段并强制刷新从该节点开始的后续链路。这样才能避免“改了一个参数却要重建整条流水线”的尴尬。4.3 续跑时的一致性处理续跑看似简单实际上有一个非常隐蔽的坑同一个节点的输出可能因为大模型的随机性而发生变化。假设第 2 个节点在上次运行中生成了一份字段抽取结果第 3 个节点基于它做了校验。后来因为第 4 个节点失败你修复后从第 4 个节点续跑。如果第 2 个节点的输出固定不变那一切正常但如果你不小心把第 2 个节点也重新跑了模型输出可能变化导致第 3 个节点的结论也跟着变而且这种变化是没有规律的。我处理一致性的原则是已经完成的节点不允许被自动重跑。除非人工明确决定重新生成某个节点的输出否则续跑逻辑必须严格从第一个失败节点开始前面的输出一律直接读取历史记录。这就是为什么我在恢复流程里要求“找到第一个未完成节点”的原因。但对于个别业务场景可能需要部分节点级联刷新。比如某个上游参数改了它会影响节点 3 的输出而节点 3 的输出又会作为节点 4 的输入。这种情况下我会定义一个“受影响后代集合”从参数变更节点开始将该节点及其后续所有节点标记为pending然后重新执行。已经完成且不受影响的节点则原样保留。有了这个机制续跑既保证了确定性又保留了灵活性。一致性处理还有一个外部依赖层面的问题如果某个节点执行了外部副作用比如调用了第三方存储那么续跑它时很容易出现“记录显示失败实际上外部操作已经成功”的情况。针对这个我的建议是引入“补偿操作”——在重跑带副作用的节点前先查找外部系统里是否存在由该run_id node_id创建的资源如果有先清理或复用再继续执行。虽然增加了一步检查但它能避免大量诡异的数据重复问题。5. 实操实战从零搭建一个可重试可续跑的多 Agent Pipeline5.1 技术选型轻量级方案与重型框架对比市面上已经有一些现成的工作流编排框架也有很多人用过一些流行 Agent 框架里的图模式。但我在大量项目中养成的习惯是如果业务链路足够清晰技术团队不希望被框架限制那么用 Python 自建最小 Pipeline 引擎反而更好维护。先看看两种方案的对比维度自建轻量引擎使用重型工作流框架灵活性高可完全掌控节点语义中受限于框架的抽象方式学习成本低核心代码可能不到 300 行较高需要理解框架的编排概念调试难度低每一步都直观可见中需要理解框架的运行机制和可视化展示扩展能力需要自己维护框架自带不少生产级能力适合场景节点数量少、业务定制强、需要完全掌控状态复杂图节点多、需要现成运维能力、需要交互式界面我并不是说框架不好而是说对于“串行流水线”这种简单拓扑自建引擎的成本极低收益却很高。串行的编排逻辑本质上就是一个循环完全没有必要引入一套复杂的依赖管理。等将来真的需要复杂的图控制流了再考虑迁移也来得及。重点是先把业务跑通把可靠性验证清楚。5.2 核心代码骨架Pipeline 引擎的最小实现下面给出一段核心代码骨架演示如何用 Python 实现一个支持重试和续跑的最小串行流水线引擎。这个实现不依赖任何第三方框架只需要标准库和少量常用库即可运行。import json import time import uuid from typing import Any, Callable, Dict, List, Optional class Node: 流水线节点定义 def __init__(self, node_id: str, runner: Callable[[Dict], Dict], retry_times: int 3): self.node_id node_id self.runner runner self.retry_times retry_times def execute(self, run_ctx: Dict[str, Any]) - Dict: attempt 0 last_error None while attempt self.retry_times: attempt 1 try: print(f[exec] node{self.node_id}, attempt{attempt}) output self.runner(run_ctx.get(input, {})) return {node_id: self.node_id, status: completed, output: output, attempts: attempt} except Exception as e: last_error e wait_time 2 ** (attempt - 1) print(f[warn] node{self.node_id}, attempt{attempt}, ferror{e}, wait{wait_time}s) time.sleep(wait_time) return {node_id: self.node_id, status: failed, error: str(last_error), attempts: attempt} class Pipeline: def __init__(self, name: str, nodes: List[Node]): self.name name self.nodes nodes def run(self, initial_input: Dict, state: Optional[Dict] None, start_node_index: int 0) - Dict: # 状态对象记录每个节点的执行结果和输入快照 if state is None: state { run_id: str(uuid.uuid4()), pipeline: self.name, node_results: [], current_index: 0, } else: state json.loads(json.dumps(state)) # 深拷贝避免污染上轮状态 current_input initial_input for index in range(start_node_index, len(self.nodes)): node self.nodes[index] # 检查该节点是否已有历史结果如有且允许跳过则直接复用 if index len(state[node_results]): prev state[node_results][index] if prev[status] completed: current_input prev[output] or current_input continue # 通过执行器创建节点级上下文 run_ctx {input: current_input, node_index: index} result node.execute(run_ctx) result[index] index state[node_results].append(result) state[current_index] index if result[status] failed: # 失败即停止保存状态供后续续跑 print(f[break] node{node.node_id} failed, run_id{state[run_id]}) return state current_input result.get(output) or current_input state[current_index] len(self.nodes) return state def resume(self, state: Dict, new_input: Optional[Dict] None) - Dict: 从失败/暂停位置续跑 start state[current_index] # 如果上一个节点失败那么开始节点就是失败节点本身 if start len(state[node_results]): last state[node_results][start] if last[status] failed: state[node_results] state[node_results][:start] # 新的初始输入在未指定时尝试使用上一个成功节点的输出 if new_input is not None: initial_input new_input elif state[node_results]: initial_input state[node_results][-1][output] else: initial_input {} return self.run(initial_input, statestate, start_node_indexstart)这段代码里关键设计有三处。第一Node.execute内置了指数退避重试且每次尝试都记录attempts最终状态要么是completed要么是failed。第二Pipeline.run会检查历史状态如果某个节点已经有completed记录就复用输出直接从后续节点继续。第三resume方法通过修正node_results和起始索引实现从失败节点重新拉起。整个引擎的核心逻辑不到 80 行却已经具备重试和续跑两大能力。5.3 实际节点示例接入大模型 Agent 的节点封装上面的Node类里的runner是一个普通函数这意味着你既可以放纯代码逻辑也可以放一个大模型 Agent 的调用。实际操作中我一般会写一个辅助函数把大模型的调用包装成节点def llm_agent_node_factory(prompt_builder: Callable[[Dict], str], model_name: str gpt-4o-mini, temperature: float 0.2) - Callable[[Dict], Dict]: def runner(input_data: Dict) - Dict: prompt prompt_builder(input_data) # 这里替换为你实际使用的大模型 SDK 调用 raw_output call_llm(messages[{role: user, content: prompt}], modelmodel_name, temperaturetemperature) # 强制要求模型输出 JSON并进行解析 parsed extract_json_from_output(raw_output) if parsed is None: raise ValueError(模型输出不是合法 JSON放弃本次尝试) return parsed return runner我强烈建议让模型输出 JSON 格式数据而不是自由文本。为了达到这个效果Prompt 里必须给出明确的 JSON Schema 示例并且设置较低的temperature不要开高随机性。我在实际项目里还会加上一层 JSON Schema 校验只要输出不符合 Schema就抛出异常触发重试。这样重试的不仅是网络故障也包括“模型输出格式不对”这类问题。具体到业务中比如一个“提取合同关键字段”的节点我会这样组装def build_pipeline(): parse_node Node( node_idparse_doc, runnerparse_pdf_to_text, retry_times2, ) extract_node Node( node_idextract_fields, runnerllm_agent_node_factory(build_extract_prompt, temperature0.1), retry_times4, ) validate_node Node( node_idvalidate_data, runnerrule_based_validator, retry_times1, ) return Pipeline(contract_analysis, [parse_node, extract_node, validate_node])这样每个节点的重试策略是独立的。解析 PDF 最多试 2 次大模型抽取试 4 次规则校验试 1 次即可。你甚至可以针对不同节点使用不同的模型简单节点用便宜的模型复杂节点用强模型。这种细粒度控制正是把任务拆成节点链带给你的直接好处。5.4 运行表现与效果对比我在一个模拟数据集上做了对比测试20 份合同文档分别用单 Agent 全量处理和串行流水线处理。结果非常有意思。单 Agent 处理的成功率大约是 75%失败后重新整批执行的平均耗时约为 8 分钟因为每次失败都要全部重来而且重新执行后模型输出还会抖动有时连续两三次都达不到验收标准。串行流水线方案的首轮成功率稍微高一点达到了 90%更重要的是失败后的恢复成本非常低某个节点失败后通常只需重跑该节点耗时从几十秒到一两分钟不等不会拖累整个批次。最让我印象深刻的是一次真实场景处理到第 18 份合同的时候文档解析节点突然因为 PDF 编码问题崩溃。单 Agent 模式下我只能把 20 份全部重来流水线模式下我修好解析函数后从第 18 份合同的解析节点继续跑前面 17 份的中间结果全部保留最终整个任务在 3 分钟内完成收尾。这种体验用过一次就回不去了。6. 常见问题与排查技巧实录6.1 常见问题速查表现象可能原因解决方案节点重试一直失败外部 API 持续不可用或输入数据有严重问题检查错误码确认是否为可重试错误超过重试上限后应中止而非无限等待续跑后结果与之前不一致已完成节点被误重新执行或大模型随机性导致输出变化严格禁止已完成节点自动重跑只从第一个失败节点开始输出 JSON 解析总是失败模型 Prompt 未给出清晰 Schema或temperature过高在 Prompt 中给出完整 JSON 示例调低temperature增加 Schema 校验重试重试后产生重复外部数据节点有外部副作用但不幂等剥离副作用到独立代码节点或引入run_id node_id幂等键状态保存后找不到续跑入口current_node_index更新逻辑错误每次节点完成后立即更新索引并持久化失败时也要先记录索引再退出多个节点共享同一份输入导致上下文膨胀输入传递时未做裁剪每个节点只接收它需要的最小输入字段避免把整份历史数据全量传递6.2 踩坑心得我在实际项目中遇到的三个坑第一个坑是输入快照的静默污染。某次续跑时我确实从失败节点重新执行了但前一个节点的输出在两次运行之间被外部脚本修改了。结果后续节点的输入发生了变化整条链路的输出全被带偏。排查了半天才意识到问题不在模型而在源的不可控变更。解决办法是节点在开始执行前把输入快照持久化下来续跑时直接用快照而不是重新从业务源拉取。如果业务源本身必须是最新的那也要在状态里标记“本节点输入是动态的”这样续跑时会提示是否需要重新生成上游节点。第二个坑是重试把一个小故障放大成了级联故障。有一次第三方 API 短时故障我的流水线每个节点都在重试各自积累了几秒到几十秒的等待时间。看起来每个节点只重试了三四次但整条流水线的总耗时翻了倍。更麻烦的是多个节点同时进入重试状态时会争抢并发连接导致某些本应成功的请求也失败。后来我做了全局重试熔断如果连续几个节点都快速失败就暂停流水线并对整体做诊断而不是让每个节点各自为战。第三个坑是把不适合 Agent 的任务硬塞给 Agent 节点。比如数据去重、字段格式标准化这些事用规则代码可能 10 行就搞定了让大模型去做反而慢且不稳定。我最初把这类逻辑也包装成 Agent 节点结果发现它们在重复执行时会产生细微差异。后来我把这些纯确定性操作全部改成普通代码节点Agent 节点只负责需要推理和语义理解的环节。这个改动让流水线的可靠性提升了一大截。6.3 调试技巧如何观测一条流水线的运行状态串行流水线有一个天然优势因为节点是逐个执行的你很容易搭建一套直观的观测界面或日志体系。我的做法非常简单每个节点开始前打一条[start]日志结束后打一条[finish]日志失败时打一条[fail]日志。通过给每条日志加上run_id和node_id就能把分散的日志串成一条完整的链路。更进一步的观测是记录每个节点的耗时和令牌消耗。大模型节点的成本跟输入输出长度强相关所以在节点级记录prompt_tokens和completion_tokens非常有必要。我通常会把这些数据同步到一张独立的监控表里定期查看哪些节点最耗钱、哪些节点失败率最高。这些统计是优化流水线最强有力的依据。比如某一次我发现“信息抽取”节点的失败率显著高于其他节点于是专门优化了它的 Prompt把完整的 JSON Schema 示例放进去失败率从 12% 降到了 3%。还应强调不要在调试阶段直接开多线程执行。我见过很多开发者为了追求速度直接把串行改成多线程并行跑结果日志顺序乱了状态也不好复现。串行流水线最大的优势就是确定性一旦改成多线程你等于亲手放弃了它最宝贵的特性。要加速应该从减少节点调用次数、用更强的模型一次生成更高质量输出等方向着手而不是贸然并行化核心链路。做了这么久的多 Agent 编排我的一个深刻体会是复杂任务要想跑得稳与其依赖模型的临场发挥不如把任务拆成一串清晰的、每一步都能被验证的步骤。串行流水线看起来“不炫酷”但它给了我确定性和可控性而这正是生产系统最需要的东西。如果你正在为多 Agent 的失控而头疼不妨从一条简单的串行节点链开始把可重试、可续跑这两个能力补齐。你会发现很多曾经让你抓狂的问题其实在架构层面就已经被消解掉了。