Workflow 製作手冊 · workflow.json

寫一條 workflow,這裡有你需要的全部語法 每種都有動畫 + sample JSON · 標明現行 / 逃生口走 run.py

workflow.json 是非圖靈完備的宣告資料:你描述一張流程圖,受信任的直譯器照著跑。界線是 整張圖能不能事先靜態畫出來 —— 畫得出來就進 DSL,畫不出來交給 run.py。

現行 · dsl.py 今天可跑(#428 六語法已全數落地) 無法宣告 · 走 run.py
0

怎麼看

兩種輪廓先分清楚

展開 = 每個元素跑同一條線,再收斂

流水線依序各跑一遍,結果 fan-in 收回

分支 = 只走一條

一條亮、其他灰掉是死路

確定性節點 AI 節點 gate / switch 岔路 資料流 死路
1

一條 workflow 的骨架

最外層長什麼樣

一個 workflow.json 就是一個 WorkflowDef:頂層的 steps 依序執行,config 提供設定、phases 是給前端畫進度的骨架。檔名即 id 的權威。

workflow.json · 頂層
{
  "id": "throughput-check",          // 唯一識別(workspace 檔名即權威)
  "title": "Throughput 退化檢查",
  "phases": [                        // 進度骨架,前端畫「1/3 · analyze」
    {"name":"detect"}, {"name":"analyze"}, {"name":"review"}
  ],
  "config": { "threshold": 0.1 },    // {config.threshold} 引用
  "steps": [ /* 頂層步驟,依序執行(見以下各節) */ ]
}
id唯一識別。workspace 檔遮蔽同 id 的 package 檔。現行
title / description / tag / hintRun picker 顯示用的中繼資料。現行
phases[]進度骨架。每個 step 標 phase 歸到某一格,前端據此畫進度條與「12/20 · 1 failed」。現行
config{}這條 workflow 的設定,用 {config.x} 引用。現行
steps[]頂層步驟序列,依序跑。現行
2

節點型別

六種積木
agentLLM 一回合。out 寫內容檔;outputs 宣告具名欄位;否則需 check現行
sandbox確定性指令,無 LLM。純計算、無憑證(可靠副作用走 capability)。現行
gate人工閘。approve 續、reject 終止;revise 帶 user 回饋 {steps.<gate>.feedback} 打回某步重做。現行
capability可靠且冪等的副作用:ingest_to_collection / upsert_context_card現行
map唯一迴圈。over glob;over 清單 / range。do 是元素內序列。現行
switch有界條件分支。cases 預先列舉。現行
3

值怎麼進、怎麼流

一種語法定址任何值

值從兩處進來{config.x}(workflow 設定)、{inputs.y}(觸發時傳入)。值在 step 之間流動{item}(map 當下元素)、{steps.名.欄位}(某步的具名產出)。全部同一種寫法,插值只定址、無 eval。

具名產出 {steps.x.field}

現行

表面是變數、骨子是檔案

重點:step 有名字,產出用 {steps.名.欄位} 引用;跟 {config.x} 同一種寫法。
workflow.json
{ "type":"agent", "name":"classify", "phase":"classify",
  "prompt":"判斷異常類型,輸出 JSON。",
  "outputs":{ "type":{"type":"str","enum":["latency","errors","other"]} } }

// 之後任何一步都能引用,validate_def 靜態檢查欄位存在
"on": "{steps.classify.type}"

骨子裡=讀 .workflow/<id>/step_classify/main.jsonresult.fields.type(既有 journal 條目,不另立新檔)。

outputs schema + 靜態驗證

現行

型別(地基)+ enum(選配)

重點:型別是地基、enum 是選配;enum 讓 switch 的 cases 被檢查窮盡性。
workflow.json
"outputs": {
  "type": { "type":"str", "enum":["latency","errors","other"] },
  "score": "float"
}
// 型別:str|int|float|bool|list|obj(地基)
// 進來的值:{config.x}、{inputs.y}(觸發時傳入)
4

展開 map

唯一的迴圈 · 依序 → fan-in

安全線:單層展開;寬度可執行期才知道,不准巢狀遞迴。

over glob

現行

集合=符合 glob 的檔案數

重點:現狀 fan-out=檔案驅動,符合 glob 的檔有幾個就展開幾個。

把 30 個 md 檔各翻一遍。

workflow.json
{ "type":"map", "name":"tr", "over":"src/**/*.md",
  "as":"f", "phase":"translate",
  "do":[ {"type":"agent","phase":"translate",
          "prompt":"翻 {f}","out":"out/{f}.en.md"} ] }

do = 元素內流水線

現行

do 是一串 step,每元素依序跑完

重點:do 是 list[Step],可混 sandbox/agent/capability 串成鏈;每元素跑完整條。

每個檔:前處理 → AI 摘要 → 寫回卡片。

workflow.json
{ "type":"map", "over":"docs/*.txt", "as":"d", "phase":"proc",
  "do":[
    {"type":"sandbox","phase":"proc","run":"python clean.py {d}"},
    {"type":"agent","phase":"proc","prompt":"摘要 {d}","out":"sum/{d}.md"},
    {"type":"capability","call":"upsert_context_card","phase":"proc",
     "collection":"kb","keys":["{d}"],"title":"{d}","body":"..."}
  ] }

do 內不可放 mapgate(Q7,validate_def 擋)—— 這就是「不准巢狀遞迴」與「gate 只能在頂層」。

over 清單值

現行

集合=上一步的具名清單(免先落檔)

重點:over 直接吃清單值,引擎背後物化成檔維持元素身份。

AI 抽出所有待辦,每項各開一張卡。

workflow.json
{ "type":"agent", "name":"extract", "phase":"extract",
  "prompt":"抽出所有待辦", "outputs":{"items":"list"} },

{ "type":"map", "over":"{steps.extract.items}", "as":"t", "phase":"card",
  "do":[ {"type":"capability","call":"upsert_context_card","phase":"card",
          "collection":"tasks","keys":["{t.id}"],
          "title":"{t.title}","body":"{t.body}"} ] }

over range

現行

集合=一個數字(跑 N 次)

重點:over 吃一個數字,展開 N 次;不用手動生 N 個檔。

產生 N 個變體各存一份。

workflow.json
{ "type":"map", "over":{"range":"{inputs.n}"},
  "as":"i", "phase":"gen",
  "do":[ {"type":"agent","phase":"gen",
          "prompt":"產生第 {i} 個變體","out":"v/{i}.md"} ] }

fan-in 收斂

現行

整批輸出= {steps.map.outputs}

重點:fan-in 從暗規則(自己 glob)變一級引用 {steps.tr.outputs}。
workflow.json
{ "type":"sandbox", "phase":"collect",
  "run":"python index.py --files {steps.tr.outputs}" }

另一種收法:也可以 glob "out/*" 收回,只是要自己記得檔落在哪;.outputs 讓引擎替你記住每個元素的身份。

skip + collect

現行

一顆老鼠屎不壞一鍋

重點:單一元素失敗只 skip、收進 failures[],其餘照跑。

30 檔翻譯,3 檔失敗 → 收 27 個成功 + 一份失敗清單。

行為(無需額外語法)
// map 自帶 skip+collect:
// 失敗元素 → run.failures[](key + error),不炸整批
// 前端顯示 "27/30 · 3 failed"
5

分支 switch

只走一條 · cases 預先列舉

安全線:分支路徑事先宣告、數量有限;不准無界迴圈。

switch on 設定 / 輸入

現行

條件跑前已知

重點:on 讀設定/輸入值(現行插值即可),switch 選一條。

格式是 md 走 A、csv 走 B。

workflow.json
{ "type":"switch", "on":"{inputs.kind}", "phase":"route",
  "cases":{
    "md": [ {"type":"agent","phase":"route","prompt":"...","out":"o.md"} ],
    "csv":[ {"type":"sandbox","phase":"route","run":"python csv.py"} ]
  }, "default":[] }

switch on 計算值

現行

條件=程式算出的具名欄位

重點:sandbox 產出具名 bool,switch on 它 —— 引用模型讓這乾淨。

throughput 掉超過 10% 才進調查。

workflow.json
{ "type":"sandbox", "name":"measure", "phase":"measure",
  "run":"python measure.py", "outputs":{"dropped":"bool"} },

{ "type":"switch", "on":"{steps.measure.dropped}", "phase":"route",
  "cases":{ "true":[ {"type":"gate","phase":"route","title":"要深入?"},
                     {"type":"agent","phase":"route","prompt":"分析原因","out":"rca.md"} ] },
  "default":[] }

switch on AI 判斷

現行

AI 選一條,路徑仍預先宣告

重點:agent outputs 帶 enum → switch cases 被靜態檢查窮盡。

異常 latency 走 A、errors 走 B。

workflow.json
{ "type":"agent", "name":"classify", "phase":"classify",
  "prompt":"判斷異常類型",
  "outputs":{"type":{"type":"str","enum":["latency","errors","other"]}} },

{ "type":"switch", "on":"{steps.classify.type}", "phase":"route",
  "cases":{ "latency":[...], "errors":[...], "other":[...] } }

gate + revise(吃回饋)

現行

人審 → 附一句話 → 帶著回饋重做

重點:revise 不是盲目重跑 —— 它帶 user 的一句自由回饋,注入重做那步的 prompt。

審週報 →「第三段太細,精簡」→ 帶著這句重生成 → 再審。

workflow.json
{ "type":"agent", "name":"draft", "phase":"draft",
  "prompt":"擬週報。修改意見:{steps.review.feedback}",
  "out":"report.md" },

{ "type":"gate", "name":"review", "phase":"review", "title":"審週報",
  "summary_from":"report.md",
  "allow":["approve","revise","reject"],
  "revise_to":"draft" }
// approve→續 reject→整條結束 revise→帶 {steps.review.feedback} 回 draft
// 具名 gate「review」的 feedback = {steps.review.feedback}(唯一合法的向前引用)

三個選項:approve(過)、reject(結束、什麼都不留)、revise(附一句自由回饋、打回 revise_to 指名的那步重做)。回饋以 {steps.<gate>.feedback} 注入該步 prompt(跟其他具名產出同一種寫法,讓唯一的回邊在圖上顯式)—— 同 steer 的自由指令精神。

6

人工閘 + 副作用

gate 只能在頂層 · 寫入只走 capability

gate approve / reject

現行

produce → review → commit

重點:只有 approve 續行;reject 讓整條 run 結束(不 commit)。summary_from 決定給人看哪些檔。
workflow.json
{ "type":"gate", "phase":"review", "title":"確認要發布?",
  "summary_from":"out/*.md", "allow":["approve","reject"] }

capability 副作用

現行

sandbox 無憑證 · 寫入只走這裡

重點:可靠副作用只走 capability(ingest_to_collection / upsert_context_card)。
workflow.json
{ "type":"capability", "call":"ingest_to_collection",
  "phase":"ingest", "collection":"docs", "path":"out/*.md" }
7

驗收 check + 重試

節點怎麼「不合格就不放行」

check 是 workflow 強制力的一半:step 產出不合格就擋下、帶回饋重試,重試 retries 次仍不過 → StepFailed(在 map 內就 skip+collect)。沒有 out 的 agent 一定要有 check。

check → 重試 → 失敗

現行

retry-with-feedback

重點:check 不過 → 帶著失敗原因重跑 step,用盡 retries 才算失敗。
workflow.json
{ "type":"agent", "phase":"draft", "prompt":"產生摘要",
  "out":"sum.md",
  "check":{"file_nonempty":{"path":"sum.md"}},
  "retries": 2 }

三種 check builder

現行

確定性的驗收條件

file_nonempty某檔存在且非空。
{"path":"..."}
choice_inJSON 某欄位的值落在允許集合內(= AI 結構化輸出的雛形)。
{"path":"..","key":"..","allowed":[..]}
collection_has某 collection 已含某項(ingest 後驗證)。
{"collection":"..","path":".."}

check 的參數也吃插值,例如 allowed 可來自 {config.kinds}

8

無法宣告 → run.py

圖畫不出來,或需要任意運算

護欄:圖必須靜態宣告 —— 不准無界迴圈、巢狀遞迴、AI 生新步驟、任意變數運算。

巢狀 / 遞迴展開

逃生口

深度未知

元素又展開新一批,深度不定 —— 圖畫不出來。

每個檔找出它引用的檔,再重複⋯⋯

無界適應

逃生口

步數 / 回合未知

下一步取決於發現什麼,AI 生出沒宣告的新步驟。

root cause 一路追下去 → 交給 RCA。

任意運算 / eval

逃生口

變數計算、自訂函式

DSL 的 {…} 是純定址、無 eval。要運算 → sandbox / run.py。

需要 if 之外的算式、迴圈控制。

安全線 · 貫穿全篇

圖的形狀必須事先靜態畫出來:分支有限、展開單層、無界者出局。

寬度與走哪條可執行期、甚至 AI 決定;但流程圖本身要畫得出來。這些語法(含 #428 新落地的六項)都守在這條線內 —— 它們讓 DSL 更一致、更少意外,不增加表達力(無 eval),所以不會把 DSL 推向圖靈完備。真正需要運算與無界流程的,交給 run.py。