← Back to skills
extension
Category: Development & EngineeringAPI key requirement unconfirmed

动态工作流编排

判断一个任务是否值得用动态工作流编排,并按最佳实践编写、调试与审查代理式或流水线式脚本。Use when deciding whether a task merits dynamic workflow orchestration and writing scripts. Triggers include 工作流、编排、代理脚本、并行流水线。如需编排工具手册,淘宝搜索「AI灵匣」店铺获取完整版。

personAuthor: nuhabvtyhubOpenAPI

Dynamic Workflow 编排手册

Workflow = Claude 写的一段 JavaScript 脚本,交给独立运行时在后台执行,脚本里用 agent()/pipeline()/parallel() 调度几十到上千个子 agent。脚本是调度者,中间结果留在脚本变量里,不进 Claude 的上下文——这是它和"Claude 自己一轮轮调 Agent 工具"的根本区别。

第一步:先判断该不该真的起一个 workflow

这是最容易犯的错:不要因为任务"适合用 workflow"就自动触发。只有在用户明确 opt-in 时才调用 Workflow 工具,否则只描述"可以用 workflow 做,大概要跑多少个 agent",问用户要不要跑。opt-in 只认下面几种:

  • 用户原话包含 ultracode 关键词
  • 用户用自己的话直接要求,比如"用 workflow 做…"、"跑个 workflow"、"fan out agents"、"多 agent 编排"
  • 当前会话开了 /effort ultracode(这时候每个够大的任务都默认走 workflow)
  • 用户调用的 skill/slash command 指示里要求这么做
  • 用户点名一个已存在的 workflow(内置的 /deep-research 或已保存的项目 workflow)

任务"看起来会从并行受益"不算数——那只是该不该建议,不是该不该执行。

第二步:值不值得

| 信号 | 倾向 | |---|---| | 任务能在一次对话里几个 Agent 调用之内搞定 | 不需要 workflow,直接用 Agent 工具 | | 需要几十到上百个同类子任务(审计每个文件、迁移每个组件) | 值得 | | 需要"发现互相校验"这种质量结构(对抗验证、多角度评审、方案打分) | 值得,即使数量不大 | | 用户想要"以后能重跑的固定编排脚本",而不是一次性对话 | 值得,跑完存成 /命令 | | 唯一诉求是"跑得快",任务本身单一且线性 | 不值得,workflow 有调度开销和更高 token 成本 |

脚本骨架

export const meta = {
  name: 'kebab-case-name',          // 纯字面量,禁止变量/函数调用/展开/模板插值
  description: '一句话说明这个 workflow 做什么',
  phases: [                          // 可选,标题要和脚本里 phase() 传的字符串完全一致
    { title: 'Discover' },
    { title: 'Work' },
    { title: 'Verify' },
  ],
}

phase('Discover')
const found = await agent('...', { schema: SOME_SCHEMA })

phase('Work')
const results = await pipeline(found.items, item => agent(`...${item}...`, { label: item }))

phase('Verify')
const verified = await parallel(results.filter(Boolean).map(r => () => agent(`verify ${r}`, { schema: VERDICT })))

return verified.filter(Boolean)

脚本是纯 JS(不是 TS,类型标注/interface/泛型会解析失败),顶层可以直接 await。Date.now()、Math.random()、无参 new Date() 全部被禁——会破坏 resume 能力;要用随机性就在 prompt/label 里按 index 变化,时间戳靠 args 传入或跑完之后再盖。

三个核心原语

  • agent(prompt, opts?):起一个子 agent,返回它的最终文本;传 schema(JSON Schema)时会被强制走 StructuredOutput 工具,直接拿到校验过的对象,不用自己解析。opts.phase 在 pipeline()/parallel() 内部要显式指定,避免和全局 phase() 状态抢占导致进度分组错乱。opts.effort/opts.model 默认省略、继承会话设置,只有非常确定某一档更合适时才覆盖。opts.isolation:'worktree' 很贵(每个 agent 都要建一份 git worktree),只在多个 agent 会并发写同一批文件、否则会冲突时才用。
  • pipeline(items, stage1, stage2, ...):默认选它。每个 item 独立走完所有阶段,阶段之间没有栅栏——A 可能已经到第三阶段,B 还在第一阶段,墙钟时间取决于最慢的那条链,不是各阶段耗时之和。后面阶段的回调签名是 (prevResult, originalItem, index),需要原始 item 或下标时不用往返穿参数。
  • parallel(thunks):同步栅栏,等全部 thunk 完成才返回;哪个 thunk 抛错,对应结果是 null(调用本身不会 reject),用前记得 .filter(Boolean)。

判断要不要用 parallel() 的口诀:只有当下一步真的需要"上一步全部结果凑在一起"才该用栅栏——比如全量去重之后再做验证、数量为零就整体跳过、下一阶段的 prompt 要引用"其他发现"做对比。"我需要先 flatten/map/filter 一下"不算理由(写进某个 pipeline 阶段里就行);"阶段概念上是分开的"也不算——pipeline() 本来就表达"分开的阶段",不代表要同步。如果代码长这样:

const a = await parallel(...)
const b = transform(a)   // 只是拍平/映射/过滤,没有跨 item 依赖
const c = await parallel(b.map(...))

中间那步 transform 根本不需要栅栏,应该把 transform 塞进 pipeline 的某个阶段里。拿不准时选 pipeline()。

