预算 | taskflow

你把一个 map 阶段指向你的仓库，让一个 scout agent 列出变更的文件。scout 返回 500 个。每一个都生成一个安全审查器。四个同时运行。等你注意到时，你已在一个本该花四毛钱的 flow 上花了十二美元。

budget 就是防止这件事的字段。它是对累计成本或 token 的运行级上限。当超出上限时，taskflow 中止运行，保留此前产生的所有部分输出，并返回一个诊断。把它设在任何可能无界 fan-out 的 flow 上 —— map、parallel、loop、tournament —— 它就成了一个被错误发现的数组与失控支出之间唯一的防线。

预算执行不花费额外 token。它读取每个子代理已经写入的 usage 记录，并在每次重试、每个 fan-out 项、每次 DAG 层切换时检查。

在每个 fan-out flow 上设置预算

如果你的 flow 包含 map、parallel、loop 或 tournament，加一个 budget。这就是全部规则。

budget-basic.json

{
  "name": "summarize-files",
  "budget": { "maxUSD": 1.5 },
  "concurrency": 4,
  "phases": [
    {
      "id": "discover",
      "type": "agent",
      "output": "json",
      "task": "List files under {args.dir}. Output a JSON array of { path }."
    },
    {
      "id": "summarize",
      "type": "map",
      "over": "{steps.discover.json}",
      "as": "file",
      "concurrency": 8,
      "task": "Read {file.path} and give a one-sentence summary."
    }
  ]
}

两个维度都可选。maxUSD 限制累计成本；maxTokens 限制累计输入 + 输出 token。设其一或两者。省略一个维度即关闭该上限。

budget-both-dimensions.json

{
  "name": "fan-out-audit",
  "budget": {
    "maxUSD": 2.0,
    "maxTokens": 2000000
  },
  "phases": [ /* ... */ ]
}

将 maxUSD 与 maxTokens 配对以纵深防御。一次模型切换或定价变化可能把花费从成本转移到 token（或反之）。同时设置两者可防止任一上限被击穿。

schema 拒绝未知键（additionalProperties: false）。空的 budget: {} 合法但是空操作 —— maxUSD / maxTokens 至少应设置一个。

触及上限时会发生什么

累计花费超出上限的那一刻，运行时接管。不再生成新阶段。进行中的 fan-out 停止生成项。运行以 blocked 结束 —— 不是 failed —— 并且超限前产生的每个部分输出都被保留。

层	发生什么
DAG 层	剩余阶段被记录为 `skipped`，带 `error: "Budget exceeded: ..."`。
进行中的 fan-out（`map` / `parallel`）	循环停止生成。未开始的项目成为 `stopReason: "budget-skipped"`，零 usage。
`loop` 迭代	下一次迭代不开始。reflexion loop 将此视为硬停止 —— 失败不作为 reflexion 输入反馈回去。
`tournament` 评委	评委被跳过；第一个存活变体获胜，带一条警告。
重试	重试循环中断 —— 不再尝试。
运行状态	`status` 变为 `"blocked"`。`finalOutput` 为 `"Budget exceeded — run halted."` 加原因和最终阶段的部分输出。

被阻止的运行不是失败的运行。在超限前完成的每个阶段都保留其 output、json 和 usage。finalOutput 由实际产生输出的最后一个阶段组合而成 —— 因此一个预算被阻止的运行仍可向你的对话返回有用的部分工作。

blocked 表示预算或 gate 中止了工作流。failed 表示一个非 optional 阶段报错退出。检查 state.status 和逐阶段的 error / budgetTruncated 字段以区分两者。

比较是严格的

检查用严格 >，不是 >=。恰好落在上限上的运行不被阻止 —— 只有超过它的下一个美分或 token 才触发中止。maxUSD: 2.0 的预算允许成本恰好为 $2.000 的运行。

有界超支

在单个 DAG 层内，阶段并发运行至多到 concurrency（默认 8）。预算检查在每个阶段完成后触发，因此同一层中至多有 concurrency - 1 个额外阶段可能在超限被检测到之前就开始。

这种有界超支是设计使然 —— 一旦 budgetBlocked 被设置，它阻止级联到后续层，但无法撤销已经在途的花费。如果你需要零超支的硬上限，将 flow 的 concurrency 设为 1。阶段随后严格顺序运行，守卫在每个开始前重新评估。

各阶段类型的成本如何累计

