fal genmedia CLIでバッチ処理を安定運用する：キュー、リトライ、コスト管理

fal genmedia CLIは、モデルの探索、スキーマ確認、非同期リクエストIDの管理、ダウンロード、JSONレシートの取得まで担える。一方で、大量ジョブを安全に回すには、同時実行数の制限、一時的な障害へのリトライ、完了済み処理の再開、非同期ジョブのリクエストID保存を行うオーケストレーターが別途必要になる。

インストール後は、どの実行も再現できる状態にする

macOSとLinuxでは、プロジェクトのREADMEに次の手順が案内されている。

curl https://genmedia.sh/install -fsS | bash
genmedia setup --non-interactive --api-key "$FAL_KEY" --no-auto-update

Windowsでは、ドキュメントに記載されたPowerShellインストーラーを使う。

irm https://genmedia.sh/install.ps1 | iex
genmedia setup --non-interactive --api-key "$env:FAL_KEY"

genmedia CLI READMEによると、genmediaはFAL_KEY、JSON出力、非対話セットアップ、任意のバックグラウンド更新チェックに対応している。CIでは、スクリプトやコマンドログへキーを書かず、ランナーのシークレットストアから注入するべきだ。

バッチを走らせる前に、使用するエンドポイントを固定し、現在の契約内容を確認しておく。

genmedia models "image to video" --json
genmedia schema bytedance/seedance-2.0/image-to-video --json
genmedia pricing bytedance/seedance-2.0/image-to-video --json

falのガイドで推奨されている流れは、models → schema → run → status → downloadだ。動画エンドポイント間でフラグがそのまま使い回せるとは限らない。workflow skillでも、バリデーションエラーが出たらスキーマを再確認するよう明示されている。

シェルのループではなく、キューとして設計する

公式のgenmedia資料には非同期実行が説明されているが、「CSVを読み込んで500行を実行する」ネイティブコマンドや、自動リトライ方針までは保証されていない。CLIはプロバイダー対応の実行役、ラッパーはキュー制御役として分けて考えるのがよい。

耐久性のあるジョブレコードには、少なくとも以下の項目が必要だ。

image-to-videoや動画のように時間がかかる処理では--asyncを使う。返されたrequest_idは即座に保存し、エンドポイントIDとリクエストIDの両方を指定してポーリングする。

genmedia run bytedance/seedance-2.0/image-to-video \
  --image_url "$IMAGE_URL" \
  --prompt "Slow product turn on a studio table; no text or logo" \
  --duration 4 --resolution 720p --aspect_ratio 16:9 \
  --async --json > logs/shot-001-submit.json

REQUEST_ID=$(jq -r '.request_id' logs/shot-001-submit.json)
genmedia status bytedance/seedance-2.0/image-to-video "$REQUEST_ID" \
  --download "outputs/{request_id}_{index}.{ext}" --json > logs/shot-001-result.json

{request_id}_{index}.{ext}というパターンなら、別ジョブの出力が気付かないうちに上書きされるリスクを下げられる。JSONは一時ディレクトリへ分けず、メディアと同じ場所に残しておこう。

回復可能な失敗だけをリトライする

最初のリトライ方針としては、次のようなものが実用的だ。

1. ネットワークタイムアウト、接続リセット、HTTP 429、一時的な5xxレスポンスをリトライ対象にする。
2. 2、4、8、16、32秒のように指数バックオフし、少量のランダムジッターを加える。
3. 行ごとの試行回数に上限を設ける。たとえば、送信は4回まで、ある時間枠内のステータスポーリングは5回までとする。
4. 401/403の認証エラー、422のスキーマ検証エラー、安全性による拒否、不正なJSONはリトライしない。
5. プロバイダーへ到達した可能性がある送信でタイムアウトした場合、2件目の有料リクエストを作る前に保存済みのリクエストレコードを確認する。

この方針を実装する場所はラッパーだ。公開されているgenmedia READMEはコマンドとライフサイクル操作を説明しているが、自動リトライの挙動までは保証していない。422が返ったらvalidation_errorsを確認し、genmedia schemaを再実行して、指摘されたフィールドを修正する。闇雲な再送は避けたい。

