Dynamic Workflow 源函数详解

本页把 Dynamic Workflow 的每个源函数单独讲透：签名、语义、返回值、失败路径、真实脚本里的用法和常见错误。样本是 workflow_research_simple.js（427 行，行号均指该文件）。执行模型与运行时约束见 principles，逐段精读见 example-research-simple。

全貌

原语	签名	一句话
`agent`	`agent(prompt, opts?) => Promise<any>`	派一个子 Agent，拿它的结果
`pipeline`	`pipeline(items, ...stages) => Promise<any[]>`	每个 item 独立流过所有 stage，无 barrier
`parallel`	`parallel(thunks) => Promise<any[]>`	并发跑一组 thunk，等全部完成
`phase`	`phase(title) => void`	开启一个进度阶段
`log`	`log(message) => void`	输出一条进度说明
`workflow`	`workflow(nameOrRef, args?) => Promise<any>`	内联调用另一个 workflow，仅一层
`args`	全局变量	调用时传入的输入
`budget`	全局对象	本轮 token 预算
`meta`	导出常量	声明工作流元信息

脚本是纯 JavaScript，支持顶层 await，可用 JSON/Math/Array 等标准内置；不支持 TypeScript 语法、import()、文件系统或 Node API。Date.now()、Math.random()、无参 new Date() 会抛错，用于保证重放可复现。

`agent(prompt, opts?)`

基本语义

派生一个子 Agent 执行 prompt，返回其结果。

// 无 schema：返回子 Agent 的最终文本字符串
const text = await agent('Summarize src/index.ts')
 
// 有 schema：子 Agent 被强制调用 StructuredOutput，返回校验通过的对象
const found = await agent('List every .ts file under src/routes/.', {
  schema: { type: 'object', required: ['files'], properties: { files: { type: 'array', items: { type: 'string' } } } },
})
console.log(found.files)   // 直接当数组用，无需解析

schema 是 agent() 最重要的能力。它不是提示词里的建议，而是运行时契约：子 Agent 必须调用 StructuredOutput 工具提交符合 schema 的 JSON，校验失败会自动重试，超过重试上限才报错。这让 Agent 之间的数据交接变成结构化对象传递，而不是「让下一个 Agent 解析上一段自然语言」。

返回 `null` 的情形

情形	结果
用户在 `/workflows` 里跳过该 Agent	`null`
子 Agent 重试后遇终态 API 错误（如额度耗尽）	`null`
子 Agent 正常完成	文本或结构化对象
schema 校验连续失败超过上限	抛错，不是 `null`

所以脚本必须显式处理 null。样本里有三种处理方式，分别对应不同业务含义：

// 方式一：致命前置，早返回错误对象（第 108-110 行）
if (!scope) {
  return { error: "Scope agent returned no result — cannot decompose the research question." }
}
 
// 方式二：可丢弃项，转 null 后被过滤（第 208-212 行）
angle => agent(SEARCH_PROMPT(angle), {...}).then(r => {
  if (!r) return null
  return { angle: angle.label, results: r.results }
})
 
// 方式三：降级为占位对象，保留在结果里（第 256-268 行）
.then(ext => {
  if (!ext) return null        // 用户跳过 → 丢弃，不误标 unreliable
  return { url: source.url, ..., claims: [...] }
}).catch(e => {
  log("fetch failed: " + source.url + " — " + (e.message || e))
  return { url: source.url, ..., sourceQuality: "unreliable", claims: [] }
})

方式三的区分尤其关键：「Agent 没跑」（null）和「来源不可靠」（占位对象）是两件不同的事。如果把 null 也标成 unreliable，报告会把「没抓到」说成「来源质量差」。

`opts` 全字段

选项	类型	语义	样本用法
`schema`	JSON Schema	强制结构化输出	5 处调用全部使用
`label`	string	进度视图显示名	`"search:" + angle.label`
`phase`	string	显式指定进度分组	并发阶段全部使用
`model`	string	覆盖模型	未使用
`effort`	`'low'\|'medium'\|'high'\|'xhigh'\|'max'`	覆盖推理档位	未使用
`isolation`	`'worktree'`	在独立 git worktree 运行	未使用
`agentType`	string	使用自定义 subagent 类型	未使用

`model` 与 `effort`