预算核算读取 state.phases[*].usage，不同阶段类型对其填充方式不同：

阶段类型	计入预算的内容
`agent`	子代理在每次尝试上的 `usage`。失败重试在重试决策前被累加。
`parallel`	每个分支 `usage` 之和。
`map`	每个已开始项目 `usage` 之和。预算跳过的项贡献零。
`loop`	每次迭代累计的 `usage`，在循环中折入（而非覆盖）—— 因此 reflexion loop 无法在迭代之间不知不觉地超过上限。
`tournament`	每个变体的 `usage` 加上评委的（当评委运行时）。
`flow`	子 flow 运行其自身预算（见下方）。父阶段记录子 flow 的累计 usage。
`gate`	gate 子代理的 `usage`。
`approval`	零 —— 不运行子代理。
`script`	零 —— 无 LLM token。
`reduce`	reducer 子代理的 `usage`。

缓存命中（cacheHit: "cross-run" | "run-only"）报告缓存的 usage 但本次运行未花费新 token。cacheRead/cacheWrite/contextTokens 不计入 token 上限 —— 缓存读取几乎免费，contextTokens 是时点指标，不可累加。

重试与预算

每次重试尝试在重试决策之前将其 usage 推入累加器。因此一个失败三次才成功的阶段花费了全部三次尝试的 token —— 且第三次尝试本身可能将运行推过上限。

失败的重试确实计入预算。失败尝试没有"退款" —— 提供方已经对其计费。

声明为 idempotent: false 的阶段禁用隐式瞬态重试（运行时安全网），因此一个不稳定的提供方错误不会通过重跑副作用而静默双重计费。但显式 retry: { max } 仍被遵守 —— 且每次显式重试仍计入预算。

idempotent-false-deploy.json

{
  "id": "deploy",
  "type": "agent",
  "idempotent": false,
  "task": "Run the production deploy. Report the rollout ID.",
  "retry": { "max": 0 }
}

子流程中的预算

flow 阶段运行子 taskflow。子流程的预算如何与父预算交互，取决于子流程是内联生成（def）还是按名加载（use）。

内联 `def` 子流程 —— 硬钳制

生成的内联定义不能抬高父上限。运行时每个维度取 min(child, parent)：

subflow-def-clamped.json

{
  "id": "plan-and-execute",
  "type": "flow",
  "def": {
    "name": "inner-plan",
    "budget": { "maxUSD": 5.0 },
    "phases": [ /* ... */ ]
  }
}

如果父 flow 有 budget: { maxUSD: 2.0 }，内层子流程的有效上限变为 $2.0，无论其声明的 $5.0。它只能比父更紧。如果钳制后没有有限维度剩下，子流程的 budget 被设为 undefined —— 但父的守卫仍能看到子流程的花费，因为父阶段记录了累计的子阶段 usage。

已保存 `use` 子流程 —— 软继承

按 use 加载的子流程执行其自身声明的预算。如果未声明，它继承父上限作为软性每流上限。

此继承是尽力而为：花费不跨 flow 边界。父已花费总额不从子继承的上限中扣除，因此继承 $2.0 的子流程即使父已花费 $1.50，自身仍可花费至多 $2.0。父自身的守卫在子流程返回时仍会捕获合并总额。

递归被栈检查阻止，因此子流程不能无限嵌套以绕过自身预算。嵌套计数器在 flow 与内联 spawn（ctx_spawn）两条递归轴上统一。

在 fan out 前估算

对昂贵的逐项工作，先运行一个廉价的估算器阶段，让其输出决定是否继续。这把无界的 map 转为有界的。

map-with-per-item-estimation.json

{
  "name": "bounded-review",
  "budget": { "maxUSD": 3.0, "maxTokens": 3000000 },
  "concurrency": 4,
  "phases": [
    {
      "id": "discover",
      "type": "agent",
      "output": "json",
      "task": "List files under src/. Output JSON: { items: [{ path }] }."
    },
    {
      "id": "estimate",
      "type": "agent",
      "output": "json",
      "context": ["{steps.discover.json}"],
      "task": "Given these files, estimate tokens to review each. Output JSON: { items: [{ path, estTokens }], totalEstTokens }.",
      "expect": {
        "type": "object",
        "required": ["items", "totalEstTokens"],
        "properties": {
          "items": { "type": "array", "items": { "type": "object" } },
          "totalEstTokens": { "type": "number" }
        }
      }
    },
    {
      "id": "review",
      "type": "map",
      "over": "{steps.estimate.json.items}",
      "as": "item",
      "when": "{steps.estimate.json.totalEstTokens} < 2500000",
      "concurrency": 6,
      "task": "Review {item.path} for security risks. List concrete findings."
    },
    {
      "id": "fallback",
      "type": "agent",
      "when": "{steps.estimate.json.totalEstTokens} >= 2500000",
      "task": "The estimated token cost ({steps.estimate.json.totalEstTokens}) exceeds half the budget. Summarize src/ at a high level instead of per-file review."
    }
  ]
}

