一、Pipeline模式概述
Pipeline（流水线/管道）模式是子代理编排中最基础也是最常用的模式之一。它将一个复杂的任务拆解为多个按顺序执行的阶段（Stage），每个阶段专注完成一个明确定义的子任务，前一阶段的输出自动成为下一阶段的输入，形成一条完整的数据处理链路。
这种模式的核心理念源自Unix哲学的"管道"思想——每个程序只做好一件事，通过标准输入输出连接组合成强大的处理流程。在AI Agent的上下文中，每个"程序"被替换为一个专门的子代理，它们各自拥有独特的指令、工具和知识库，协同完成远超单个Agent能力范围的复杂任务。
任务划分为多个顺序执行的阶段
按照处理逻辑的自然边界将任务拆解，每个阶段内部高度内聚，对外仅暴露清晰的输入输出接口。
每个阶段由一个或一组子代理处理
根据阶段复杂度，可以安排单个子代理独立完成，也可以组织多个子代理以扇出模式并行处理。
前一阶段的输出是下一阶段的输入
数据在阶段之间以明确定义的格式流转，通常通过共享文件或结构化消息传递，保证数据传递的可靠性和可追溯性。
适合有明确处理顺序的工作流
当任务具有天然的顺序依赖关系（如先采集再清洗后分析）时，Pipeline模式提供了最直观和高效的解决方案。
核心思想：Pipeline模式通过"分而治之"的策略，将一个难以一次性完成的复杂任务，分解为多个可管理、可测试、可复用的子任务阶段，每个阶段由最擅长该任务的子代理负责执行。
输入数据 → 阶段1 → 阶段2 → 阶段3 → ... → 阶段N → 最终输出
（每个阶段由专用子代理执行，数据依次流经所有阶段）
二、Pipeline阶段的定义
阶段（Stage）是Pipeline模式的基本组成单元。一个好的阶段定义直接影响整个工作流的效率、可维护性和可靠性。阶段划分需要遵循清晰的原则，确保每个阶段既有明确的职责边界，又能与上下游顺畅衔接。
阶段划分原则
每个阶段有清晰边界：阶段的职责范围应当明确且不重叠。例如在数据处理Pipeline中，"数据清洗"和"数据分析"是两个边界清晰的阶段——清洗负责修正格式、去除噪声，分析负责从中提取洞察。如果两个阶段的职责模糊不清，不仅会导致子代理指令难以编写，还会增加调试和排错的难度。
阶段输入输出定义明确：每个阶段必须明确定义其接收什么格式的数据、输出什么格式的结果。明确的接口契约是阶段之间能够顺畅衔接的前提。通常建议使用结构化的数据格式（如JSON）作为阶段间的传递载体，并在每个阶段入口处进行数据校验。
阶段间耦合度低，便于独立开发和测试：低耦合意味着每个阶段可以独立开发、单独测试，不依赖其他阶段的具体实现。这使得团队可以并行开发不同阶段，也使得后续对某个阶段的优化或替换不会影响到整个Pipeline的稳定性。
最佳实践：在设计和实现阶段时，可以借鉴"单一职责原则"——每个阶段只做一件事，并且把这件事做到极致。如果一个阶段的描述中出现了"和"、"以及"等连词，通常意味着它应该被进一步拆分为两个独立的阶段。
典型Pipeline示例：数据处理流程
一个典型的数据处理Pipeline通常包含以下阶段，每个阶段都有明确的输入输出：
| 阶段 |
输入 |
处理内容 |
输出 |
| 数据采集 |
源数据URL/文件路径 |
从多个数据源抓取原始数据，处理网络异常和重试 |
原始数据集（JSON/CSV） |
| 数据清洗 |
原始数据集 |
去除重复记录、修正格式错误、填充缺失值 |
清洗后的标准化数据 |
| 数据分析 |
标准化数据 |
执行统计分析、模式识别、异常检测 |
分析报告（含图表和关键指标） |
| 报告生成 |
分析结果 |
将分析数据格式化为可视化报告、生成摘要 |
最终报告（HTML/PDF） |
阶段粒度控制：阶段的粒度不是越细越好。粒度过细会导致Pipeline过长、调度开销增大；粒度过粗则失去拆分意义。经验法则是：如果一个阶段的处理逻辑可以清晰描述为3-5个步骤，且执行时间在可接受的范围内，那么这个粒度就是合适的。
三、阶段间的数据传递
数据在Pipeline的各个阶段之间高效、可靠地传递，是Pipeline模式能够正常运转的基础。选择合适的数据传递方式，直接影响系统的性能、可追溯性和错误处理能力。
前一阶段将结果写入共享文件
最常用的数据传递方式是共享文件系统。每个阶段完成处理后，将结果写入预定义的共享目录中，后一阶段从该目录读取所需文件。这种方式的优势在于：文件是持久化的，便于调试时查看中间结果；文件系统天然支持大体积数据传输；无需额外的消息中间件即可实现。
# 阶段1：数据采集 — 输出到共享文件
agent_data_collector.run(
task="从API获取用户数据",
output_file="/pipeline/shared/raw_users.json"
)
# 阶段2：数据清洗 — 从共享文件读取
agent_data_cleaner.run(
task="清洗用户数据，去除重复和无效记录",
input_file="/pipeline/shared/raw_users.json",
output_file="/pipeline/shared/cleaned_users.json"
)
支持多种数据格式
不同的阶段可能需要不同的数据格式。Pipeline设计应当支持灵活的格式转换：
- JSON：结构化数据的首选格式，适合对象、数组等复杂结构，具有良好的可读性和跨语言兼容性。
- CSV：适合表格数据，轻量且易于用电子表格工具查看，在处理大量记录时比JSON更紧凑。
- 纯文本：适合自然语言处理场景，如文本摘要、翻译等阶段的输入输出。
- 二进制文件：适合图像、音频等多媒体数据的处理场景。
数据的验证和转换
在每个阶段读取输入数据后，应当进行数据验证，确保数据格式和内容符合预期。验证内容包括：数据结构完整性（必要字段是否存在）、数据类型正确性（字段类型是否匹配）、数据值合理性（数值是否在有效范围内）等。验证失败时，Pipeline应能够优雅地处理错误——可以选择重试当前阶段、跳过异常记录，或终止Pipeline并报告失败原因。
数据转换通常在阶段入口和出口进行。入口转换将上游的输出格式调整为当前阶段内部处理的格式；出口转换将处理结果标准化为下游期望的格式。这种"适配器"模式可以降低阶段间的耦合，使不同阶段可以独立演进。
注意：数据传递过程中必须考虑数据的安全性。如果Pipeline处理的是敏感信息（如个人身份信息、医疗数据等），共享文件应当设置适当的访问权限，必要时对敏感字段进行脱敏或加密处理。阶段之间传递的数据应当遵循"最小必要"原则——只传递下游阶段真正需要的数据，避免无关信息的泄露风险。
四、并行和条件分支
基本的线性Pipeline虽然简单直观，但现实世界的任务往往需要更灵活的控制流。在Pipeline中引入并行处理和条件分支机制，可以显著提升处理效率和工作流的表达能力。
扇出（Fan-out）：多个子代理并行执行
当一个阶段的输出需要被多个独立的子任务同时处理时，可以采用扇出模式。例如，在内容审核Pipeline中，一篇文章可能需要同时进行"敏感词检查"、"版权检测"和"格式校验"三项审核，三个子代理可以并行执行，互不干扰。扇出模式充分利用了子代理的独立性，大幅缩短了整体处理时间。
原始内容 →
├─ 敏感词检查（子代理A）
├─ 版权检测（子代理B）
└─ 格式校验（子代理C）
扇入（Fan-in）：并行结果合并
多个并行子代理的处理结果需要汇总后进入下一阶段，这就是扇入操作。扇入阶段负责收集所有并行子任务的输出，进行结果合并、冲突消解和一致性检查。例如上述内容审核案例中，扇入阶段会收集三个子代理的审核结果，综合判断文章是否通过审核——只有当所有并行任务都返回"通过"时，才允许进入下一阶段。
class FanInAgent:
def collect_results(self, parallel_results):
# 收集所有并行子代理的输出
all_passed = all(
r["status"] == "pass"
for r in parallel_results
)
# 合并审核意见
combined_report = {
"overall_status": "pass" if all_passed else "fail",
"details": parallel_results,
"summary": self.generate_summary(parallel_results)
}
return combined_report
条件分支：根据阶段结果选择不同路径
Pipeline并不是只能走直线。引入条件分支逻辑后，工作流可以根据某个阶段的处理结果，动态决定下一步的方向。常见的分支场景包括：
- 质量门禁：如果代码审查阶段发现严重问题，Pipeline直接终止并通知开发者，不再继续后续的构建和部署。
- 内容分类：根据文档分类结果，将文档路由到不同的处理链路（如技术文档走技术审核Pipeline，营销文案走市场审核Pipeline）。
- 异常处理：当某个阶段处理失败时，进入专门的异常处理分支进行重试或降级处理，而不是让整个Pipeline崩溃。
代码提交 → 格式检查 →
├─ 通过 → 代码审查 → 构建
└─ 失败 → 通知开发者修复 → （等待修复后重试）
循环反馈：回退到前序阶段
循环反馈机制允许Pipeline在特定条件下回退到之前的某个阶段进行修正。这在质量要求较高的场景中非常有用。例如，代码审查阶段发现代码风格不符合规范时，可以自动将代码回退到"格式修正"阶段，由专门的子代理进行自动修复后重新进入审查。循环反馈需要设定最大回退次数，以防止出现死循环。
设计建议：并行和条件分支虽然强大，但也会增加Pipeline的复杂度。建议在基本线性Pipeline验证通过后，再逐步引入这些高级特性。每个分支路径都应当有对应的测试用例，确保条件判断逻辑的正确性。
五、完整Pipeline案例
下面通过一个完整的"代码提交流程"案例，展示如何将上述所有概念综合运用到一个实际的Pipeline编排中。该Pipeline在开发团队提交代码后自动触发，包含5个顺序阶段，其中部分阶段包含并行和条件分支逻辑。
代码提交流Pipeline阶段详解
| 阶段 |
子代理 |
输入 |
输出 |
| 格式检查 |
格式检查子代理 |
提交的代码文件 |
格式检查报告（通过/失败+详情） |
| 代码审查 |
代码审查子代理 |
代码文件+格式检查报告 |
审查意见（问题列表+严重级别） |
| 安全扫描 |
安全扫描子代理 |
代码文件+依赖列表 |
安全漏洞报告 |
| 构建验证 |
构建验证子代理 |
审查通过+安全通过的代码 |
构建结果（成功/失败+日志） |
| 通知结果 |
通知子代理 |
所有阶段的汇总报告 |
发送通知给相关开发者 |
Pipeline执行流程
整个Pipeline的执行逻辑如下：
- 格式检查阶段：格式检查子代理接收提交的代码文件，运行代码格式化工具进行检查。如果格式不符合规范，子代理会生成包含具体违规位置和修改建议的报告。条件分支：如果格式问题严重（违规数量超过阈值），Pipeline会生成修复建议并通知开发者，不继续后续阶段。如果问题轻微，则自动修复后继续。
- 代码审查阶段（扇出并行）：代码通过格式检查后，进入代码审查阶段。在此阶段，多个审查子代理可以并行工作——一个子代理负责检查业务逻辑正确性，另一个子代理检查代码性能和可维护性，第三个子代理检查测试覆盖率。扇入合并：所有审查意见汇总后，由合并模块综合评估是否通过审查。
- 安全扫描阶段：安全扫描子代理对代码及其依赖包进行安全漏洞扫描。它会检查已知漏洞数据库、硬编码密钥、不安全的API调用等安全问题。
- 构建验证阶段：在审查和安全扫描都通过后，构建验证子代理执行代码编译和单元测试，确保代码能够成功构建且所有测试通过。
- 通知结果阶段：所有阶段完成后，通知子代理收集整个Pipeline的执行结果，生成格式化的报告，并通过合适渠道（如邮件、即时消息）发送给相关的开发人员。
# Pipeline编排伪代码示例
pipeline = Pipeline(name="代码提交流程")
# 定义阶段1：格式检查
stage1 = Stage(
name="格式检查",
agent=format_check_agent,
condition=lambda result: result["severity"] < 5,
on_fail="通知开发者修复"
)
# 定义阶段2：代码审查（扇出并行）
stage2 = ParallelStage(
name="代码审查",
agents=[logic_agent, perf_agent, test_agent],
merge_strategy="all_pass"
)
# 定义阶段3-5：安全扫描、构建验证、通知
stage3 = Stage(name="安全扫描", agent=security_agent)
stage4 = Stage(name="构建验证", agent=build_agent)
stage5 = Stage(name="通知结果", agent=notify_agent)
# 组装Pipeline
pipeline.add_stages(stage1, stage2, stage3, stage4, stage5)
pipeline.run(input_data=code_commit)
案例要点总结：该Pipeline展示了线性执行（阶段1→3→4→5）与并行执行（阶段2的三个审查子代理）的混合编排，同时引入了条件分支（格式检查失败时终止Pipeline）和扇入合并（并行审查结果汇总）两种控制流模式。每个子代理仅需关注自己专业领域内的任务，职责明确、易于测试和迭代优化。
实践提示：在实际构建Pipeline时，建议为每个阶段添加超时控制和重试机制。超时防止某个子代理陷入无响应状态，重试机制则能应对偶发的临时故障。同时，记录每个阶段的执行日志和耗时信息，便于后期对Pipeline进行性能分析和优化。