为 Claude Code 的 Workflow 工具编写可运行的编排脚本:用纯 JavaScript 控制流,把工作确定性地扇出给一批「全新上下文」子智能体。只有叶子 agent() 调用消耗 token,主会话保持干净,整个运行可中断、可恢复。
何时使用
满足以下任一条件,工作流才值得它的成本:工作可并行或多阶段、必须可复现、链路足够长以致可能中途失败(恢复很关键)、或每步都需隔离在各自的上下文窗口里。
选型对照:
| 场景 | 用什么 |
|---|---|
| 单个子智能体、一个任务 | 普通 Agent 工具 |
| 可复用流程、由 Claude 动态决定步骤 | 一个 Skill |
| 多个子智能体、固定拓扑、确定且可恢复 | Workflow(本技能) |
不该用的边界:一次性任务直接让 Claude 做;若 intake 后发现只是单智能体单任务,明确建议改用 Agent;若是需要模型自行挑选步骤而非固定拓扑的流程,建议改用 Skill。
步骤
每次会话必须先做 intake,不要跳过直接写代码。
- 问清要建什么。核心问句:要自动化的可重复多步任务是什么?单个子智能体一次做的「一个工作单元」是什么?单元数量是已知清单还是靠循环发现?后续步骤是否需要一次性拿到全部前序结果,还是每条数据可独立流转?是否有步骤需要返回结构化数据(判定/列表/分数)?大概要消耗多少 token、走多深(定预算护栏)?
- 用户含糊时不要追问到底,跑推荐引擎把已知信息转成 1-2 个具体方案,并连同理由一起给出(「这是我会建的方案和原因」)。
- 与用户确认形态(拓扑 + 阶段 + 并行还是流水线)后再落盘——这是唯一的审批门。
答案映射拓扑:独立单元+已知清单+末尾汇总 → 扇出后汇总;有序阶段+各条独立推进 → 流水线(默认首选);某阶段需要全部前序集合做去重/合并/计数提前退出 → 屏障;数量未知、按目标/预算/枯竭停止 → 受护栏约束的循环;解空间宽、要 best-of-N → 评审团;错误代价高 → 在该结论上加怀疑投票验证。
指令
构建 → 校验 → 运行循环:
- 从确认的拓扑生成脚手架:
python scripts/scaffold_workflow.py --topology pipeline --name pr-triage \
--description "Triage open PRs" > .claude/workflows/pr-triage.js
- 编辑文件:先写
meta块(纯字面量、首条语句),再写 async 主体,使用注入的全局:agent()、pipeline()、parallel()、phase()、log()、budget、args、workflow()。 - 运行前先校验,拦住会让解析器致命失败的错误:
python scripts/validate_workflow.py .claude/workflows/pr-triage.js
- 运行:
export CLAUDE_CODE_WORKFLOWS=1开启特性,把文件存到.claude/workflows/,用/workflows启动并实时观察。按 P 暂停/恢复,按 X 跳过某个子智能体;失败的智能体会自动重试。
关键 API:agent(prompt, opts?) 返回文本或(设了 schema 时)校验过的对象;opts 含 label、phase、schema、model(haiku/sonnet/opus/inherit/完整模型 ID)、isolation: 'worktree'。pipeline(items, ...stages) 各条数据独立穿过所有阶段、阶段间无屏障;parallel(thunks) 并发跑一组 () => Promise 缩略函数并等全部完成(屏障)。budget 有 total/spent()/remaining(),超额抛 WorkflowBudgetExceededError。模型选择:分类/抽取用轻模型(Haiku),综合/硬推理用重模型(Opus)。
示例
扇出后汇总——已知清单、各自一遍、末尾合一:
const findings = await parallel(
questions.map((q, i) => () =>
agent(`Research and report verified facts:\n\n${q}`,
{ label: `q${i + 1}`, schema: RESEARCH_SCHEMA }))
)
const report = await agent(
`Synthesize these findings into one report:\n${JSON.stringify(findings.filter(Boolean))}`,
{ model: 'opus' }
)
流水线——各条数据就绪即推进,不等同伴:
const results = await pipeline(
DIMENSIONS,
d => agent(d.prompt, { label: `review:${d.key}`, phase: 'Review', schema: FINDINGS_SCHEMA }),
review => parallel((review?.findings ?? []).map(f => () =>
agent(`Adversarially verify: ${f.title}`, { schema: VERDICT_SCHEMA })))
)
受预算约束的循环——深度随 token 目标伸缩,护栏必备:
const issues = []
while (budget.total && budget.remaining() > 50_000) {
const r = await agent('Find one more issue not yet reported...', { schema: ISSUE_SCHEMA })
if (!r?.issues?.length) break
issues.push(...r.issues)
}
meta 块写法(首条语句、纯字面量):
export const meta = {
name: 'pr-triage',
description: 'One-line summary',
whenToUse: 'When to run this',
phases: [{ title: 'Review', detail: 'One reviewer per PR', model: 'haiku' }]
}
注意事项
校验器强制的硬规则:
meta必须是纯字面量且为首条语句——内部不得有变量、展开、模板字符串或函数调用;保留键(__proto__、constructor、prototype)会被拒。- 禁止非确定性:
Date.now()、Math.random()、无参new Date()都会破坏恢复——时间戳通过args传入,需要随机性时改用索引来变化提示词。 - 编排器内禁用文件系统/Node API(
require、fs、process、网络)——这些工作要放进agent()提示词里(子智能体有完整工具权限)。 parallel()接收缩略函数(() => agent(...)),不是裸 Promise。多阶段工作默认用pipeline(),除非某阶段确实需要整批前序结果。- 每个开放式循环都要用计数器或
budget.remaining()护栏约束,否则会撞上 1000 个智能体的上限(WorkflowAgentCapError)。 - 过滤被跳过/失败的智能体:
results.filter(Boolean)。
其他上限:并发智能体数 min(16, max(2, 核数 − 2));脚本体积上限 524288 字节;每个智能体卡死超时 180000 ms(3 分钟),最多重试 5 次;同步超时 30000 ms 用于捕获无限同步循环。恢复缓存键包含 schema/model/isolation/agentType,改动其一会在恢复时重跑该智能体;label 与 phase 不影响缓存。恢复仅限同一会话:编辑保存的文件后用 scriptPath 重新调用。阶段间用 JSON.stringify(...) 把结构化数据塞进下一个提示词;schema 只约束单次 agent() 调用返回的形状。
互见
- skill-creator:当 intake 结论是「应做成 Skill 而非工作流」时改用它。
- mcp-builder:子智能体需要自定义工具/资源时,先用它把能力封装成 MCP。
- claude-api:把同类编排思路落到 Anthropic SDK / Managed Agents 代码里时参考。
- code-reviewer:PR 评审、对抗式验证等是本技能扇出/怀疑投票模式的典型应用场景。
- first-principles-thinking:intake 阶段拆解「一个工作单元」与拓扑选型时的思维框架。
本条采编自 alirezarezvani/claude-skills(MIT 许可证)。