review 上的 when 守卫在估算超过预算一半时完全跳过 fan-out，为 fallback 阶段留出余量。即使有此守卫，budget 仍能防止低估。

给 loop 和 tournament 各自的余量

reflexion loop 可能在精炼同一输出上花费多次迭代。tournament 运行 N 个变体加一个评委。将预算设得足够覆盖预期的迭代或变体数，再让守卫捕获失控情况。

tournament-with-budget.json

{
  "name": "pick-best-title",
  "budget": { "maxUSD": 0.80 },
  "phases": [
    {
      "id": "compete",
      "type": "tournament",
      "variants": 4,
      "task": "Write a title for: {args.topic}.",
      "judge": "Pick the single best title. Output only the winner."
    }
  ]
}

检查预算用量

持久化到 .pi/taskflows/runs/<flow>/<runId>.json 的每个运行记录都暴露了审计花费所需的信息：

字段	告诉你什么
`def.budget`	生效的上限（`{ maxUSD?, maxTokens? }`）。
`status`	超出预算时为 `"blocked"`；否则为 `"completed"` / `"failed"` / `"paused"`。
`phases.<id>.usage`	该阶段的 `{ input, output, cacheRead, cacheWrite, cost, contextTokens, turns }`。`input + output` 是 token 上限所汇总的。
`phases.<id>.budgetTruncated`	当 `map` / `parallel` fan-out 被上限（或 `MAX_DYNAMIC_MAP_ITEMS` 安全限制）截断时为 `true`。
`phases.<id>.attempts`	包括重试在内的子代理总尝试次数。`> 1` 时表示发生了重试 —— 且每次尝试的花费都在 `usage` 中。
`phases.<id>.cacheHit`	由缓存提供时为 `"cross-run"` 或 `"run-only"`（本次运行无新 token）。
`phases.<id>.status`	从未运行的阶段为 `"skipped"`，带 `error: "Budget exceeded: ..."`。

预算被阻止运行的终态 finalOutput 为：

blocked-final-output

Budget exceeded — run halted.
Reason: cost $2.143 exceeded cap $2.000

<final phase's partial output>

要获得运行级总额而无需自己汇总阶段，读取 executeTaskflow 返回的值 —— result.totalUsage 是跨所有阶段的运行后累计。运行记录不存储顶层 totalUsage；TUI 在读取时计算。

常见错误

错误	原因	修复
`Budget exceeded — run halted. Reason: cost $X exceeded cap $Y`	累计 `cost` 超过 `maxUSD`。	提高 `maxUSD`、减少 fan-out（`concurrency` / `over` 大小），或加 `when` 守卫跳过昂贵阶段。
`Budget exceeded — run halted. Reason: tokens N exceeded cap M`	累计 `input + output` 超过 `maxTokens`。	提高 `maxTokens`、切换更廉价的 `model`，或预读更少 context（`contextLimit`）。
`N item(s) skipped: budget exceeded`	`map` / `parallel` fan-out 被截断。	在运行记录中检查 `phases.<id>.budgetTruncated`。提高预算或降低 `concurrency`。
`judge skipped: run aborted or budget exceeded`	`tournament` 在超限后到达评委步骤。	第一个存活变体被选为赢家。提高预算或减少 `variants`。
运行状态 `blocked` 但 `finalOutput` 看似完整	最终阶段在超限前完成，但更后的（非最终）阶段把总额推过了上限。	重排让昂贵阶段在最终阶段之前运行，或标记一个更廉价的阶段为 `final: true`。

blocked 运行不是 failed 运行。blocked 表示预算或 gate 中止了工作流；failed 表示非可选阶段报错退出。检查 state.status 和逐阶段的 error / budgetTruncated 字段以区分两者。

Last updated on

预算