ワークフローを設計する¶

フレームワークが面倒を見るのはメカニズムだけだ:1 ステップをどう走らせるか、セッションをどうつなぐか、失敗したらどうするか、コンテキストをどう節約するか。 ワークフローを書くのはあなただ —— フレームワークはあなたがどんなプロジェクトを、どの言語でやっているかを知らないし、知るべきでもない。このページはワークフローの設計方法を扱う。Step と Workflow の全フィールド表は Python API にある。

何を解決するか¶

1 回のロングホライズン実行は、プロンプト 1 文で言い切れる代物ではない:まず要件を詰め、次に調査し、実装し、レビューする。各区間にそれぞれの役割、それぞれのコンテキスト、それぞれの受け入れ条件がある。 これを全部 1 文のプロンプトに書き込むと、モデルはどの区間を飛ばすかを自分で決めてしまう。ワークフローとして書けば、順序・終了条件・状態の受け渡しが Python コードになる —— 読めるし、テストできるし、壊れたステップだけ再実行できる。

Workflow がやるのは 3 つだけだ:

• 一連のステップを順番に走らせる
• 各ステップが前の何を見えるかを決める(3 種類のセッション接続 + ctx という辞書 1 つ)
• いつリトライし、いつ早期終了するかを決める

ドメインの仮定は一切含まない。どこで切るか、各ステップで何を受け入れ条件にするか、通らなかったらどうするか —— この 4 つが「ワークフローを設計する」ということだ。

使い方(最小コード)¶

# flows.py
from flower import AgentSpec, Step, Workflow

terse = AgentSpec(
    name="terse",
    instructions="回答极简,一行以内,不解释不寒暄。",
    allowed_tools=["Read", "Glob"],
    max_turns=4,
)


def main() -> Workflow:
    return Workflow([
        # 新規セッション:prompt に渡したものしか食わない
        Step("取词", terse, "读 seed.txt,只回文件里那个词。"),
        # これも新規セッション。前ステップの出力を prompt に注入する(安い・汚染しない)
        Step("造句", terse, lambda ctx: f"用「{ctx['取词']}」造一个五字短句,只回短句。"),
    ])

flower run flows.py:main -w /path/to/repo

flower run の引数は モジュール:属性 あるいは ファイルパス:属性。取得したオブジェクトが呼び出し可能ならまず 1 回呼び、Workflow を得てから走らせる。走り終わるとターミナルに総コストと実行マニフェストのパスが出る。

ドライバを自分で書いてもいい。Workflow.run の第 1 引数は Runtime だ:

ctx = await wf.run(rt, on_step=lambda step, r: print(f"{step.name} ok={r.ok} ${r.cost_usd:.4f}"))

実際に何をしているか¶

Step は何を受け取り、何を返さなければならないか¶

Step は関数ではなく宣言だ。実際に走るのは Runtime.run(step.spec, レンダリングされた prompt, ...) —— 1 ステップ = 1 回の Runtime.run = 1 本のセッション。

先頭 3 フィールドは位置引数、Step(name, spec, prompt):

• name —— ステップ名。これは同時に ctx のキー名であり、runs/manifest.json の行名であり、プロセスをまたぐリネージのキーでもある。
• spec —— どの AgentSpec で走らせるか。このステップのツールホワイトリスト、モデル、予算を決める。
• prompt —— str、または (ctx) -> str。呼び出し可能なら現在の ctx を受け取る。前ステップの出力を食わせる一番安い方法がこれだ(もう 1 つはセッションを接ぐ方法。下記参照)。

このステップが「返す」のは StepResult だが、ワークフローの中で手に入るのは 2 つだ:

• ctx[step.name] —— 既定では result.text。reduce を与えた場合は reduce の返り値に差し替わる。
• ctx["_results"][step.name] —— 完全な StepResult(コスト、ターン数、試行回数、session_id)。

result.text はメインスレッドの本文しか拾わない:subagent の発言はそれ自身の transcript にあり、subagent に渡したタスク指示書は kind="prompt"、切断時の合成エラーは kind="error" —— この 3 つはいずれも入らない。

reduce:糖衣ではない¶

既定で下流に渡るのはモデルの生の発言だ。ステップによっては、生の発言をそのまま下流に流すべきではない:

Step("确认需求", spec=确认者, prompt="帮我做一个 X",
     reduce=lambda r, ctx: ctx["_brief"].prompt_block())

実測では、要件確認のステップは 4 セクションのほかにコード全文を貼り付けてくる。下流に渡すべきはパース済みの 4 セクションでなければならない。さもないとそのコードの塊が次のステップの prompt に入る。clarify_step はこのフィールドで受け止めている。

reduce は同期関数でなければならない。gate / when / on_reject は async でよい。

状態は ctx をどう流れるか¶

