fal genmedia CLI: Batch-Warteschlangen, Retries und Kostenkontrolle

Die fal genmedia CLI übernimmt Modellsuche, Schema-Inspektion, asynchrone Request-IDs, Downloads und JSON-Belege. Was noch fehlt, ist die Orchestrierung: begrenzte Parallelität, Wiederholungen bei vorübergehenden Fehlern, das Fortsetzen bereits erledigter Jobs und die persistente Ablage laufender IDs für asynchrone Aufgaben.

Einmal installieren, jeden Lauf reproduzierbar machen

Für macOS und Linux dokumentiert die Projekt-README:

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

Unter Windows kommt der dokumentierte PowerShell-Installer zum Einsatz:

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

In der README der genmedia CLI steht, dass genmedia FAL_KEY, JSON-Ausgaben, ein nicht interaktives Setup und optional einen Hintergrund-Checker für Updates unterstützt. In CI solltest du den Schlüssel über den Secret-Store des Runners einschleusen, statt ihn in ein Skript oder ein Command-Log zu schreiben.

Vor jedem Batch solltest du den Endpoint festlegen und seinen aktuellen Vertrag prüfen:

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

Die fal-Anleitung empfiehlt die Abfolge models → schema → run → status → download. Gehe nicht davon aus, dass die Flags eines Video-Endpoints auch bei einem anderen funktionieren. Das Workflow-Skill empfiehlt ausdrücklich, das Schema nach einem Validierungsfehler erneut zu prüfen.

Den Batch als Queue planen – nicht als Shell-Schleife

Die offiziellen genmedia-Materialien dokumentieren die asynchrone Ausführung. Einen nativen Befehl nach dem Muster „Lies diese CSV ein und verarbeite 500 Zeilen“ oder eine automatische Retry-Strategie versprechen sie jedoch nicht. Behandle die CLI als providerbewussten Runner und deinen Wrapper als Steuerzentrale der Queue.

Ein belastbarer Job-Datensatz braucht mindestens:

Für längere Image-to-Video- oder Video-Jobs verwendest du --async. Speichere die zurückgegebene request_id sofort und frage anschließend mit Endpoint-ID und Request-ID gemeinsam den Status ab:

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

Das Muster {request_id}_{index}.{ext} senkt das Risiko, dass zwei Jobs sich gegenseitig unbemerkt überschreiben. Lege die JSON-Datei direkt neben dem Medium ab und nicht in einem separaten temporären Verzeichnis.

Nur Fehler wiederholen, die sich tatsächlich beheben lassen

Als praxistauglicher Ausgangspunkt bietet sich diese Strategie an:

1. Wiederhole Netzwerk-Timeouts, Verbindungsabbrüche, HTTP 429 und vorübergehende 5xx-Antworten.
2. Verwende einen exponentiellen Backoff, zum Beispiel 2, 4, 8, 16 und anschließend 32 Sekunden, ergänzt um einen kleinen zufälligen Jitter.
3. Begrenze die Versuche pro Zeile, etwa auf vier Übermittlungen oder fünf Statusabfragen innerhalb eines Zeitfensters.
4. Wiederhole keine 401/403-Authentifizierungsfehler, 422-Schemafehler, Safety-Ablehnungen oder fehlerhaftes JSON.
5. Wenn eine Übermittlung abläuft, obwohl der Request den Provider möglicherweise bereits erreicht hat, prüfe den gespeicherten Request-Datensatz, bevor du einen zweiten kostenpflichtigen Request erzeugst.

Diese Regeln gehören in deinen Wrapper. Die öffentliche genmedia-README dokumentiert Befehle und Lebenszyklus-Operationen, aber kein garantiertes automatisches Retry-Verhalten. Bei einem 422 liest du validation_errors aus, führst genmedia schema erneut aus und korrigierst das genannte Feld, statt blind erneut zu senden.

Ein direkt nutzbarer Python-Batch-Wrapper

Der folgende Wrapper ist ein synchroner Einstiegspunkt für Bild-Batches. Er verwendet subprocess.run mit einer Argumentliste, überspringt bereits vorhandene Ausgabepfade, begrenzt die parallele Verarbeitung, wiederholt vorübergehende Prozessfehler und schreibt ein atomisches Manifest. Bei langen Video-Jobs solltest du die asynchrone Übermittlung vor dem Polling persistieren: submit -> write endpoint/request_id -> restart -> poll saved request_id -> download. Erneut senden solltest du nur, wenn keine Request-ID vorhanden ist.

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

