taskflow

流程定义

把阶段包成一个 taskflow 的顶层字段。

流程定义是一个 JSON 对象,声明一个多阶段工作流:它的名字、它接受的参数、它该多激进地并行、它被允许花多少钱,以及构成其 DAG 的有序阶段数组。每个 taskflow —— 无论保存到磁盘还是作为一行内联传入 —— 都会先被反糖化为同一形状,再由运行时校验和执行。

你可能在想:这些字段我都得用吗? 不用。结构上只有 phases 是必需的(保存的 flow 还要 name)。其余字段都是行为调参 —— 并发、成本上限、缓存默认值、严格度。本页按你实际会需要它们的顺序逐一介绍,详尽的参考材料放在最后。

快速心智模型

一个 flow 只是个容器。它装着元数据、几个执行调参,以及一个阶段列表。阶段干真正的活;容器决定它们怎么跑。

最小形状
{
  "name": "my-flow",
  "phases": [
    { "id": "do", "type": "agent", "task": "...", "final": true }
  ]
}

下面的一切都是在这个形状上做加法。运行时在生成任何子代理之前先校验整个对象 —— 非法 flow 永远不会花你一个 token。

一个贯穿始终的示例

我们会逐步构建一个真实的 flow —— 审计一个目录的路由是否缺认证检查 —— 一块一块来。每一节加一个顶层字段并解释你为什么会需要它。完整的 flow 出现在本页末尾。

名字与描述

每个保存的 flow 都需要一个 name。它会成为 /tf:<name> 命令别名和运行状态目录名。description 是可选的,但值得加 —— 它会出现在 /tf list 和运行记录里。

给 flow 命名
{
  "name": "audit-auth",
  "description": "Audit routes for missing auth checks",
  "phases": [/* ... */]
}

缺失或为空的 name 是一个硬校验错误(Missing or invalid 'name'),在任何阶段校验之前就会检查。省略 name 的内联简写 flow 会得到一个合成的回退名("task""parallel""chain")。

声明你的参数:args

大多数可复用的 flow 都接受参数 —— 一个目录、一个严重程度阈值、一个 feature flag。在 args 里声明它们,运行时就能应用默认值并给它们做文档。每个 arg 是一个对象,带可选的 defaultdescription 和建议性的 required 标志。

声明带默认值的 args
{
  "name": "audit",
  "args": {
    "dir": { "default": "src/routes", "description": "Directory to audit" },
    "force": { "default": false },
    "severity": { "default": "high", "required": true }
  },
  "phases": [/* ... */]
}

在调用时传参:

带参数运行
/tf run audit dir=src/api force=true

然后在任意阶段里用 {args.<name>} 插值:

插值一个 arg
{
  "id": "scan",
  "type": "agent",
  "task": "Scan {args.dir} for missing auth checks."
}

声明 args 是可选的。运行时传入的未声明参数仍可通过 {args.X} 访问 —— 只是没有默认值。声明它们能给你默认值、文档,以及更清晰的校验信息。

你可能在想:如果一个阶段引用了 {args.X},但调用方没传 X 也没有默认值,会怎样?默认情况下占位符保持字面值,运行时发一条警告。在 strictInterpolation: true 下(见安全网),那条警告会变成硬错误 —— 对生产 flow 很有用,因为字面的 {args.X} 漏进 prompt 是个 bug,不是个意外。

控制并行度:concurrency

concurrency 是整个 flow 中同时跑多少个子代理的默认上限。默认值 8mapparallel 阶段会继承它,除非它们设了自己的阶段级 concurrency

flow 级 vs 阶段级并发
{
  "name": "fanout",
  "concurrency": 16,
  "phases": [
    {
      "id": "scan",
      "type": "map",
      "over": "{steps.discover.json}",
      "task": "...",
      "concurrency": 4
    }
  ]
}

上面阶段级的 concurrency: 4 只对该阶段覆盖 flow 级的 16

给成本封顶:budget

有些 flow 是开放式的 —— 一个深度调研 loop、一个有很多变体的 tournament。budget 让你设一个运行级的 USD 成本和/或 token 用量上限。任一超出时,剩余阶段被跳过,运行以 blocked 状态结束。