ctx は dict[str, Any] で、Workflow.context そのものだ。各ステップが走り終わるとこの表のとおりに書かれる:

通過したかどうかに関わらず ctx["_results"][ステップ名] は書かれる。result.session_id が空でなければ ctx["_sessions"] にも書かれ、リネージにも記録される。

今回のワークフローが成功したかどうかは ctx.get("_failed_at") を見る。最後のステップに出力があるかどうかではない。

アンダースコアで始まるキーはすべて Workflow.run が自分で置くものだ:_runtime、_on_event、_sessions、_results、_lineage、_woke、_aborted、_failed_at。これらを自分のステップ名に使わないこと。各メカニズムも自前のキーを置く(_brief / _goal / _verdict など)。完全な一覧は Python API にある。

このうち _runtime と _on_event は gate のためにある:gate は自分で agent を派遣して判定させることができ、その判定過程もそのまま UI に流れる —— そうしないと十数秒のあいだ画面が真っ暗になり、固まったように見える。 ゴールガードはこうやって実装されている。

ctx は同じ dict だ:同じ Workflow オブジェクトを 2 回目に走らせると、前回のキーがまだ残っている。 きれいにやり直したければ新しく作るか、明示的に context={} を渡すこと。

判定と差し戻し:gate、on_reject、StepAbort¶

gate(result, ctx) -> bool が判断するのは「走り切ったが、合格か」だ。必ず知っておくべき細部が 2 つある:

• result.ok が偽なら gate はそもそも呼ばれない(短絡)。
• 1 回の試行につき 1 回しか呼ばれない。結論は後で使い回す —— 副作用を持ちうるからだ。clarify_step の gate は要件確認書をディスクに書き出すので、重複して発火すれば重複して書き込む。

gate が通らなかったあとどうやり直すかは、on_reject を与えたかどうかで決まる:

2 つ目は「差し戻して、どこが足りないかを伝えて、続きをやらせる」だ —— すでに済んだ作業もコンテキストもそのまま残る。 on_reject が空文字列を返した場合、あるいはその試行がそもそも session_id を得ていない場合は、最初からやり直しに退化する。

gate は StepAbort を投げることもできる。意味はもう試しても無駄だから残りのラウンドを消費するなだ:

from flower import StepAbort

def gate(result, ctx):
    if "这个环境装不了依赖" in result.text:
        raise StepAbort("环境缺依赖,再跑几轮也一样")
    return "验收通过" in result.text

投げられたあと:理由が ctx["_aborted"] に記録され、このステップは失敗扱いになって on_fail(既定は "stop")に従い、リトライループはその場で break し、残りの retries は 1 回も消費されない。

区別をしっかり覚えること:False を返すのは「今回はダメ、もう 1 ラウンド」。StepAbort は「もう 1 回やっても無駄」。 典型的な場面は、目標がこの環境では達成不能と判定され、しかも聞ける相手もいないときだ —— 空回りを続けるのが一番高くつく。

2 層のリトライを混同しない¶

続行に使うプロンプトには意図的にエラーの詳細を一切含めていない —— モデルが知る必要があるのは「中断された、続きをやれ」であって、ENOTFOUND なのか 503 なのかではない。

ステップをつなぐ¶

ステップ間の状態の受け渡しには 3 種類の接ぎ方があり、どれを選ぶかで次のステップに何が見えるかが決まる:

resume_from が指すステップは実際に session を生んでいなければならない。when でスキップされた、あるいはそもそも走っていない場合、Workflow.run はそのまま ValueError を投げる —— 黙って新規セッションに降格したりはしない。それをやると「連続した記憶」という前提が静かに崩れるからだ。

何度も痛い目を見て得た設計上の経験をいくつか:

1. 1 ステップに 1 つの受け入れ可能な目標。 ステップ境界はコンテキスト境界だ:resume_from=None の場所では、それ以前のツール結果は完全に常駐しなくなる。コンテキスト経済学を参照。
2. 迷ったらまず clarify_step。 ロングホライズンでは「目標を取り違えた」が一番高くつく誤りであり、しかもそれはコンテキストを節約するどの層でも消せない類のものだ。事前確認を参照。
3. 派遣するタスクは自己完結させる。 subagent はクリーンなコンテキストで、コーディネーターが知っていることを知らない。必要な背景はタスク指示書に書くか、どの artifact を読めばいいかを伝えること。
4. 長い出力はディスク経由で、会話経由にしない。 これは既に WORKER_RULES に書き込まれている。あなたの instructions でそれを打ち消さないこと(「完全なログを貼り戻して見せてくれ」など)。
5. gate はまず硬い条件で止める。 ファイルがあるか、終了コードが 0 か —— 1 行の Python で判定できることにモデルを派遣しない。モデルに判定させたいなら既製の with_goal を使う —— これは gate を、独立した判定者を走らせる実装に置き換えるものだ。gate の中で自作しないこと。
6. 同じリポジトリを並行して変更するなら worker(isolate=True)。 後始末(マージ、worktree の掃除、PR を開く)は今のところあなたのワークフロー側の仕事だ。harness が保証するのは、変更がそれぞれの worktree に落ちることだけ。