// 机械性阶段用低档，最难验证/裁判阶段才升档
const scanned = await parallel(files.map(f => () =>
  agent(`Extract imports from ${f}`, { model: 'claude-haiku-4-5', effort: 'low' })
))
const verdict = await agent('Adjudicate these contradictory findings...', {
  effort: 'xhigh',
})

官方建议默认省略 model：继承会话模型几乎总是对的，只有确信某个阶段需要不同档位时才指定。随意切模型会让 prompt cache 前缀不匹配（模型是缓存键的一部分），还可能引入模型被组织白名单拦截后的替换行为。effort 的分工更明确：low 给便宜的机械步骤，xhigh/max 留给最难的验证和裁判。

`isolation: 'worktree'`

const results = await parallel(files.map(f => () =>
  agent(`Refactor ${f} to TypeScript`, { isolation: 'worktree', label: f })
))

在全新 git worktree 里运行，改动不影响主工作区和其他 Agent。
昂贵：每个 Agent 约 200-500ms 建 worktree 加磁盘开销，所以只在「多个 Agent 并行改文件会冲突」时用。
未产生改动的 worktree 自动删除；有改动的保留供审查。
只读任务（如样本的 Web 研究）不需要。

`agentType`

const review = await agent('Review this diff', {
  agentType: 'code-reviewer',
  schema: FINDINGS_SCHEMA,
})

从与 Agent 工具相同的注册表解析自定义 subagent 类型。可与 schema 组合：自定义 Agent 的系统提示词之后会追加结构化输出指令。适合已经有成熟自定义 Agent（如 code-reviewer、general-purpose）时复用其系统提示词。

`label` 与 `phase`

// 并发 stage 内：显式 phase，避免全局状态竞态
agent(SEARCH_PROMPT(angle), { label: "search:" + angle.label, phase: "Search", schema: SEARCH_SCHEMA })
 
// 串行代码：可以用全局 phase()
phase("Scope")
const scope = await agent("...", { label: "scope", schema: SCOPE_SCHEMA })

label 决定 /workflows 面板里能不能一眼看出谁在干什么。样本的 label 策略：搜索 Agent 用角度名、抓取 Agent 用净化后的 host、验证 Agent 用 claim 前 40 字符、综合 Agent 用 "synthesize"。

常见错误

错误	后果	正确写法
`parallel([agent(...), agent(...)])`	Agent 在 `parallel()` 调用前就启动，报 `TypeError`	`parallel([() => agent(...), () => agent(...)])`
不检查返回值直接用 `result.field`	`null` 时抛 TypeError	判空或 `.filter(Boolean)`
把 `agent()` 的文本输出用正则解析	脆弱、易被措辞变化打破	用 `schema` 拿结构化对象
无脑指定 `model`	缓存不命中、成本与行为不可控	默认省略，必要时才指定
只读任务加 `isolation`	白白付 200-500ms/Agent 的建 worktree 成本	只在并行写文件时用

`pipeline(items, ...stages)`

语义

让每个 item 独立流经所有 stage，stage 之间没有 barrier。item A 可以在 stage 3 时 item B 还在 stage 1。

// 样本第 203-272 行：两个 stage，Search → Fetch
const searchResults = await pipeline(
  scope.angles,
  angle => agent(SEARCH_PROMPT(angle), { phase: "Search", ... }),
  searchResult => parallel(novel.map(source => () => agent(FETCH_PROMPT(...), { phase: "Fetch", ... }))),
)

stage 回调签名

pipeline(items, (prevResult, originalItem, index) => { ... })

参数	含义
`prevResult`	上一个 stage 的返回值；第一个 stage 收到原始 item
`originalItem`	该 item 最初传入 `pipeline()` 的值
`index`	item 在输入数组里的下标

样本第二个 stage 用了 prevResult（searchResult）；originalItem/index 没用，因为 searchResult 里已经带了 angle 字段。如果后续 stage 需要重新拿到原始 angle 或下标做 label，用后两个参数，不要把上下文一路穿过前一个 stage 的返回值。

失败语义

某个 stage 抛错 → 该 item 落 null，跳过它剩余的 stage，其他 item 不受影响。样本 fetch stage 内部用 .catch 转成占位对象，所以正常不会触发；但 search stage 里若 agent() 抛错，对应 angle 会变 null，最终被 .filter(Boolean) 丢弃。

多 stage 写法