そのまま使えるPythonバッチラッパー

以下は同期型の画像バッチ向けスターターだ。subprocess.runを引数リストで呼び出し、完了済みの出力パスをスキップし、同時実行数を制限し、一時的なプロセス失敗をリトライして、マニフェストをアトミックに書き出す。長時間の動画ジョブでは、ポーリング前に非同期送信の情報を保存することが重要だ。流れはsubmit -> write endpoint/request_id -> restart -> poll saved request_id -> downloadであり、request IDがない場合にのみ再送する。

#!/usr/bin/env python3
import concurrent.futures as pool
import json, os, random, subprocess, tempfile, threading, time
from pathlib import Path

ENDPOINT = "fal-ai/flux/dev"
OUT = Path("outputs/images")
LOG = Path("outputs/logs")
MAX_WORKERS = 3
MAX_ATTEMPTS = 4
MANIFEST_LOCK = threading.Lock()

JOBS = [
    {"id": "shoe-001", "prompt": "Black running shoe, clean studio product photo", "file": "shoe-001.png"},
    {"id": "shoe-002", "prompt": "Black running shoe on wet pavement at dawn", "file": "shoe-002.png"},
]

OUT.mkdir(parents=True, exist_ok=True)
LOG.mkdir(parents=True, exist_ok=True)
MANIFEST = Path("outputs/manifest.json")
PREVIOUS = json.loads(MANIFEST.read_text()) if MANIFEST.exists() else {"results": []}
STATE = {r["id"]: r for r in PREVIOUS.get("results", [])}
DONE = {k: r for k, r in STATE.items() if r.get("status") == "complete"}
MAX_REQUESTS = len(JOBS) * MAX_ATTEMPTS
if MAX_REQUESTS > 100:
    raise SystemExit(f"request ceiling exceeded: {MAX_REQUESTS}")

def save_result(result):
    with MANIFEST_LOCK:
        STATE[result["id"]] = result
        payload = {"endpoint": ENDPOINT, "max_requests": MAX_REQUESTS,
                   "results": list(STATE.values())}
        fd, tmp = tempfile.mkstemp(dir=MANIFEST.parent, prefix="manifest.", text=True)
        with os.fdopen(fd, "w") as f:
            json.dump(payload, f, indent=2)
        os.replace(tmp, MANIFEST)

TRANSIENT_WORDS = ("429", "500", "502", "503", "504", "timeout", "temporarily", "connection")

def run_one(job):
    target = OUT / job["file"]
    receipt = LOG / f"{job['id']}.json"
    if job["id"] in DONE and target.exists() and target.stat().st_size > 0 and receipt.exists():
        return DONE[job["id"]]

    cmd = ["genmedia", "run", ENDPOINT, "--prompt", job["prompt"],
           "--num_images", "1", "--download", str(target), "--json"]
    last_error = ""
    for attempt in range(1, MAX_ATTEMPTS + 1):
        try:
            p = subprocess.run(cmd, text=True, capture_output=True, timeout=900)
            raw = p.stdout.strip()
            if p.returncode != 0:
                last_error = p.stderr[-1000:] or raw[-1000:]
                if not any(w in last_error.lower() for w in TRANSIENT_WORDS):
                    break
                if attempt < MAX_ATTEMPTS:
                    time.sleep((2 ** attempt) + random.random())
                continue
            try:
                data = json.loads(raw) if raw else {}
            except json.JSONDecodeError as exc:
                return {**job, "status": "failed", "attempts": attempt, "error": f"invalid JSON: {exc}"}
            if target.exists():
                (LOG / f"{job['id']}.json").write_text(json.dumps(data, indent=2))
                return {**job, "status": "complete", "attempts": attempt, "output": str(target)}
            last_error = p.stderr[-1000:] or raw[-1000:]
            if not any(w in last_error.lower() for w in TRANSIENT_WORDS):
                break
        except (subprocess.TimeoutExpired, OSError) as exc:
            last_error = str(exc)
        if attempt < MAX_ATTEMPTS:
            time.sleep((2 ** attempt) + random.random())
    return {**job, "status": "failed", "attempts": MAX_ATTEMPTS, "error": last_error}

