pipeline

pipeline 技能

概述

pipeline 技能支持在定义好的工作流中将多个 Agent 串联起来，其中一个 Agent 的输出会成为下一个 Agent 的输入。这可以构建强大的 Agent 流水线，类似 Unix 管道，但专为 AI Agent 编排而设计。

核心概念

1. 顺序流水线

最简单形式：Agent A 的输出流向 Agent B，再流向 Agent C。

explore -> architect -> executor

流程：

1. explore Agent 搜索代码库并产出发现
2. architect 接收发现并产出分析/建议
3. executor 接收建议并实施变更

2. 分支流水线

根据输出条件路由到不同的 Agent。

explore -> {
  if "complex refactoring" -> architect -> executor-high
  if "simple change" -> executor-low
  if "UI work" -> designer -> executor
}

3. 并行再合并流水线

并行运行多个 Agent，并合并它们的输出。

parallel(explore, document-specialist) -> architect -> executor

内置 pipeline 预设

Review pipeline

用途： 全面的代码审查与实施

/pipeline review <task>

阶段：

1. explore - 查找相关代码和模式
2. architect - 分析架构和设计影响
3. critic - 审查并批判性评估分析
4. executor - 在完整上下文中实施

适用于： 大型功能、重构、复杂变更

---

Implement pipeline

用途： 有计划的实施与测试

/pipeline implement <task>

阶段：

1. planner - 创建详细的实施计划
2. executor - 实施该计划
3. test-engineer - 添加/验证测试

适用于： 需求明确的新功能

---

Debug pipeline

用途： 系统化调试工作流

/pipeline debug <issue>

阶段：

1. explore - 定位错误位置和相关代码
2. architect - 分析根因
3. build-fixer - 应用修复并验证

适用于： Bug、构建错误、测试失败

---

Research pipeline

用途： 外部研究 + 内部分析

/pipeline research <topic>

阶段：

1. parallel(document-specialist, explore) - 外部文档 + 内部代码

2. architect - 综合发现

3. writer - 记录建议

适用于： 技术决策、API 集成

---

Refactor pipeline

用途： 安全、经验证的重构

/pipeline refactor <target>

阶段：

1. explore - 查找所有用法和依赖
2. architect-medium - 设计重构策略
3. executor-high - 执行重构
4. qa-tester - 验证无回归

适用于： 架构变更、API 重新设计

---

Security pipeline

用途： 安全审计与修复

/pipeline security <scope>

阶段：

1. explore - 查找潜在漏洞
2. security-reviewer - 审计并识别问题
3. executor - 实施修复
4. security-reviewer-low - 重新验证

适用于： 安全审查、漏洞修复

---

自定义 pipeline 语法

基本顺序

/pipeline agent1 -> agent2 -> agent3 "task description"

示例：

/pipeline explore -> architect -> executor "add authentication"

指定模型

/pipeline explore:haiku -> architect:opus -> executor:sonnet "optimize performance"

使用分支

/pipeline explore -> (
  complexity:high -> architect:opus -> executor-high:opus
  complexity:medium -> executor:sonnet
  complexity:low -> executor-low:haiku
) "fix reported issues"

使用并行阶段

/pipeline [explore, document-specialist] -> architect -> executor "implement OAuth"

数据传递协议

pipeline 中的每个 Agent 都会收到来自上一阶段的结构化上下文：

{
  "pipeline_context": {
    "original_task": "user's original request",
    "previous_stages": [
      {
        "agent": "explore",
        "model": "haiku",
        "findings": "...",
        "files_identified": ["src/auth.ts", "src/user.ts"]
      }
    ],
    "current_stage": "architect",
    "next_stage": "executor"
  },
  "task": "specific task for this agent"
}

错误处理

重试逻辑

当某个 Agent 失败时，pipeline 可以：

1. 重试 - 重新运行同一 Agent（最多 3 次）
2. 跳过 - 使用部分输出继续下一阶段
3. 中止 - 停止整个 pipeline
4. 回退 - 路由到替代 Agent

配置：

/pipeline explore -> architect -> executor --retry=3 --on-error=abort

错误恢复模式

模式 1：回退到更高层级

executor-low -> on-error -> executor:sonnet

模式 2：咨询 architect

executor -> on-error -> architect -> executor

模式 3：人在回路

any-stage -> on-error -> pause-for-user-input

pipeline 状态管理

pipeline 在 .omc/pipeline-state.json 中维护状态：

