多Agent串行流水线：从失控到可控的可重试可续跑编排实践

多 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，它能跑，但效果不稳定。我最终拆成了六个串行节点：

这个拆法有几个明显好处。第一，每个节点可以独立测试，比如“信息抽取”节点我可以准备 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，定位到第一个未完成的节点，从该节点重新执行。

恢复流程我一般这样写：

1. 加载PipelineRun对象，检查总体状态。
2. 遍历nodes，找到第一个status != "completed"的节点。
3. 对于该节点，如果已有输入快照，直接用快照作为输入；如果没有，从前一个节点的输出组装输入。
4. 重置该节点的attempts为 0，状态改为running，开始执行。
5. 该节点完成后，继续执行后续节点，直到所有节点完成或再次失败。

这里有几个细节很容易踩坑。第一个是被续跑的节点输入可能已被历史输出污染。如果节点的输入引用了前一个节点的输出，而那个输出是在上次运行中生成的，最好确认它仍然有效。我的做法是节点输入快照与节点状态一起持久化，续跑时优先使用快照，而不是去重新读取可能已经变化的业务源数据。

第二个是人工干预后有新需求。比如用户在流程暂停期间修改了某个参数，那续跑时就要支持为特定节点覆盖输入。我会在恢复流程里加一个input_overrides参数，允许外部传入新的输入片段，并强制刷新从该节点开始的后续链路。这样才能避免“改了一个参数却要重建整条流水线”的尴尬。

4.3 续跑时的一致性处理

续跑看似简单，实际上有一个非常隐蔽的坑：同一个节点的输出可能因为大模型的随机性而发生变化。假设第 2 个节点在上次运行中生成了一份字段抽取结果，第 3 个节点基于它做了校验。后来因为第 4 个节点失败，你修复后从第 4 个节点续跑。如果第 2 个节点的输出固定不变，那一切正常；但如果你不小心把第 2 个节点也重新跑了，模型输出可能变化，导致第 3 个节点的结论也跟着变，而且这种变化是没有规律的。

我处理一致性的原则是：已经完成的节点不允许被自动重跑。除非人工明确决定重新生成某个节点的输出，否则续跑逻辑必须严格从第一个失败节点开始，前面的输出一律直接读取历史记录。这就是为什么我在恢复流程里要求“找到第一个未完成节点”的原因。

但对于个别业务场景，可能需要部分节点级联刷新。比如某个上游参数改了，它会影响节点 3 的输出，而节点 3 的输出又会作为节点 4 的输入。这种情况下，我会定义一个“受影响后代集合”，从参数变更节点开始，将该节点及其后续所有节点标记为pending，然后重新执行。已经完成且不受影响的节点则原样保留。有了这个机制，续跑既保证了确定性，又保留了灵活性。

一致性处理还有一个外部依赖层面的问题：如果某个节点执行了外部副作用（比如调用了第三方存储），那么续跑它时很容易出现“记录显示失败，实际上外部操作已经成功”的情况。针对这个，我的建议是引入“补偿操作”——在重跑带副作用的节点前，先查找外部系统里是否存在由该run_id + node_id创建的资源，如果有，先清理或复用，再继续执行。虽然增加了一步检查，但它能避免大量诡异的数据重复问题。

5. 实操实战：从零搭建一个可重试可续跑的多 Agent Pipeline

5.1 技术选型：轻量级方案与重型框架对比

市面上已经有一些现成的工作流编排框架，也有很多人用过一些流行 Agent 框架里的图模式。但我在大量项目中养成的习惯是：如果业务链路足够清晰，技术团队不希望被框架限制，那么用 Python 自建最小 Pipeline 引擎反而更好维护。先看看两种方案的对比：

我并不是说框架不好，而是说对于“串行流水线”这种简单拓扑，自建引擎的成本极低，收益却很高。串行的编排逻辑本质上就是一个循环，完全没有必要引入一套复杂的依赖管理。等将来真的需要复杂的图控制流了，再考虑迁移也来得及。重点是先把业务跑通，把可靠性验证清楚。

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}, " f"error={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, state=state, start_node_index=start)

这段代码里，关键设计有三处。第一，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}], model=model_name, temperature=temperature) # 强制要求模型输出 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_id="parse_doc", runner=parse_pdf_to_text, retry_times=2, ) extract_node = Node( node_id="extract_fields", runner=llm_agent_node_factory(build_extract_prompt, temperature=0.1), retry_times=4, ) validate_node = Node( node_id="validate_data", runner=rule_based_validator, retry_times=1, ) 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 常见问题速查表

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 的失控而头疼，不妨从一条简单的串行节点链开始，把可重试、可续跑这两个能力补齐。你会发现，很多曾经让你抓狂的问题，其实在架构层面就已经被消解掉了。