AIREITER

fal genmedia CLI：批次佇列、重試與成本控管

最近更新: 2026-10-06 01:35:32

fal genmedia CLI 已經處理了模型探索、schema 查詢、非同步 request 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 環境執行，應透過 runner 的 secret store 注入金鑰，避免把它寫進腳本或命令紀錄。

啟動批次前，先固定 endpoint，並確認它目前的實際介面規格：

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。不要假設某個影片 endpoint 的旗標能直接套用到另一個 endpoint；workflow skill 特別建議，在遇到驗證錯誤後應重新查閱 schema。

把批次工作當成佇列管理，不是 shell 迴圈

官方 genmedia 文件說明了非同步執行方式，但沒有承諾提供「讀取這份 CSV 並執行 500 列」的原生指令，也未提供自動重試策略。較合理的分工是：CLI 負責理解供應商的任務執行，外層 wrapper 則負責控制佇列。

一筆可長期保存的任務紀錄，至少應包含：

欄位用途
id可用於續跑與去重的穩定輸入識別碼
endpoint實際使用的完整模型路徑
prompt用於重現與稽核
statuspending、submitted、complete、failed 或 skipped
request_idgenmedia status 所需的識別碼
attempts避免重試失控
output可預測的本機輸出路徑
error讓失敗列可以後續處理

圖片轉影片或其他耗時影片任務應使用 --async。取得回傳的 request_id 後要立刻保存，接著以 endpoint ID 與 request 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. 401/403 驗證錯誤、422 schema 驗證錯誤、安全拒絕與格式錯誤的 JSON 都不應重試。
  5. 如果提交逾時，但請求可能已送達供應商，建立第二筆付費請求前應先檢查已保存的請求紀錄。

這套策略應由 wrapper 實作；公開的 genmedia README 說明的是命令與任務生命週期操作，並未保證提供自動重試行為。遇到 422 時，請讀取 validation_errors、再次執行 genmedia schema，並修正被點名的欄位，而不是盲目重新提交。

可直接複製使用的 Python 批次 wrapper

下方 wrapper 是同步圖片批次處理的起點：它使用帶有引數列表的 subprocess.run，略過已完成的輸出路徑、限制並行工作數、重試暫時性的程序失敗，並以原子方式寫入 manifest。若要處理長時間影片任務，必須先保存非同步提交結果再輪詢：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，再加入該 endpoint schema 所要求的旗標即可。建立圖片轉影片鏈路時，先以 genmedia upload ./frame.png --json 上傳本機圖片一次，將回傳 URL 傳給影片任務，並在 manifest 中保留兩筆紀錄。這個 wrapper 並不表示 genmedia 本身會估算或強制執行最高支出；它只會避免重複的本機工作，並限制重試次數。

付費生成前先設成本閘門

genmedia pricing <endpoint_id> --json 是查價工具，不是費用保留或預算上限。請在批次前使用它，再依列數、每列輸出數、解析度／時長設定與最大重試次數，計算保守的成本上限。

可採取的實務閘門如下：

控制項實作方式
列數硬上限manifest 超過核准數量時拒絕啟動
模型層級先以較便宜或快速的 endpoint 製作草稿，通過 QA 後才渲染最終版本
輸出上限明確指定 num_images，不要依賴預設值
重試預算將重試與首次嘗試分開計數
續跑略過已有驗證本機輸出的列
取消適當時以 genmedia status ... --cancel 取消佇列中的工作

應將查價回應和 manifest 一併保存，因為模型價格與 endpoint 可用性可能改變。若供應商沒有公開可比較的計價單位，應把結果標示為請求數量上限，而不是帳單金額。

fal genmedia CLI、Replicate CLI 與自訂腳本怎麼選

這些工具解決的是不同層次的問題。Replicate 官方 CLI 提供執行與串流 prediction、查看模型 schema、上傳、訓練與模型管理等命令。若工作從探索 fal endpoint 開始，並需走完 fal 的佇列與下載生命週期，genmedia 會更順手。

選擇最適合的情境主要取捨
fal genmedia CLIfal 原生模型搜尋、schema 查詢、非同步任務、下載與 agent shell 操作供應商專用；批次策略仍需放在 CLI 外部
Replicate CLIReplicate prediction、串流、模型／schema 操作與訓練命令模型目錄與生命週期都不同；不要期待 fal 的 endpoint ID 或旗標可以直接沿用
自訂 Python/HTTP 腳本多供應商路由、核准閘門、資料庫狀態、佇列與帳務策略驗證、schema 變動、輪詢、下載與錯誤處理都由你維護

我的建議很直接：探索階段直接使用 genmedia；要處理只使用 fal 的正式批次工作，則加上一個小型 Python wrapper。只有在切換供應商已是明確需求時，才需要升級成自訂的供應商抽象層，而不是因為 wrapper 看起來不夠「enterprise」。

把輸出當成可追溯紀錄，再進行 QA

公開 workflow skill 建議建立精簡 manifest，內容包括目標、node ID、endpoint ID、request ID、輸入 URL、輸出 URL、下載檔案與瑕疵備註。這比留下一整個存放自動命名檔案的資料夾實用得多。

接受整批輸出前，請確認：

  1. 每一列 complete 都有本機檔案與 JSON 收據。
  2. 沒有任何 failed 列被悄悄遺漏。
  3. 圖片的尺寸符合預期，且檔案大小不為零。
  4. 影片可以開啟，並具有預期的時長、解析度與影格率；ffprobe 適合用於這項檢查。
  5. 保留下來的資產仍保有對應的 prompt 與 endpoint ID。
  6. 重新執行同一份 manifest 時，應略過既有工作，而非產生重複檔案。

fal 指南強調，生成媒體應和它的 JSON 中繼資料放在一起。這個習慣也讓日後遷移供應商成為可能：你可以比較輸出、請求與成本紀錄，不必再從檔名中嘗試還原當初的執行過程。

常見問題

fal genmedia CLI 原生支援數百條 prompt 的批次處理嗎？

文件中的命令涵蓋模型執行、非同步狀態處理、JSON 輸出、上傳與下載。若要建立可續跑的數百條 prompt 批次作業，仍需要 manifest 讀取器與併發控制器。

genmedia 比 Replicate CLI 便宜嗎？

CLI 不會決定供應商的模型價格。請依你的工作負載比較具體 endpoint 的價格、輸出設定、重試次數與傳輸行為；面對不同模型目錄，工具層級的「較便宜」並沒有實際意義。

若想走最短路徑，使用 genmedia 進行探索與執行，加入由 manifest 驅動的 wrapper 處理批次；只有在需要多供應商路由或集中化任務狀態時，再選擇自訂腳本。