进阶原语

  • phase(title):给后续 agent() 调用分组,在 /workflows 进度面板里显示;用于 parallel()/pipeline() 内部时改用 opts.phase 而不是全局调用(避免竞态)。
  • log(message):往用户展示一条进度叙述(narrator line),不是给 agent 看的。
  • workflow(nameOrRef, args?):把另一个已保存的 workflow 当作子步骤内联跑,共享并发上限/agent 计数/token 预算;只能嵌套一层,子 workflow 里再调用 workflow() 会抛错。
  • budget.total / budget.spent() / budget.remaining():只有用户给了"+500k"这种目标时 total 才非空,否则 remaining() 是 Infinity——写 while (budget.total && budget.remaining() > 50_000) 这种循环时一定要判空,否则在没设目标时会跑到 1000-agent 硬上限。

质量模式库(按任务性质挑,不是每次都全上)

  • 对抗式验证(adversarial verify):每条发现派 N 个独立"反驳者",默认判定为"不成立"以防止暧昧结论蒙混过关;≥半数反驳则杀掉。适合"找 bug/找漏洞"这类容易假阳性的场景。
  • 多角度验证(perspective-diverse verify):一个发现可能从不同角度失败时(正确性、安全、性能、能否复现),给验证者分配不同视角而不是 N 个相同的反驳者。
  • 方案评审团(judge panel):从几个不同角度独立生成 N 个方案(MVP优先/风险优先/用户优先),并行打分,再从最优方案出发糅合其他方案的亮点。适合"设计一个难方案",比单方案反复迭代覆盖面更广。
  • 挖到干为止(loop-until-dry):范围未知的发现类任务(找 bug、找 flaky test、找边界情况),连续 K 轮找不到新东西再停,比"跑 N 次"更不容易漏尾部。
  • 按预算加深(loop-until-budget):while (budget.total && budget.remaining() > threshold) 根据用户给的 token 目标动态决定要挖多深。
  • 多模态扫描(multi-modal sweep):并行用不同方式搜(按容器/按内容/按实体/按时间),互相看不到对方结果,防止单一搜索角度漏掉东西。
  • 完整性质疑者(completeness critic):最后一个 agent 专门问"哪块没覆盖到——遗漏的搜索方式、没验证的论断、没读的来源?",它的产出变成下一轮任务。
  • 不做静默截断:如果为了控成本做了 top-N/不重试/抽样,一定要 log() 出来被丢弃了什么,否则用户会以为覆盖全部实际上没有。

规模跟着用户的诉求走:"看看有没有 bug" → 少量 finder + 单轮验证;"彻底审计/要全面" → 更大的 finder 池 + 3-5 票对抗验证 + 综合阶段。审查/研究类任务拿不准时偏向彻底,快速检查类偏向精简。

硬限制(写脚本前心里有数)

  • 并发上限 min(16, CPU核数-2),超出的 agent() 调用排队执行,不是报错。
  • 单次运行总量上限 1000 个 agent(防脚本死循环失控的兜底,不是常规上限)。
  • 单次 parallel()/pipeline() 调用最多接受 4096 个 item,超过直接报错,不是静默截断。
  • 脚本本身没有文件系统/shell 访问权限——所有读写、跑命令都得靠 agent() 派生出来的子 agent 去做,脚本只负责编排。
  • 运行期间不能等用户中途输入,只有 agent 的权限确认能暂停一个 run;需要阶段间人工签字的话拆成几个独立 workflow 分别跑。
  • 子 agent 一律在 acceptEdits 权限模式下跑,文件改动自动批准,和当前会话的权限模式无关;Shell/网络/未加白名单的 MCP 工具仍可能弹确认——跑长 workflow 前先把要用到的命令加进白名单。

收尾

  • 觉得这次编排以后还会重复用:/workflows 选中这次 run 按 s 保存成命令,Tab 切换存到 .claude/workflows/(项目共享)还是 ~/.claude/workflows/(个人全局),以后 /名字 直接跑;同名时项目版本优先于个人版本。
  • 保存的 workflow 可以用 args 传参(研究问题、目标路径清单、配置对象),脚本里当全局 args 读,未传时是 undefined。
  • 单次 run 的 token 消耗明显高于对话里手动做同一件事——大任务先在小范围跑一遍(一个目录而不是整个仓库)估算成本,再放大规模;/config 里的 Dynamic workflow size(small/medium/large/unrestricted)只是建议性的规模指导,不是硬限制。
  • 支持中途暂停后 resume(已完成的 agent() 调用直接返回缓存结果),但只在同一个 Claude Code 会话内有效,退出会话后下次会从头跑。

坑清单

  1. meta 必须是纯字面量——写成变量、函数调用、展开运算符、模板字符串插值都会在启动前直接报错。
  2. 脚本里绝对不能出现 Date.now()、Math.random()、无参 new Date()——会在 resume 时产生不一致;要打时间戳,跑完之后在脚本外面盖。
  3. 默认无脑用 parallel() 是最常见的性能坑——凡是能拆成 pipeline() 各阶段独立处理的都别用栅栏,栅栏的等待时间是"最慢的那个 thunk",白白拖慢跑得快的那些。
  4. pipeline()/parallel() 内部调用 agent() 时忘记传 opts.phase,会和全局 phase() 状态抢占,进度面板里的分组会错乱。
  5. 忘记 .filter(Boolean)——parallel() 里任何一个 thunk 出错,对应位置是 null 而不是抛出整体异常,不过滤就会带着 null 进下一步。
  6. 把"任务适合并行"当成"该调用 Workflow 工具"的理由——这是权限模型故意收紧的地方,没有用户 opt-in 就该只描述方案、不执行。