const results = await pipeline(
  files,
  f => agent(`Analyze ${f}`, { phase: 'Analyze', schema: ANALYSIS }),
  (analysis, file) => agent(`Fix ${file} based on: ${JSON.stringify(analysis)}`, { phase: 'Fix' }),
  (fix, file) => agent(`Verify ${file}: ${JSON.stringify(fix)}`, { phase: 'Verify', schema: VERDICT }),
)

每个 stage 只关心「上一阶段给我什么」，不需要知道全局状态。

何时不用 pipeline

只有当 stage N 真正需要 stage N-1 的全部结果时才该引入 barrier：

在昂贵的下游工作前做跨 item 去重/合并
总数为零时提前退出
stage N 的提示词要引用「其他发现」做比较

样本第 274-283 行就是后者：必须先把所有来源的 claims 汇总，才能按 importance/quality 排序并截取前 3 条。

不成立的理由：只是想 flatten/map/filter（放进 stage 做）、觉得阶段概念上分开（pipeline 就是建模这个的）、觉得 barrier 代码更整齐。官方反例：

const a = await parallel(...)
const b = transform(a)        // 没有跨 item 依赖
const c = await parallel(b.map(...))
// 中间的 transform 应该放进 pipeline stage，不该付 barrier 的延迟

`parallel(thunks)`

语义

并发执行一组 thunk，barrier：等全部完成才返回，结果数组与输入顺序一一对应。

// 样本第 233-270 行：Fetch 阶段扇出
return parallel(
  novel.map(source => () => agent(FETCH_PROMPT(source, searchResult.angle), {
    label: "fetch:" + sourceLabel, phase: "Fetch", schema: EXTRACT_SCHEMA,
  }))
)
 
// 样本第 297-306 行：Verify 内层，按票扇出
parallel(
  Array.from({ length: VOTES_PER_CLAIM }, (_, v) => () =>
    agent(VERIFY_PROMPT(claim, v), { phase: "Verify", schema: VERDICT_SCHEMA })
  )
)

thunk 要求

参数必须是返回 Promise 的函数：

parallel([() => agent('a'), () => agent('b')])   // 正确
parallel([agent('a'), agent('b')])               // 错误：agent() 已在此刻启动

后者会报 TypeError: parallel() expects an array of functions, not promises。因为 Promise 是立即执行的，直接传数组会让所有 Agent 在 parallel() 之前启动，失去并发控制和统一错误处理。

失败语义

单个 thunk 抛错 → 对应位置为 null，parallel() 本身不 reject。所以使用前要 .filter(Boolean)：

const results = (await parallel(files.map(f => () => agent(`Check ${f}`)))).filter(Boolean)

与 `pipeline` 的选择

问题	选择
我需要全部结果才能做下一步吗？	是 → `parallel`
各 item 可以独立流完所有阶段吗？	是 → `pipeline`（默认，更快）

样本里两者都用：Fetch 阶段在 pipeline 的某个 stage 内部用 parallel 扇出当前已知的小数组；Verify 用嵌套 parallel 因为必须收齐同一 claim 的全部票才能裁定。而 Search → Fetch 用 pipeline，让快的角度先进入抓取。

`phase(title)`

设置全局阶段状态，后续 agent() 归入该阶段。

phase("Scope")                       // 样本第 91 行
const scope = await agent(..., { label: "scope" })
 
phase("Synthesize")                  // 样本第 358 行
const report = await agent(..., { label: "synthesize" })

只在串行代码里安全。pipeline()/parallel() 的 stage 并发执行时会争抢这个全局状态，所以并发阶段应该给每个 agent() 传 phase: 选项：

// 样本 Search/Fetch/Verify 全部这样做
agent(prompt, { phase: "Search", ... })

同一个 phase 字符串归到同一个进度框。meta.phases 里的 title 必须与这些字符串逐字匹配；没有匹配条目的 phase() 会自己生成一个分组。

`log(message)`

在进度树上方输出一条 narrator 行。

log("Q: " + QUESTION.slice(0, 80) + (QUESTION.length > 80 ? "…" : ""))   // 第 111 行
log("Decomposed into " + scope.angles.length + " angles: " + ...)         // 第 112 行
log("\"" + claim.claim.slice(0, 50) + "…\": " + (valid.length - refuted) + "-" + refuted + ...)  // 第 319 行

样本用了 12 次，覆盖：问题摘要、角度分解、每个角度的结果数、去重统计、抓取/抽取/待验证计数、每条 claim 的票型、验证汇总、抓取失败原因。