results = []
with pool.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
    futures = [executor.submit(run_one, job) for job in JOBS]
    for future in pool.as_completed(futures):
        result = future.result()
        results.append(result)
        save_result(result)

print(json.dumps(results, indent=2))

動画ではENDPOINTを差し替え、対象エンドポイントのスキーマに応じたフラグを追加する。image-to-videoチェーンなら、ローカル画像をgenmedia upload ./frame.png --jsonで一度アップロードし、返されたURLを動画ジョブへ渡す。両方のレコードをマニフェストに保存しておこう。このラッパーは、genmedia自体が最大支出額を見積もったり強制したりするものではない。防げるのはローカルでの重複作業であり、リトライ回数を制限することまでだ。

有料生成の前にコストゲートを置く

genmedia pricing <endpoint_id> --jsonは価格の照会であって、予約や予算上限の設定ではない。バッチの開始前に確認し、行数、行ごとの出力数、解像度・尺の設定、最大リトライ回数から保守的な上限を計算する。

実用的なゲートには、次のようなものがある。

モデル価格やエンドポイントの提供状況は変わり得るため、価格レスポンスもマニフェストと一緒に記録する。プロバイダーが比較可能な単位を公開していない場合、結果は請求額ではなくリクエスト数の上限として扱うべきだ。

fal genmedia CLI、Replicate CLI、カスタムスクリプトの使い分け

これらのツールは、問題の異なる層を解決する。Replicate公式CLIには、予測の実行・ストリーミング、モデルスキーマ確認、アップロード、トレーニング、モデル管理のコマンドがある。一方、falエンドポイントを探し、falのキューとダウンロードのライフサイクルに沿ってメディアを扱うなら、genmediaのほうが適している。

結論はシンプルだ。探索にはgenmediaを直接使い、falだけで完結する本番バッチには小さなPythonラッパーを足す。プロバイダーを切り替える必要があるときだけ、カスタムのプロバイダー抽象化へ進めばよい。ラッパーのほうが「エンタープライズらしい」から、という理由で移行する必要はない。

出力を記録として残し、QAを回す

公開workflow skillでは、目的、ノードID、エンドポイントID、リクエストID、入力URL、出力URL、ダウンロード済みファイル、不具合メモを含むコンパクトなマニフェストが推奨されている。生成名のファイルが並ぶだけのフォルダより、はるかに役に立つ。

バッチを受け入れる前に、次を確認する。

1. すべてのcomplete行にローカルファイルとJSONレシートがある。
2. failed行が黙って除外されていない。
3. 画像が想定した寸法であり、ファイルサイズがゼロではない。
4. 動画を開け、想定どおりの尺、解像度、フレームレートになっている。ここではffprobeが適している。
5. 採用するアセットについて、プロンプトとエンドポイントIDが保持されている。
6. 同じマニフェストを再実行しても、重複ファイルの生成ではなくスキップになる。

falのガイドでは、生成メディアをJSONメタデータの近くに置くことが強調されている。この運用は、後からプロバイダーを移行するときにも役立つ。ファイル名から実行内容を復元しようとせず、出力、リクエスト、コストの記録を比較できるからだ。

FAQ

fal genmedia CLIだけで数百件のプロンプトをネイティブにバッチ処理できる？

ドキュメント化されているコマンドには、モデル実行、非同期ステータス処理、JSON出力、アップロード、ダウンロードがある。ただし、数百件のプロンプトを中断後に再開できる形で処理するには、マニフェストの読み取りと同時実行制御が別途必要だ。

genmediaはReplicate CLIより安い？

CLIがプロバイダーのモデル価格を決めるわけではない。ワークロードに対して、個別エンドポイントの価格、出力設定、リトライ回数、転送時の挙動を比較する必要がある。異なるモデルカタログをまたいで、ツール単位で「安い」と結論付けることに意味はない。

最短ルートを選ぶなら、探索と実行にはgenmediaを使い、バッチ処理にはマニフェスト駆動のラッパーを加える。マルチプロバイダーの振り分けや中央集約されたジョブ状態が必要になったときにだけ、カスタムスクリプトを選べばよい。