fal genmedia CLI: colas por lotes, reintentos y control de costes

fal genmedia CLI se encarga del descubrimiento, la inspección de esquemas, los identificadores de solicitudes asíncronas, las descargas y los recibos JSON. Lo que falta es una capa de orquestación que limite la concurrencia, reintente los fallos temporales, reanude el trabajo ya completado y conserve los identificadores de las solicitudes en curso.

Instálalo una vez y haz que cada ejecución sea reproducible

Para macOS y Linux, el README del proyecto documenta estos comandos:

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

En Windows se utiliza el instalador de PowerShell documentado:

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

El README de genmedia CLI indica que genmedia admite FAL_KEY, salida JSON, configuración no interactiva y un comprobador opcional de actualizaciones en segundo plano. En CI, inyecta la clave desde el almacén de secretos del runner en lugar de dejarla en un script o en los registros de comandos.

Antes de lanzar un lote, fija el endpoint e inspecciona su contrato actual:

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 secuencia útil que propone la guía de fal es models → schema → run → status → download. No des por hecho que las opciones de un endpoint de vídeo funcionan en otro: la workflow skill recomienda volver a comprobar el esquema después de un error de validación.

Diseña el lote como una cola, no como un bucle de shell

La documentación oficial de genmedia cubre la ejecución asíncrona, pero no promete un comando nativo del tipo «lee este CSV y ejecuta 500 filas», ni una política automática de reintentos. Trata el CLI como el ejecutor consciente del proveedor y reserva a tu wrapper el control de la cola.

Como mínimo, cada registro de trabajo debería incluir:

Usa --async para trabajos largos de image-to-video o vídeo. Guarda inmediatamente el request_id devuelto y consulta después el estado con el ID del endpoint y el de la solicitud:

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

El patrón {request_id}_{index}.{ext} reduce el riesgo de que dos trabajos sobrescriban archivos sin avisar. Guarda el JSON junto al contenido multimedia, no en un directorio temporal separado.

Reintenta solo los fallos recuperables

Una política inicial razonable sería:

1. Reintentar los tiempos de espera de red, los reinicios de conexión, las respuestas HTTP 429 y los errores 5xx temporales.
2. Aplicar una espera exponencial, por ejemplo 2, 4, 8, 16 y después 32 segundos, añadiendo una pequeña variación aleatoria.
3. Limitar los intentos por fila, por ejemplo a cuatro envíos o cinco consultas de estado dentro de una ventana de tiempo.
4. No reintentar errores de autenticación 401/403, errores de validación del esquema 422, rechazos de seguridad ni JSON mal formado.
5. Si un envío agota el tiempo de espera después de que la solicitud pueda haber llegado al proveedor, revisa el registro guardado antes de crear una segunda solicitud de pago.

Esta política debe vivir en tu wrapper; el README público de genmedia documenta los comandos y las operaciones del ciclo de vida, no un comportamiento automático de reintento garantizado. Ante un 422, lee validation_errors, vuelve a ejecutar genmedia schema y corrige el campo indicado en lugar de reenviar la solicitud a ciegas.

Un wrapper de Python para lotes, listo para copiar

El siguiente wrapper es un punto de partida síncrono para lotes de imágenes: utiliza subprocess.run con una lista de argumentos, omite las rutas de salida ya completadas, limita el trabajo concurrente, reintenta los fallos temporales del proceso y escribe un manifiesto de forma atómica. Para trabajos de vídeo largos, guarda el envío asíncrono antes de consultar el estado: submit -> write endpoint/request_id -> restart -> poll saved request_id -> download; vuelve a enviar solo cuando no exista un ID de solicitud.

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

Para vídeo, sustituye ENDPOINT y añade las opciones específicas del esquema del endpoint. En una cadena de image-to-video, sube una vez la imagen local con genmedia upload ./frame.png --json, pasa la URL devuelta al trabajo de vídeo y conserva ambos registros en el manifiesto. El wrapper no afirma que genmedia estime o aplique por sí mismo un gasto máximo; solo evita repetir trabajo local y limita los reintentos.

Coloca un control de costes antes de generar contenido de pago

genmedia pricing <endpoint_id> --json sirve para consultar precios, no para reservar capacidad ni fijar un límite presupuestario. Úsalo antes del lote y calcula después un techo conservador a partir del número de filas, las salidas por fila, la resolución y duración configuradas y el número máximo de reintentos.

Un control práctico sería:

Guarda la respuesta de precios junto al manifiesto, porque los precios de los modelos y la disponibilidad de los endpoints pueden cambiar. Si el proveedor no ofrece una unidad comparable, presenta el resultado como un límite de número de solicitudes, no como una factura.

fal genmedia CLI frente a Replicate CLI y un script personalizado

Estas herramientas resuelven capas distintas del problema. El CLI oficial de Replicate ofrece comandos para ejecutar y transmitir predicciones, consultar esquemas de modelos, subir archivos, entrenar y gestionar modelos. genmedia resulta más útil cuando el trabajo empieza por descubrir endpoints de fal y continúa con el ciclo de colas y descargas de fal.

Mi recomendación es sencilla: utiliza genmedia directamente para explorar y un wrapper pequeño de Python para un lote de producción basado exclusivamente en fal. Pasa a una abstracción personalizada de proveedores solo cuando cambiar de proveedor sea un requisito, no porque un wrapper parezca más «enterprise».

Trata las salidas como registros y después ejecuta el control de calidad

La workflow skill pública recomienda un manifiesto compacto con el objetivo, el ID del nodo, el ID del endpoint, el ID de la solicitud, las URL de entrada, las URL de salida, los archivos descargados y las notas sobre defectos. Es mucho más útil que una carpeta llena de archivos con nombres generados automáticamente.

Antes de dar por bueno el lote, comprueba lo siguiente:

1. Cada fila con estado complete tiene un archivo local y un recibo JSON.
2. Ninguna fila con estado failed se ha omitido sin dejar constancia.
3. Las imágenes tienen las dimensiones esperadas y un tamaño distinto de cero.
4. Los vídeos se abren y tienen la duración, resolución y frecuencia de imagen esperadas; ffprobe es adecuado para esta comprobación.
5. Los prompts y los IDs de endpoint se conservan para los recursos que mantengas.
6. Volver a ejecutar el mismo manifiesto produce omisiones, no archivos duplicados.

La guía de fal insiste en mantener el contenido generado cerca de sus metadatos JSON. Esta práctica también facilita una futura migración de proveedor: puedes comparar las salidas, las solicitudes y los registros de costes sin tener que reconstruir una ejecución a partir de los nombres de archivo.

Preguntas frecuentes

¿fal genmedia CLI permite crear de forma nativa lotes de cientos de prompts?

Los comandos documentados ofrecen ejecución de modelos, gestión asíncrona del estado, salida JSON, subidas y descargas. Para crear un lote reanudable de cientos de prompts todavía necesitas un lector de manifiestos y un controlador de concurrencia.

¿genmedia es más barato que Replicate CLI?

El CLI no determina el precio del modelo del proveedor. Compara los precios del endpoint concreto, la configuración de salida, el número de reintentos y el comportamiento de transferencia de tu carga de trabajo; no tiene sentido dictar qué herramienta es «más barata» entre catálogos de modelos diferentes.

La ruta más corta es utilizar genmedia para descubrir y ejecutar, añadir un wrapper basado en manifiestos para procesar lotes y recurrir a un script personalizado solo si necesitas enrutamiento entre varios proveedores o un estado de trabajos centralizado.