{
  "pipeline_id": "uuid",
  "name": "review",
  "active": true,
  "current_stage": 2,
  "stages": [
    {
      "name": "explore",
      "agent": "explore",
      "model": "haiku",
      "status": "completed",
      "output": "..."
    },
    {
      "name": "architect",
      "agent": "architect",
      "model": "opus",
      "status": "in_progress",
      "started_at": "2026-01-23T10:30:00Z"
    },
    {
      "name": "executor",
      "agent": "executor",
      "model": "sonnet",
      "status": "pending"
    }
  ],
  "task": "original user task",
  "created_at": "2026-01-23T10:25:00Z"
}

验证规则

在 pipeline 完成前，请验证：

• 所有阶段均成功完成
• 最终阶段的输出解决了原始任务
• 任何阶段都没有未处理的错误
• 所有被修改的文件均通过 lsp_diagnostics
• 测试通过（如适用）

高级特性

条件分支

根据 Agent 输出，路由到不同路径：

explore -> {
  if files_found > 5 -> architect:opus -> executor-high:opus
  if files_found <= 5 -> executor:sonnet
}

循环结构

重复执行各阶段，直到满足条件：

repeat_until(tests_pass) {
  executor -> qa-tester
}

合并策略

当并行 Agent 完成时：

• concat：拼接所有输出
• summarize：使用 architect 汇总结果
• vote：使用 critic 选择最佳输出

使用示例

示例 1：功能实现

/pipeline review "add rate limiting to API"

→ 触发：explore → architect → critic → executor

示例 2：Bug 修复

/pipeline debug "login fails with OAuth"

→ 触发：explore → architect → build-fixer

示例 3：自定义链

/pipeline explore:haiku -> architect:opus -> executor:sonnet -> test-engineer:sonnet "refactor auth module"

示例 4：研究驱动的实现

/pipeline research "implement GraphQL subscriptions"

→ 触发：parallel(document-specialist, explore) → architect → writer

取消

停止活动中的流水线：

/pipeline cancel

或使用通用的取消命令，它会检测活动中的流水线。

与其他技能的集成

流水线可在其他技能中使用：

• Ralph：循环执行流水线，直到验证完成
• Ultrawork：并行运行多个流水线
• Autopilot：将流水线用作构建块

最佳实践

1. 从预设开始 - 在创建自定义流水线之前先使用内置流水线
2. 让模型匹配复杂度 - 不要在简单任务上浪费 opus
3. 保持各阶段专注 - 每个 Agent 都应有一个明确的职责
4. 使用并行阶段 - 同时运行相互独立的工作
5. 在检查点验证 - 使用 architect 或 critic 验证进度
6. 记录自定义流水线 - 保存成功的模式以便复用

故障排查

流水线卡住

检查： .omc/pipeline-state.json 中的当前阶段 修复： 使用 /pipeline resume 恢复，或取消并重启

Agent 反复失败

检查： 重试次数和错误消息 修复： 路由到更高层级的 Agent，或添加 architect 咨询

输出未流转

检查： Agent 提示词中的数据传递结构 修复： 确保每个 Agent 的提示词中都包含 pipeline_context

技术实现

流水线编排器：

1. 解析流水线定义 - 验证语法和 Agent 名称
2. 初始化状态 - 创建 pipeline-state.json
3. 按顺序执行各阶段 - 使用 Task 工具启动 Agent
4. 在阶段之间传递上下文 - 为下一个 Agent 组织输出结构
5. 处理分支逻辑 - 评估条件并进行路由
6. 管理并行执行 - 启动并发 Agent 并合并结果
7. 持久化状态 - 在每个阶段后更新状态文件
8. 强制验证 - 在完成前运行检查

完成时清理状态

重要：完成时删除状态文件 - 不要只是设置 active: false

当流水线完成时（所有阶段已完成或已取消）：

# Delete pipeline state file
rm -f .omc/state/pipeline-state.json

这能为未来的会话确保状态干净。不应留下 active: false 的过期状态文件。

技能调用

此技能在以下情况下激活：

• 用户输入 /pipeline 命令
• 用户提到 "agent chain"、"workflow"、"pipe agents"
• 检测到模式：带有 Agent 名称的 "X then Y then Z"

显式调用：

/oh-my-claudecode:pipeline review "task"

自动检测：

"First explore the codebase, then architect should analyze it, then executor implements"

→ 自动创建流水线：explore → architect → executor