fal genmedia CLI : files d’attente, nouvelles tentatives et maîtrise des coûts

fal genmedia CLI sait découvrir les modèles, inspecter leurs schémas, gérer les identifiants de requêtes asynchrones, télécharger les résultats et produire des reçus JSON. Ce qui lui manque, c’est une couche d’orchestration capable de plafonner la concurrence, de retenter les erreurs transitoires, de reprendre les tâches déjà terminées et de conserver les identifiants des requêtes encore en cours.

Installez-le une fois et rendez chaque exécution reproductible

Pour macOS et Linux, le README du projet indique :

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

Sous Windows, utilisez l’installateur PowerShell documenté :

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

Le README de genmedia CLI précise que genmedia prend en charge FAL_KEY, la sortie JSON, la configuration non interactive et, en option, un vérificateur de mises à jour exécuté en arrière-plan. En CI, injectez la clé depuis le gestionnaire de secrets du runner plutôt que de la placer dans un script ou dans les journaux de commande.

Avant de lancer un lot, verrouillez le endpoint et inspectez son contrat réel :

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 séquence utile proposée par le guide fal est models → schema → run → status → download. Ne partez pas du principe que les options d’un endpoint vidéo fonctionnent sur un autre : le workflow skill recommande précisément de vérifier à nouveau le schéma après une erreur de validation.

Concevez le batch comme une file d’attente, pas comme une boucle shell

La documentation officielle de genmedia couvre l’exécution asynchrone, mais ne promet ni une commande native du type « lire ce CSV et exécuter 500 lignes », ni une politique automatique de nouvelles tentatives. Considérez donc le CLI comme le moteur conscient du fournisseur, et votre wrapper comme le contrôleur de file d’attente.

Un enregistrement de tâche fiable doit au minimum contenir :

Utilisez --async pour les tâches longues de type image-to-video ou vidéo. Enregistrez immédiatement le request_id renvoyé, puis interrogez le statut avec l’identifiant de l’endpoint et celui de la requête :

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

Le motif {request_id}_{index}.{ext} réduit le risque que deux tâches écrasent discrètement le même fichier. Conservez le JSON à côté du média, et non dans un répertoire temporaire séparé.

Ne retentez que les erreurs susceptibles de disparaître

Une politique de départ raisonnable peut suivre ces règles :

1. Retenter les timeouts réseau, les réinitialisations de connexion, les réponses HTTP 429 et les erreurs 5xx transitoires.
2. Appliquer un backoff exponentiel — par exemple 2, 4, 8, 16 puis 32 secondes — avec une légère part d’aléatoire.
3. Limiter le nombre d’essais par ligne, par exemple à quatre soumissions ou cinq interrogations de statut sur une période donnée.
4. Ne pas retenter les erreurs d’authentification 401/403, les erreurs de validation de schéma 422, les refus de sécurité ou le JSON malformé.
5. Si une soumission expire alors que la requête a peut-être atteint le fournisseur, inspecter l’enregistrement sauvegardé avant de créer une seconde requête payante.

Cette politique relève de votre wrapper : le README public de genmedia documente les commandes et les opérations du cycle de vie, mais ne garantit pas de comportement automatique de retry. En cas de 422, lisez validation_errors, relancez genmedia schema et corrigez le champ indiqué au lieu de soumettre machinalement la même requête.

Un wrapper Python batch prêt à l’emploi

Le wrapper ci-dessous constitue un point de départ synchrone pour des lots d’images : il utilise subprocess.run avec une liste d’arguments, ignore les fichiers de sortie déjà terminés, limite le travail concurrent, retente les erreurs transitoires du processus et écrit un manifeste de manière atomique. Pour les longues tâches vidéo, persistez la soumission asynchrone avant d’interroger le statut : submit -> write endpoint/request_id -> restart -> poll saved request_id -> download ; ne soumettez à nouveau que si aucun identifiant de requête n’existe.

#!/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))