ワークベンチは Workflow に付ける¶

ワークフローがワークベンチにファイルを書く場面 —— 典型は clarify_step(brief_path=...) —— では、必ず自分で Workbench を作り、同時に Workflow.workbench に付け、かつ Runtime にも渡すこと:

from pathlib import Path

from flower import (HumanChannel, Runtime, Step, Workbench, Workflow,
                    clarify_step, coordinator, worker)

wb = Workbench(Path.cwd()).ensure()
ch = HumanChannel(log_path=wb.notes / "问答记录.md", timeout_s=1800.0)

主控 = coordinator("协调者", "", {
    "coder": worker("写代码与测试。要动手实现的活派给它。",
                    "你负责实现。每改一处就跑一次验证,别攒到最后。"),
}, channel=ch)

wf = Workflow(
    [
        clarify_step(ch, brief_path=wb.notes / "需求.md", prompt="帮我做一个 X"),
        Step("干活", spec=主控, prompt=lambda ctx: f"照这份需求做:\n\n{ctx['确认需求']}"),
    ],
    channel=ch,
    workbench=wb,
)

rt = Runtime(workspace=Path.cwd(), run_dir="runs", workbench=wb)

channel を workflow に付ける理由は 2 つある:run() がその on_event を同じイベント出口につなぐこと(channel.on_event がまだ None の場合に限る)、そしてドライバがこのフィールドで「誰に答えればいいか」を知ること。

continuous=True:同じパスでもう一度走らせる¶

上の 3 種類の接ぎ方は1 回の実行の内側でのステップ間の話だ。プロセスをまたぐのは別の軸になる:

Workflow([...], continuous=True)     # 既定値

同じワークスペースでもう一度走らせると、各ステップは前回のそのセッションの続きから話す —— <run_dir>/lineage.json の「ステップ名 → session_id」に依っている。ロード時には各レコードを runtime.has_session() で検証し、session がまだストアにあるかを確かめ、生きているものだけを使う:リネージファイルは sessions.db より長生きしうるし、存在しない session を resume すると子プロセスが起動するまで落ちないからだ。

帰結が 3 つある:

• resume_from=None は「まっさらな新規セッション」を意味しない。 1 回目はそうだが、2 回目はそうではない。毎回新規セッションにしたければ、明示的に Workflow(..., continuous=False) と書くこと。
• 継続ではコンテキストが伸び続ける。 継続時に別のことを言いたいなら Step.resume_prompt を与える —— 相手のコンテキストに既にあるものを再送すべきではない。
• 明示的に resume_from を書いたステップは影響を受けない。そちらが優先される。

完全な設計と --new については継続を参照。

使うべきでないとき¶

• agent を 1 つ走らせるだけで、判定も要らない —— Workflow を被せない。素直に await rt.run(spec, "…")、あるいはコマンドラインで flower once "读一眼这个仓库"。

• 形が「要件を詰める → 目標を決める → 作業する」そのもの —— 既製の starter_flow() を使えばいい。自分で書く必要はない:

  from flower import starter_flow
  
  wf = starter_flow("帮我做一个 X", workspace=".", run_dir="runs",
                    rounds=3, timeout_s=1800.0, isolate=False)
  

  これは3 ステップだ:确认需求 → 设定目标 → 干活(判定ループ付き、判定ステップの名前は 干活·判定#N)。 goal=False なら第 2 ステップと判定ループがなくなり、clarify_only=True なら第 1 ステップだけが残る。 HumanChannel と Workbench を自前で持ち、workflow に付けてあるので、Runtime(workbench=wf.workbench) をそのまま使えばいい。別に組み立てないこと。

  コードを書かなくてもいい。プロジェクトディレクトリで flower "帮我做一个 X" を走らせれば、これが動く。 これは「推奨するワークフロー設計」ではない。設定ゼロで走り出せるようにしてあるだけだ。

• ステップを「1 つの受け入れ可能な目標」より細かく切る —— 純損だ。各ステップは新しいセッションを 1 本立てる必要があり、新しいセッションには起動フロア(コーディネーターの実測で約 34k コンテキスト)があるので、薄まらない。

• 事後に特定のメッセージまで巻き戻したい —— Workflow ではこの道は通れない。resume_at を渡すことが一切ないからだ。直接 Runtime.run(spec, "从这里重来", resume=sid, resume_at=uuid) を呼ぶこと。

役割の選び方(coordinator / worker / clarify / judge / oracle)、および Step と Workflow のフィールドごとの意味は Python API を、用語は用語集を参照。