两条纪律：

不打印大段不可信文本：样本对问题和 claim 都做 slice() 截断，避免把网页内容灌进进度视图。
打印可操作的信息：失败时打印具体 URL 和错误信息，而不是笼统的 “fetch failed”。

`workflow(nameOrRef, args?)`

内联运行另一个已保存的 workflow 作为子步骤。

// 按名字调用已保存的 workflow
const audit = await workflow('audit-routes', { paths: ['src/routes'] })
 
// 或运行指定脚本文件
const result = await workflow({ scriptPath: '/tmp/my-workflow.js' }, someArgs)

特性	说明
共享资源	子 workflow 共享本次运行的并发上限、Agent 计数器、abort signal 和 token 预算
进度展示	子 Agent 在 `/workflows` 里归到 `<name>` 分组下
预算计入	子 workflow 的 token 计入父级的 `budget.spent()`
`args` 传递	第二个参数成为子 workflow 的 `args` 全局
嵌套限制	只允许一层；在子 workflow 里再调 `workflow()` 会抛错
错误处理	未知 name、不可读 scriptPath、子脚本语法错误都会抛异常，需要 `try/catch`

样本没有用 workflow()——它是单文件自包含的。当你有多个可复用工作流（如 find-flaky-tests、audit-routes）需要组合时，用它可以避免复制粘贴。

`budget`

本轮运行的 token 预算，来自用户 +500k 这类指令。

属性	语义
`budget.total`	目标 token 数；未设置时为 `null`
`budget.spent()`	本轮已消耗 output tokens；主循环和所有 workflow 共享同一个池
`budget.remaining()`	`max(0, total - spent())`；未设置 total 时返回 `Infinity`

total 是硬上限：spent() 达到 total 后，后续 agent() 抛 WorkflowBudgetExceededError；正在飞行的 Agent 会完成并保留结果。parallel/pipeline 会把因预算被丢弃的槽位记为 null，并在日志里累计「budget dropped」。

// 动态循环：按预算决定继续还是收手
while (budget.total && budget.remaining() > 50_000) {
  const result = await agent("Find bugs in this codebase.", { schema: BUGS_SCHEMA })
  bugs.push(...result.bugs)
  log(`${bugs.length} found, ${Math.round(budget.remaining()/1000)}k remaining`)
}
 
// 静态缩放：按预算决定舰队规模
const FLEET = budget.total ? Math.floor(budget.total / 100_000) : 5

必须同时有硬迭代上限。未设置预算时 remaining() 返回 Infinity，while (budget.remaining() > 0) 会一路跑到 1000 个 Agent 的全局上限；运行时专门有一条报错文案指出这个陷阱，建议加 budget.total && 守卫或独立的计数器。

样本用固定常量（MAX_FETCH、MAX_VERIFY_CLAIMS）控制规模，所以没有用 budget。

`args`

调用时传入的输入，原样暴露为全局变量，未传时为 undefined。

// 样本第 92 行：类型守卫
const QUESTION = (typeof args === "string" && args.trim()) || ""
if (!QUESTION) {
  return { error: "No research question provided. Pass it as args: ..." }
}

三条纪律：

传真正的 JSON 值，不要传 JSON 编码后的字符串。args: ["a.ts", "b.ts"] 正确；args: "[\"a.ts\", ...]" 会让 args 在脚本里是一个字符串，args.filter/args.map 抛错。
显式类型守卫。样本只接受字符串，用 typeof 排除 undefined 和对象；数组型参数应加 Array.isArray(args)。
唯一的非确定性入口。需要时间戳、随机种子就从 args 传入；脚本内部禁止 Date.now()/Math.random()。

`meta`

必须是脚本第一条语句，且是纯字面量。字段：

字段	必填	作用
`name`	是	保存为命令后的斜杠名，也是注册表键
`description`	是	权限对话框里的一行说明
`whenToUse`	否	工作流列表里展示，给 Claude 判断何时调用
`phases`	否	进度视图分组；每项可带 `title`、`detail`，需要覆盖模型时加 `model`