设置预算
{
  "name": "capped-run",
  "budget": {
    "maxUSD": 2.0,
    "maxTokens": 2000000
  },
  "phases": [/* ... */]
}

两个字段都可选 —— 设其一或都设。已经在跑的子代理不会在调用中途被中断;上限是在阶段之间检查的。0 的预算会被当作"立即超出",只对干跑结构检查有用。

agent 从哪来?agentScope

阶段按名字引用 agent("agent": "security-reviewer")。运行时在三种作用域之一里发现 agent 定义(带 YAML frontmatter 的 .md 文件):

取值搜索路径冲突规则
"user"(默认)~/.pi/agent/agents/*.md
"project".pi/agents/*.md
"both"user,然后 project名称冲突时 project 优先
同时用两个作用域
{
  "name": "team-flow",
  "agentScope": "both",
  "phases": [/* ... */]
}

当你有共享的用户级 agent、但想让某个项目覆盖其中几个时,用 "both"

安全网:strictInterpolation

默认情况下,taskflow 对非致命问题是宽松的 —— 它警告并继续。打开 strictInterpolation 把这些警告提升为硬错误。对保存的、生产用的 flow 这是对的取舍:它能在一次运行浪费 token 之前抓住静默的占位符泄漏和过时字段引用。

开启严格模式
{
  "name": "production-flow",
  "strictInterpolation": true,
  "phases": [/* ... */]
}

strictInterpolation 控制那个硬性的 {steps.X} 可达性检查 —— 那始终是硬错误。它只把警告提升为错误。

始终是硬错误的部分(与该标志无关):

  • {steps.X.*}X 无法经 dependsOn/from 到达(阶段会和它的生产者并行跑,看到字面占位符)。
  • 阶段引用自身输出(loop 阶段除外,它在 until 里合法地这样做)。
  • 未知的 dependsOn / from 目标、依赖环、重复的阶段 id、多于一个 final: true

仅在 strictInterpolation: true 下才变成硬错误的部分:

警告默认strict
引用了 {args.X} 但未提供(且无默认值)警告错误
调用 cwdargs.codebase 不匹配警告错误
仅 loop 用的字段(untilmaxIterationsconvergence)出现在非 loop 阶段警告(忽略)错误
idempotent: falseapproval / flow 上是空操作警告错误
idempotent: false 覆盖 cache.scope: "cross-run"警告错误
gate 上同时设了 evalscore警告错误

跨运行复用工作:incremental

默认情况下,每次运行都从零开始 —— 每个阶段都重新执行,哪怕什么都没变。设 incremental: true 把每个阶段默认为跨运行缓存(cache.scope: "cross-run"),这样重新跑 flow 时会跨运行和跨会话复用未变阶段的结果。

增量 flow
{
  "name": "ci-audit",
  "incremental": true,
  "phases": [
    { "id": "lint", "type": "script", "run": "npm run lint" },
    { "id": "review", "type": "agent", "task": "Review recent changes." }
  ]
}

关于 incremental 与阶段级设置如何交互,有几条规则:

  • 它等价于在每个阶段上设 cache: { scope: "cross-run" }
  • 阶段自己的 cache 设置优先 —— cache.scope: "off""run-only" 仍被尊重。
  • 被跨运行阻止的阶段类型(gateapprovallooptournamentscript)即使在 incremental 下也永不跨运行缓存 —— 它们每次运行都必须产出新结果。在上面的例子里,lint 尽管 incremental 仍每次调用都重跑;review 会被缓存。
  • idempotent: false 的阶段完全排除在缓存之外。在 incremental 下这会产生一条警告,让你知道该阶段不会被复用。

运行时的 incremental 参数(调用时传入)会覆盖 flow 声明的值,让你无需改 flow 就能强制一次全新或增量运行。

在阶段间共享上下文:contextSharing

当多个阶段读同样的文件时,你可以用一个标志把它们统统纳入共享上下文树。contextSharing: true 是在每个阶段上设 shareContext: true 的简写 —— 每个阶段的子代理会得到 ctx_read/ctx_write(一个与兄弟和祖先共享的 blackboard)以及 ctx_report/ctx_spawn(向上汇报 + 排队子任务)。个别阶段仍可用 shareContext: false 退出。

