flowing

動的ワークフローに取って代わるものではありません — 最初にお読みください

Claude Code の動的ワークフローはサブエージェント（個別のコンテキスト、 最大16並列／1000エージェントへのファンアウト）をオーケストレーションします。このスキルは別のプリミティブです。 耐久性のある副作用とチェックポイントからの再開に対応した、独自のツール呼び出しを制御する単一コンテキストの制御フローです。 ワークフローのランタイムは、ファイルシステムやシェルを直接操作できない仕様です。処理はエージェントが実行し、スクリプトはそれらを調整するだけです。Flowing はその逆で、スクリプトが処理を実行します。

コンテキスト内のパイプライン（3つ以上のステップ、分岐、再試行、検証、切り離された副作用）には Flowing を使います。多くのサブエージェントが必要なときはワークフローを使います。両者は組み合わせられるもので、競合するものではありません。ワークフローのために Flowing を捨てないでください。そうすると、hub-spoke が依存する耐久性のある副作用と、セッションをまたいだチェックポイントが失われます。

Flowing — 散文ではなくコードで制御フローを記述する

手順に3つ以上のステップ、分岐、再試行、または契約が必要な場合は、命令文の散文ではなく、Python タスクの DAG としてエンコードします。「まず X、次に Y、そして Z の場合は3回再試行する」のような散文は、読み飛ばされたり、生成後に無視されたりします。@task グラフは構造的です。入力がバインドされるまでステップを実行できず、不正な入力で動作するゲートはスキップできません。

ランナーが制御フロー（分岐、再試行、検証、失敗の伝播、並列化）を担当します。あなたは末端で判断を行います。ランナー：scripts/flowing.py。

クイックスタート

from flowing import task, Flow

@task
def fetch_data():
    return {"items": [1, 2, 3]}

@task(depends_on=[fetch_data])
def process(fetch_data):          # param name must match the dep's name
    return sum(fetch_data["items"])

@task(depends_on=[process])
def store(process):
    print(f"Result: {process}")

Flow(store).run()                 # topo-sorts, runs each layer, parallel within a layer

各タスクは、依存関係をその名前を付けた kwargs として受け取ります。同じレイヤー内の独立したタスクは並列実行されます。

制御フローのプリミティブ

タスク本体内の if 文ではなく、グラフ構造として分岐や契約をエンコードします。

when= — 条件ゲート

述語（収集した依存関係の値に対する）が真の場合にのみタスクを実行します。偽の場合は SKIPPED となり、そのスキップは依存タスクにも伝播します。

@task(depends_on=[fetch], when=lambda fetch: fetch["needs_processing"])
def process(fetch):
    return transform(fetch["payload"])

validate= — エッジの契約

本体の実行前に、収集した依存関係の値を検査します。例外が発生すると、再試行なしで FAILED になります（不正な入力が自動的に修正されることはありません）。検査に合格すると、処理を続行します。

def must_have_items(fetch):
    if not fetch.get("items"):
        raise ValueError("fetch returned empty payload")

@task(depends_on=[fetch], validate=must_have_items)
def process(fetch):
    return sum(fetch["items"])

retry_until= — 述語駆動のループ

本体を実行した後、retry_until(value) を呼び出します。True なら完了です。False なら、retry= の上限を消費しながら再試行します。自己修正する LLM ステップに使います。生成し、確認し、再生成します。

@task(retry=4, retry_until=lambda r: r["valid"])
def generate_until_valid():
    candidate = llm_call(...)
    return {"valid": passes_schema(candidate), "candidate": candidate}

例外が発生した場合にのみ再試行する retry= 単独の指定とは異なります。

その他の機能

• 並列実行 — レイヤー内の独立したタスクをスレッドプールで実行します（max_workers=）。
• detached=True — 副作用を伴うタスク（メモリへの書き込み、通知など）はメインの DAG の後に実行され、失敗してもメインの DAG を停止させません。
• プロセス内での再開 — flow.run() → 修正 → flow.resume() の順に実行すると、失敗した箇所から再実行され、成功済みのタスクはメモリ内にキャッシュされます（同じプロセス内に限ります）。flow.override(task, value) を使うと、修正済みの結果を注入できます。
• 永続ジャーナル（journal_path=） — オプトインのコンテンツアドレス方式による再生機能で、コンテナの終了後も状態を保持します。Flow(term, journal_path="/path/run.jsonl").run() は、成功した各タスクの結果を追記専用の JSONL に追記します。キーとなる step_key は、タスクのバイトコードと、その when／validate／retry_until の本体、および依存先のキー（連鎖するため、上流の変更は下流にも伝播します）を対象にした SHA-256 です。後続の run() は、新しいコンテナで実行する場合でも、ジャーナルから変更のない先行部分を再生し、キーが存在しないタスクのみを実行します。タスク本体を編集するとキーが無効になり、そのタスクと依存タスクが再実行されますが、見た目だけの設定（retry=、timeout_s=、name）は影響しません。これは、セッションをまたいでチェックポイントを使うハブ・スポーク型の処理を支えます。注意点：結果は pickle 化されるため、pickle 化できない戻り値は単に再実行されます。また、クロージャーにキャプチャされた値はキーに含まれません（タスク本体のコードのみが対象です）。
• timeout_s=、指数バックオフ付きの retry=、fail_fast=。

クイックスタートと上記の 3 つの基本機能以外を使う前に、references/reference.md をお読みください。すべての @task パラメーター、Flow のメソッド、再開／上書き、detached タスクの自動検出、および validate=／when= のシグネチャ照合に関する注意点を説明しています。

使用する場面

• 分岐が重要な手順がある場合 → when= で分岐を構造化できます。
• ステップに入力の契約がある場合 → validate= で強制できます。
• LLM のステップを収束させる必要がある場合 → retry_until= でチェックをループに組み込めます。
• 並列化できる独立した操作が 3 つ以上ある場合。
• 後半の失敗で、それまでの作業を無駄にしたくない複数ステップのパイプラインがある場合。
• クリティカルパスを止めてはならない副作用がある場合 → detached=True。

使用しない場面

• 単一の逐次処理 — 関数を直接呼び出してください。
• 次のステップで、述語では表現できない直前の結果に対する推論が必要な場合 — think ループを使ってください。
• 非同期または分散型のワークフロー — これは単一コンテナ内のスレッドプールを基盤としています。

作成時の指針

*「まず X を呼び出し、Y を検証し、それから Z の場合は最大 3 回再試行する」*のような説明を書いていることに気付いたら、それはフローグラフです。リリース前にリファクタリングしてください。文章で指示しても強制はできませんが、@task グラフなら強制できます。