export const meta = {
  name: 'workflow_research-simple',
  description: 'Deep research harness — fan-out web searches, fetch sources, adversarially verify claims, synthesize a cited report.',
  whenToUse: 'When the user wants a deep, multi-source, fact-checked research report...',
  phases: [
    { "title": "Scope", "detail": "Decompose question (from args) into 2 search angles" },
    { "title": "Search", "detail": "2 parallel WebSearch agents, one per angle" },
    { "title": "Fetch", "detail": "URL-dedup, fetch top 4 sources, extract falsifiable claims" },
    { "title": "Verify", "detail": "3-vote adversarial verification per claim (need 2/3 refutes to kill)" },
    { "title": "Synthesize", "detail": "Merge semantic dupes, rank by confidence, cite sources" },
  ],
}

约束：

纯字面量：不能有变量、函数调用、展开运算符或模板插值，否则命令会从 / 自动补全里静默消失（不报错，只是不见了）。
phases[].title 与 phase() 调用逐字匹配：标题是精确匹配的，拼写或空格不同会各生成一个分组。
phases[].model：该阶段需要特定模型覆盖时加上，与 agent({ model }) 配合。

样本的 whenToUse 是一个好范例：它把「问题太模糊时先问 2-3 个澄清问题」的策略编码进元数据，而不是留给调用者记忆。

注意样本 meta 自身有两处与实际常量不一致：Verify detail 写「3-vote / 2-of-3 refutes」，实际是 VOTES_PER_CLAIM = 2、REFUTATIONS_REQUIRED = 2；Fetch detail 写「fetch top 4 sources」，实际 MAX_FETCH = 3。meta 只是展示用元数据，运行时以脚本常量为准——这也是为什么改脚本参数时要同步检查 description/phases 的措辞。

返回值与失败语义总表

情形	行为
`agent()` 被用户跳过	返回 `null`
`agent()` 重试后遇终态 API 错误	返回 `null`
`agent({ schema })` 校验连续失败超上限	抛错
`parallel()` 中某 thunk 抛错	对应位置 `null`，整体不 reject
`pipeline()` 中某 stage 抛错	该 item 落 `null`，跳过剩余 stage
超出 token 预算	`agent()` 抛 `WorkflowBudgetExceededError`；并行原语把槽位记 `null`
达到 1000 Agent 上限	抛 `WorkflowAgentCapError`
脚本本体 `throw`	整个运行失败，显示错误和栈
脚本返回函数	报错，返回值必须可序列化

核心原则：默认把 null 当正常路径处理，把业务判定和基础设施失败分开。样本 Verify 阶段的三态裁定就是标准做法——survives（通过）、isRefuted（实质驳回）、unverified（验证 Agent 报错，无法裁决），避免把限流失败误报成「声明被推翻」。

写作检查清单

写完一个 workflow 脚本后逐条核对：

zhaoweiguo 的知识库

目录

Dynamic Workflow 源函数详解

全貌

`agent(prompt, opts?)`

基本语义

返回 `null` 的情形

`opts` 全字段

`model` 与 `effort`

`isolation: 'worktree'`

`agentType`

`label` 与 `phase`

常见错误

`pipeline(items, ...stages)`

语义

stage 回调签名

失败语义

多 stage 写法

何时不用 pipeline

`parallel(thunks)`

语义

thunk 要求

失败语义

与 `pipeline` 的选择

`phase(title)`

`log(message)`

`workflow(nameOrRef, args?)`

`budget`

`args`

`meta`

返回值与失败语义总表

写作检查清单

相关

关系图谱

目录

反向链接

zhaoweiguo 的知识库

目录

Dynamic Workflow 源函数详解

全貌

agent(prompt, opts?)

基本语义

返回 null 的情形

opts 全字段

model 与 effort

isolation: 'worktree'

agentType

label 与 phase

常见错误

pipeline(items, ...stages)

语义

stage 回调签名

失败语义

多 stage 写法

何时不用 pipeline

parallel(thunks)

语义

thunk 要求

失败语义

与 pipeline 的选择

phase(title)

log(message)

workflow(nameOrRef, args?)

budget

args

meta

返回值与失败语义总表

写作检查清单

相关

关系图谱

目录

反向链接

`agent(prompt, opts?)`

返回 `null` 的情形

`opts` 全字段

`model` 与 `effort`

`isolation: 'worktree'`

`agentType`

`label` 与 `phase`

`pipeline(items, ...stages)`

`parallel(thunks)`

与 `pipeline` 的选择

`phase(title)`

`log(message)`

`workflow(nameOrRef, args?)`

`budget`

`args`

`meta`