开启上下文共享
{
  "name": "collab-flow",
  "contextSharing": true,
  "phases": [
    { "id": "explore", "type": "agent", "task": "..." },
    { "id": "synthesize", "type": "agent", "task": "...", "dependsOn": ["explore"] }
  ]
}

上下文共享激活时,阶段 id 成为文件系统节点 id,必须匹配 [A-Za-z0-9._-]+。两个清理后相同的 id 会静默合并它们的 blackboard —— 这作为硬错误强制执行。

phases 数组

phases 是唯一必需的字段 —— 一个有序的阶段定义数组(minItems: 1)。每个阶段有唯一 id、一个 type(默认 "agent"),并通过 dependsOn 构成 DAG 边。没有依赖的阶段在第一拓扑层运行。

如果没有阶段标记 final: true,数组顺序中的最后一个阶段成为工作流结果。如果多于一个标记 final,校验失败。

运行时通过 Kahn 算法计算拓扑层:同层阶段并发运行(至多 concurrency 个)。一个阶段在它所有 dependsOn 完成后启动 —— 或在 join: "any" 下任一完成后启动。

每种阶段类型的每个字段,见阶段类型页。

简写:简单工作跳过 DAG

对一次性委派,你不需要写一个完整的 phases DAG。taskflow 接受三种简写形状,运行时的 desugar() 会在校验前把它们展开成完整 flow。当一个 spec 没有 phases 数组、tasktaskschain 之一时,它被识别为简写。

单个任务

单个 task 简写
{ "task": "Summarize src/", "agent": "explorer" }

反糖化为一个 agent 阶段(id "main"final: true)。可选的 agentcontextcontextLimit 会透传。

并行任务

并行 tasks 简写
{
  "tasks": [
    { "task": "Audit auth in src/api", "agent": "analyst" },
    { "task": "Audit validation in src/api", "agent": "analyst" }
  ]
}

反糖化为一个 parallel 阶段(id "parallel"final: true)。context 在所有分支间共享:spec 级 context 加上各步 context 数组的并集会被合并、只读一次。

顺序链

chain 简写
{
  "chain": [
    { "task": "Find all TODO comments" },
    { "task": "Summarize them: {previous.output}" },
    { "task": "Open issues for: {previous.output}" }
  ]
}

反糖化为顺序的 agent 阶段(step1step2、……),每个依赖前一个;最后一个是 final: true。每一步用 {previous.output} 引用上一步的输出。

chain 模式下,顶层的 context / contextLimit 会被忽略(运行时会警告)。把它们放在各个步骤上 —— chain 模式下的 flow 级 context 是刻意不支持的。

简写透传什么

三种形式都会把这些可选的顶层字段透传到反糖化后的 flow:name(默认为 "task""parallel""chain")、descriptionconcurrencyagentScopeargsbudgetstrictInterpolation

简写很适合做实验。当你需要条件路由、循环、gate、缓存或跨会话续跑时,再迁移到完整的 phases DAG。

完整示例

这是把我们讨论过的所有内容都用上的 audit-auth flow:

audit-auth — 完整版
{
  "name": "audit-auth",
  "description": "Audit routes for missing auth checks",
  "version": 2,
  "args": {
    "dir": { "default": "src/routes", "description": "Directory to audit" },
    "severity": { "default": "high", "required": true }
  },
  "concurrency": 8,
  "agentScope": "both",
  "strictInterpolation": true,
  "incremental": false,
  "budget": {
    "maxUSD": 2.0,
    "maxTokens": 2000000
  },
  "phases": [
    {
      "id": "discover",
      "type": "agent",
      "agent": "scout",
      "task": "List every route handler in {args.dir} as JSON.",
      "output": "json"
    },
    {
      "id": "audit-each",
      "type": "map",
      "over": "{steps.discover.json}",
      "as": "route",
      "agent": "security-reviewer",
      "task": "Does {route.path} require auth? Output {\"ok\": true|false}.",
      "output": "json",
      "dependsOn": ["discover"]
    },
    {
      "id": "report",
      "type": "reduce",
      "from": ["audit-each"],
      "agent": "writer",
      "task": "Summarize the audit results: {steps.audit-each.output}",
      "final": true,
      "dependsOn": ["audit-each"]
    }
  ]
}

