fal genmedia CLI: code batch, retry e controllo dei costi

fal genmedia CLI copre discovery, ispezione dello schema, request ID asincroni, download e ricevute JSON. Quello che manca è un orchestratore capace di limitare la concorrenza, ritentare gli errori temporanei, riprendere il lavoro già avviato e conservare gli ID delle richieste ancora in corso.

Installa una volta e rendi ogni esecuzione riproducibile

Per macOS e Linux, il README del progetto documenta:

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

Su Windows si usa l’installer PowerShell documentato:

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

Il README di genmedia CLI indica che genmedia supporta FAL_KEY, output JSON, configurazione non interattiva e un controllo opzionale degli aggiornamenti in background. In CI, passa la chiave tramite il gestore dei secret del runner invece di inserirla in uno script o nei log dei comandi.

Prima di avviare un batch, fissa l’endpoint e controlla il contratto effettivo:

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

La sequenza utile indicata dalla guida di fal è models → schema → run → status → download. Non dare per scontato che i flag di un endpoint video funzionino anche su un altro: la workflow skill raccomanda esplicitamente di ricontrollare lo schema dopo un errore di validazione.

Progetta il batch come una coda, non come un ciclo shell

La documentazione ufficiale di genmedia descrive l’esecuzione asincrona, ma non promette un comando nativo del tipo “leggi questo CSV ed esegui 500 righe”, né una politica automatica di retry. Considera quindi la CLI il runner che conosce il provider e il tuo wrapper il controller della coda.

Un record di lavoro persistente deve contenere almeno:

Per i job lunghi di image-to-video o video usa --async. Salva subito il request_id restituito, poi interroga lo stato indicando sia l’ID dell’endpoint sia quello della richiesta:

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

Il pattern {request_id}_{index}.{ext} riduce il rischio che due job sovrascrivano silenziosamente lo stesso file. Conserva il JSON accanto al contenuto multimediale, non in una directory temporanea separata.

Ritenta solo gli errori recuperabili

Una politica pratica da cui partire è questa:

1. Ritenta i timeout di rete, le connessioni interrotte, gli HTTP 429 e le risposte 5xx temporanee.
2. Usa un backoff esponenziale, per esempio 2, 4, 8, 16 e poi 32 secondi, aggiungendo un piccolo jitter casuale.
3. Imposta un limite di tentativi per riga, ad esempio quattro invii o cinque poll dello stato nell’arco di una finestra temporale.
4. Non ritentare gli errori di autenticazione 401/403, gli errori di validazione dello schema 422, i rifiuti per motivi di sicurezza o il JSON malformato.
5. Se un invio va in timeout dopo che la richiesta potrebbe essere arrivata al provider, controlla il record salvato prima di creare una seconda richiesta a pagamento.

Questa politica deve stare nel wrapper: il README pubblico di genmedia documenta comandi e operazioni del ciclo di vita, non un comportamento automatico di retry garantito. In caso di 422, leggi validation_errors, esegui di nuovo genmedia schema e correggi il campo indicato invece di reinviare alla cieca.

Un wrapper Python per i batch, pronto da copiare

Il wrapper seguente è una base sincrona per batch di immagini: usa subprocess.run con una lista di argomenti, salta i percorsi di output già completati, limita il lavoro concorrente, ritenta gli errori temporanei del processo e scrive un manifest in modo atomico. Per i job video lunghi, salva l’invio asincrono prima del polling: submit -> write endpoint/request_id -> restart -> poll saved request_id -> download; invia di nuovo la richiesta solo quando non esiste alcun 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))

Per i video, sostituisci ENDPOINT e aggiungi i flag previsti dallo schema dell’endpoint. In una pipeline image-to-video, carica una sola volta l’immagine locale con genmedia upload ./frame.png --json, passa l’URL restituito al job video e conserva entrambi i record nel manifest. Il wrapper non sostiene che sia genmedia a stimare o imporre una spesa massima: si limita a evitare il lavoro locale duplicato e a limitare i retry.

Inserisci un controllo dei costi prima della generazione a pagamento

genmedia pricing <endpoint_id> --json è una consultazione, non una prenotazione né un tetto di budget. Usalo prima del batch, quindi calcola un limite prudenziale considerando il numero di righe, gli output per riga, le impostazioni di risoluzione e durata e il numero massimo di retry.

Un controllo pratico può includere:

Registra la risposta sui prezzi insieme al manifest, perché i prezzi dei modelli e la disponibilità degli endpoint possono cambiare. Se il provider non espone un’unità di confronto equivalente, presenta il risultato come limite al numero di richieste, non come una fattura.

fal genmedia CLI, Replicate CLI o uno script personalizzato?

Questi strumenti operano a livelli diversi. La CLI ufficiale di Replicate offre comandi per eseguire e trasmettere le predizioni, consultare gli schemi dei modelli, gestire upload, training e modelli. genmedia è più adatto quando il lavoro parte dalla ricerca degli endpoint fal e prosegue attraverso il ciclo di vita di code e download di fal.

La mia raccomandazione è semplice: usa genmedia direttamente per l’esplorazione e un piccolo wrapper Python per un batch di produzione limitato a fal. Passa a un’astrazione personalizzata dei provider solo quando cambiare provider è un requisito, non perché un wrapper sembri più “enterprise”.

Tratta gli output come record e poi fai il controllo qualità

La workflow skill pubblica raccomanda un manifest compatto con obiettivo, node ID, endpoint ID, request ID, URL degli input, URL degli output, file scaricati e note sui difetti. È molto più utile di una cartella piena di file dai nomi generati automaticamente.

Prima di accettare il batch, verifica che:

1. Ogni riga complete abbia un file locale e una ricevuta JSON.
2. Nessuna riga failed sia stata ignorata senza segnalarlo.
3. Le immagini abbiano dimensioni previste e dimensione diversa da zero.
4. I video si aprano e abbiano durata, risoluzione e frame rate attesi; per questo controllo è adatto ffprobe.
5. Prompt e ID degli endpoint siano conservati per gli asset mantenuti.
6. Ripetere lo stesso manifest produca skip invece di file duplicati.

La guida di fal sottolinea l’importanza di tenere i media generati vicino ai relativi metadati JSON. Questa pratica rende possibile anche una futura migrazione a un altro provider: puoi confrontare output, richieste e costi invece di ricostruire un’esecuzione partendo dai nomi dei file.

Domande frequenti

fal genmedia CLI gestisce nativamente centinaia di prompt in batch?

I comandi documentati offrono esecuzione dei modelli, gestione asincrona dello stato, output JSON, upload e download. Per un batch ripristinabile di centinaia di prompt servono comunque un lettore di manifest e un controller della concorrenza.

genmedia è più economico di Replicate CLI?

Il prezzo del modello non dipende dalla CLI, ma dal provider. Confronta i prezzi dello specifico endpoint, le impostazioni degli output, il numero di retry e il trasferimento dati del tuo carico di lavoro: stabilire quale strumento sia “più economico” non ha senso tra cataloghi di modelli diversi.

Per la strada più breve, usa genmedia per discovery ed esecuzione, aggiungi un wrapper basato su manifest per i batch e scegli uno script personalizzato solo quando ti servono routing multi-provider o uno stato dei job centralizzato.