Für Video ersetzt du ENDPOINT und ergänzt die schemaspezifischen Flags des jeweiligen Endpoints. Bei einer Image-to-Video-Kette lädst du das lokale Bild einmal mit genmedia upload ./frame.png --json hoch, übergibst die zurückgegebene URL an den Video-Job und hältst beide Datensätze im Manifest fest. Der Wrapper behauptet nicht, dass genmedia selbst die maximalen Ausgaben berechnet oder ein maximales Budget durchsetzt. Er verhindert lediglich doppelte lokale Arbeit und begrenzt Wiederholungen.

Vor der kostenpflichtigen Generierung eine Kostenprüfung einbauen

genmedia pricing <endpoint_id> --json liefert eine Preisauskunft, reserviert aber kein Budget und setzt auch kein Kostenlimit. Verwende den Befehl vor dem Batch und berechne anschließend eine konservative Obergrenze aus der Zahl der Zeilen, den Ausgaben pro Zeile, Auflösung und Dauer sowie der maximalen Anzahl an Wiederholungen.

Eine praktische Kostenprüfung kann so aussehen:

Speichere die Preisauskunft zusammen mit dem Manifest, denn sich Modellpreise und die Verfügbarkeit von Endpoints ändern können. Wenn der Provider keine vergleichbare Einheit liefert, solltest du das Ergebnis als Request-Obergrenze und nicht als Rechnung kennzeichnen.

fal genmedia CLI, Replicate CLI oder eigenes Skript?

Die drei Werkzeuge setzen an unterschiedlichen Stellen an. Die offizielle Replicate CLI bietet Befehle zum Ausführen und Streamen von Predictions, zur Inspektion von Modellschemas, für Uploads, Training und die Modellverwaltung. genmedia ist besonders praktisch, wenn ein Job mit der Suche nach fal-Endpoints beginnt und Medien durch fals Queue- und Download-Lebenszyklus laufen.

Meine Empfehlung ist klar: Verwende genmedia direkt für die Exploration und einen kleinen Python-Wrapper für produktive Batches innerhalb von fal. Eine eigene Provider-Abstraktion lohnt sich erst, wenn der Wechsel zwischen Providern tatsächlich erforderlich ist – nicht, weil ein Wrapper dadurch automatisch „enterprise-tauglicher“ wirkt.

Ausgaben als Datensätze behandeln und anschließend prüfen

Das öffentliche Workflow-Skill empfiehlt ein kompaktes Manifest mit Ziel, Node-ID, Endpoint-ID, Request-ID, Eingabe-URLs, Ausgabe-URLs, heruntergeladenen Dateien und Fehlernotizen. Das ist deutlich hilfreicher als ein Ordner voller Dateien mit automatisch erzeugten Namen.

Vor der Freigabe des Batches solltest du Folgendes prüfen:

1. Jede Zeile mit Status complete besitzt eine lokale Datei und einen JSON-Beleg.
2. Keine Zeile mit Status failed wurde stillschweigend ausgelassen.
3. Bilder haben die erwarteten Abmessungen und eine Dateigröße größer als null.
4. Videos lassen sich öffnen und haben die erwartete Dauer, Auflösung und Bildrate. Für diese Prüfung eignet sich ffprobe.
5. Prompts und Endpoint-IDs sind für die behaltenen Assets weiterhin gespeichert.
6. Ein erneuter Lauf desselben Manifests überspringt fertige Jobs, statt doppelte Dateien zu erzeugen.

Die fal-Anleitung betont, generierte Medien zusammen mit ihren JSON-Metadaten abzulegen. Das erleichtert auch eine spätere Migration zu einem anderen Provider: Du kannst Ausgabe-, Request- und Kostendaten vergleichen, statt einen Lauf anhand von Dateinamen rekonstruieren zu müssen.

FAQ

Kann die fal genmedia CLI von Haus aus Hunderte Prompts als Batch verarbeiten?

Die dokumentierten Befehle decken Modellausführung, asynchrone Statusabfragen, JSON-Ausgaben, Uploads und Downloads ab. Für einen fortsetzbaren Batch mit Hunderten Prompts brauchst du zusätzlich einen Manifest-Reader und eine Steuerung der parallelen Verarbeitung.

Ist genmedia günstiger als die Replicate CLI?

Die CLI legt den Preis des Providers nicht fest. Vergleiche für deinen konkreten Anwendungsfall die Preise der jeweiligen Endpoints, Ausgabeeinstellungen, Retry-Anzahl und das Verhalten bei der Datenübertragung. Ein pauschales „günstiger“ lässt sich bei unterschiedlichen Modellkatalogen nicht sinnvoll bestimmen.

Der kürzeste Weg sieht so aus: Verwende genmedia für Suche und Ausführung, ergänze für Batches einen manifestgesteuerten Wrapper und greife erst dann zu einem eigenen Skript, wenn du providerübergreifendes Routing oder einen zentralen Jobstatus brauchst.