字段参考

本页剩下的是详尽参考 —— 每个顶层字段、ArgSpec 形状,以及完整的校验表。需要精确答案时来查。

顶层字段

字段类型默认值必需说明
namestringminLength: 1(保存的 flow)工作流标识符。成为 /tf:<name> 和运行状态目录名。
descriptionstring人类可读摘要。出现在 /tf list 和运行记录中。
versionnumber1Schema/flow 版本标签,供作者追踪。不强制。
argsRecord<string, ArgSpec>带默认值的已声明调用参数。见 ArgSpec
concurrencynumber8默认最大并发子代理数。被 map/parallel 阶段继承。
budgetBudget运行级成本/token 上限。见给成本封顶:budget
agentScope"user" | "project" | "both""user"在何处发现 agent。
strictInterpolationbooleanfalse把校验警告提升为硬错误。
contextSharingbooleanfalse为所有阶段启用共享上下文树。
incrementalbooleanfalse把每个阶段默认为跨运行缓存。
phasesPhase[]minItems: 1有序阶段定义。见阶段类型

ArgSpec

args 中的每个条目是一个带这些字段的对象:

字段类型默认值说明
defaultunknown运行时未提供该参数时使用的值。类型不被 schema 强制 —— 插值需谨慎。
descriptionstring人类可读的参数文档。
requiredboolean作者声明的必需标志(建议性;缺失时运行时应用 default)。

校验行为

每个 flow 在执行前都会被校验,有两种模式:

  1. 静态/tf verify,无运行时参数):仅结构检查 —— schema 形状、DAG 完整性、环检测、{steps.X} 引用的可达性。零 token。
  2. 运行时(运行启动时,带已解析的参数):以上全部,外加参数缺失警告和 cwd/codebase 不匹配警告。

硬错误(始终让运行失败)

错误原因
Missing or invalid 'name'name 缺失、为空、或不是字符串。
Taskflow must have at least one phasephases 缺失或为空。
Duplicate phase id: X两个阶段共用一个 id
Phase 'X': unknown type 'Y'type 不在 10 个允许值里。
Phase 'X': dependsOn unknown phase 'Y'dependsOnfrom 条目没有对应阶段。
Dependency cycle detected: A -> B -> AKahn 算法发现环。
Only one phase may be marked 'final' (found N)多于一个 final: true 阶段。
Phase 'X': task references {steps.Y.*} but 'Y' is not reachable via dependsOn一个 {steps.Y} 引用中 Y 不是传递祖先(除非 join: "any")。
Phase 'X': references its own output via {steps.X.*}自引用(loop 阶段除外)。
Phase 'X': id uses underscores阶段 id 含 _(应用连字符)。
Phase 'X': agent name 'Y' uses underscoresagent 名含 _
Phase 'X': agent 'Y' has invalid name formatagent 不匹配 /^[a-z][a-z0-9-]*$/
Phase 'X': ids used with context sharing must match [A-Za-z0-9._-]+上下文共享的阶段 id 字符集非法。

警告(在 strictInterpolation 下提升为错误)

警告说明
引用了 {args.X} 但未提供无默认值,无运行时值。占位符保持字面。
调用 cwdargs.codebase依赖 cwd 的 agent 可能查错仓库。
非 loop 阶段上的 loop 专用字段until / maxIterations / convergence 被静默忽略。
idempotent: falseapproval / flow 上是空操作这些阶段不跑子代理;该标志无效。
idempotent: false 覆盖 cache.scope: "cross-run"有副作用的阶段永不被缓存。
incremental 下的 idempotent: false阶段排除在缓存外,每次调用都重跑。
gate 上同时设 evalscoreeval 先运行;通常一个就够。

尽早、经常跑 /tf verify。它能在零成本下抓住上面每一个硬错误,在任何子代理被生成之前。

isShorthand 识别

当一个 spec 没有 phases 数组、至少满足下列之一时,被当作简写处理:

  • chain 是非空数组
  • tasks 是非空数组
  • task 是字符串

否则它作为完整 flow 按 TaskflowSchema 校验。

下一步

Last updated on

这页内容对你有帮助吗?

帮助我们改进文档,或在社区中提问。

On this page