Pour la vidéo, remplacez ENDPOINT et ajoutez les options propres au schéma de l’endpoint. Dans une chaîne image-to-video, envoyez une fois l’image locale avec genmedia upload ./frame.png --json, transmettez l’URL renvoyée à la tâche vidéo et conservez les deux enregistrements dans le manifeste. Ce wrapper ne prétend pas que genmedia estime ou impose lui-même une dépense maximale ; il évite seulement les doublons locaux et limite les retries.

Ajoutez un garde-fou budgétaire avant toute génération payante

genmedia pricing <endpoint_id> --json fournit une information tarifaire, pas une réservation ni un plafond budgétaire. Utilisez-le avant le batch, puis calculez un plafond prudent à partir du nombre de lignes, du nombre de sorties par ligne, des réglages de résolution et de durée, ainsi que du nombre maximal de tentatives.

Un garde-fou pratique peut combiner les contrôles suivants :

Enregistrez la réponse tarifaire avec le manifeste, car les prix des modèles et la disponibilité des endpoints peuvent évoluer. Si le fournisseur ne propose pas d’unité comparable, présentez le résultat comme un plafond du nombre de requêtes, et non comme le montant d’une facture.

fal genmedia CLI, Replicate CLI ou script personnalisé ?

Ces outils n’interviennent pas au même niveau. Le CLI officiel de Replicate propose des commandes pour exécuter et streamer des prédictions, inspecter les schémas des modèles, gérer les uploads, entraîner des modèles et administrer les modèles. genmedia devient plus intéressant lorsque le travail commence par la découverte d’endpoints fal et se poursuit dans le cycle de vie fal des files d’attente et des téléchargements.

Ma recommandation est simple : utilisez genmedia directement pour l’exploration, puis ajoutez un petit wrapper Python pour un batch de production limité à fal. Ne passez à une abstraction multi-fournisseurs que si le changement de fournisseur est réellement nécessaire, pas parce qu’un wrapper donne une allure plus « enterprise » au projet.

Considérez les sorties comme des données, puis effectuez le contrôle qualité

Le workflow skill public recommande un manifeste compact contenant l’objectif, l’identifiant du nœud, l’identifiant de l’endpoint, l’identifiant de la requête, les URL d’entrée, les URL de sortie, les fichiers téléchargés et les défauts constatés. C’est bien plus exploitable qu’un dossier rempli de fichiers aux noms générés automatiquement.

Avant d’accepter le batch, vérifiez les points suivants :

1. Chaque ligne complete possède un fichier local et un reçu JSON.
2. Aucune ligne failed n’a été écartée silencieusement.
3. Les images ont les dimensions attendues et une taille non nulle.
4. Les vidéos s’ouvrent et présentent la durée, la résolution et la fréquence d’images attendues ; ffprobe convient à cette vérification.
5. Les prompts et les identifiants d’endpoint sont conservés pour les ressources que vous gardez.
6. La réexécution du même manifeste provoque des omissions, et non la création de doublons.

Le guide fal insiste sur l’intérêt de conserver les médias générés à proximité de leurs métadonnées JSON. Cette pratique facilite aussi une éventuelle migration vers un autre fournisseur : vous pouvez comparer les sorties, les requêtes et les coûts au lieu de tenter de reconstituer une exécution à partir des seuls noms de fichiers.

FAQ

fal genmedia CLI sait-il traiter nativement des centaines de prompts en batch ?

Les commandes documentées assurent l’exécution des modèles, le suivi asynchrone, la sortie JSON, les uploads et les téléchargements. Pour obtenir un batch reprenable de centaines de prompts, il faut encore ajouter un lecteur de manifeste et un contrôleur de concurrence.

genmedia est-il moins cher que Replicate CLI ?

Le CLI ne détermine pas le prix des modèles du fournisseur. Comparez les tarifs de l’endpoint choisi, les réglages de sortie, le nombre de retries et les transferts associés à votre charge de travail ; conclure qu’un outil est « moins cher » n’a pas de sens entre deux catalogues de modèles différents.

Pour aller au plus court, utilisez genmedia pour découvrir les modèles et lancer les tâches, ajoutez un wrapper piloté par manifeste pour les batchs, et ne choisissez un script personnalisé que si vous avez besoin d’un routage multi-fournisseurs ou d’un état centralisé des jobs.