fal genmedia CLI: filas em lote, novas tentativas e controle de custos

O fal genmedia CLI já resolve descoberta de modelos, inspeção de schema, IDs de requisições assíncronas, downloads e recibos em JSON. O que falta é uma camada de orquestração capaz de limitar a concorrência, repetir apenas falhas temporárias, retomar o que já foi concluído e registrar os IDs das requisições em andamento.

Instale uma vez e torne cada execução reproduzível

Para macOS e Linux, o README do projeto documenta:

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

No Windows, use o instalador documentado para PowerShell:

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

O README do genmedia CLI informa que o genmedia oferece suporte a FAL_KEY, saída em JSON, configuração não interativa e um verificador opcional de atualizações executado em segundo plano. Em CI, injete a chave pelo gerenciador de segredos do runner, em vez de colocá-la em um script ou no log de comandos.

Antes de iniciar um lote, fixe o endpoint e confira o contrato vigente:

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

A sequência útil apresentada pelo guia da fal é models → schema → run → status → download. Não presuma que as flags de um endpoint de vídeo funcionarão em outro: a skill de workflow recomenda consultar o schema novamente depois de um erro de validação.

Modele o lote como uma fila, não como um loop de shell

Os materiais oficiais do genmedia documentam a execução assíncrona, mas não prometem um comando nativo do tipo “leia este CSV e execute 500 linhas”, nem uma política automática de novas tentativas. Pense no CLI como o executor que conhece o provedor e no seu wrapper como o controlador da fila.

Um registro de job durável precisa ter, no mínimo:

Use --async em jobs longos de image-to-video ou vídeo. Salve o request_id retornado imediatamente e, depois, faça o polling usando tanto o ID do endpoint quanto o ID da requisição:

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

O padrão {request_id}_{index}.{ext} reduz o risco de dois jobs sobrescreverem arquivos silenciosamente. Mantenha o JSON ao lado da mídia, não em um diretório temporário separado.

Repita apenas falhas que podem se recuperar

Uma política inicial prática seria:

1. Repetir timeouts de rede, conexões resetadas, respostas HTTP 429 e respostas 5xx temporárias.
2. Aplicar backoff exponencial — por exemplo, 2, 4, 8, 16 e depois 32 segundos — com um pequeno jitter aleatório.
3. Limitar o número de tentativas por linha, como quatro submissões ou cinco consultas de status em uma janela de tempo.
4. Não repetir erros de autenticação 401/403, erros de validação de schema 422, recusas de segurança ou JSON malformado.
5. Se uma submissão expirar depois que a requisição possivelmente já tiver chegado ao provedor, consulte o registro salvo antes de criar uma segunda requisição paga.

Essa política fica por conta do seu wrapper; o README público do genmedia documenta comandos e operações do ciclo de vida, mas não garante um comportamento automático de novas tentativas. Em um erro 422, leia validation_errors, execute genmedia schema novamente e corrija o campo indicado, em vez de simplesmente reenviar a mesma requisição.

Um wrapper Python de lote pronto para copiar

O wrapper abaixo é um ponto de partida síncrono para lotes de imagens: usa subprocess.run com uma lista de argumentos, ignora caminhos de saída já concluídos, limita o trabalho concorrente, repete falhas temporárias do processo e grava um manifesto atômico. Para jobs longos de vídeo, persista a submissão assíncrona antes de fazer polling: submit -> write endpoint/request_id -> restart -> poll saved request_id -> download; faça uma nova submissão somente quando não existir um ID de requisição.

#!/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, troque ENDPOINT e adicione as flags específicas do schema do endpoint. Em um fluxo de image-to-video, faça o upload da imagem local uma única vez com genmedia upload ./frame.png --json, passe a URL retornada para o job de vídeo e mantenha os dois registros no manifesto. O wrapper não afirma que o próprio genmedia estima ou impõe um gasto máximo; ele apenas evita trabalho local duplicado e limita as novas tentativas.

Coloque um limite de custo antes da geração paga

genmedia pricing <endpoint_id> --json é uma consulta, não uma reserva nem um teto de orçamento. Use o comando antes do lote e calcule um limite conservador com base no número de linhas, na quantidade de saídas por linha, nas configurações de resolução/duração e no número máximo de novas tentativas.

Um controle prático seria:

Registre a resposta de preços junto com o manifesto, porque os preços dos modelos e a disponibilidade dos endpoints podem mudar. Se o provedor não oferecer uma unidade comparável, descreva o resultado como um limite de quantidade de requisições, não como uma fatura.

fal genmedia CLI vs Replicate CLI vs um script personalizado

Essas ferramentas atuam em camadas diferentes do problema. O CLI oficial da Replicate oferece comandos para executar e transmitir predictions, consultar schemas de modelos, fazer uploads, treinar e gerenciar modelos. O genmedia é mais útil quando o trabalho começa pela descoberta de endpoints da fal e passa pelo ciclo de filas e downloads da fal.

Minha recomendação é simples: use o genmedia diretamente para exploração e um pequeno wrapper Python para um lote de produção restrito à fal. Só migre para uma abstração personalizada de provedores quando trocar de provedor for um requisito — não porque um wrapper pareça mais “enterprise”.

Trate as saídas como registros e faça o QA

A skill pública de workflow recomenda um manifesto compacto com o objetivo, o ID do node, o ID do endpoint, o ID da requisição, as URLs de entrada, as URLs de saída, os arquivos baixados e as observações sobre defeitos. Isso é muito mais útil do que uma pasta cheia de arquivos com nomes gerados automaticamente.

Antes de aceitar o lote, verifique:

1. Toda linha complete tem um arquivo local e um recibo em JSON.
2. Nenhuma linha failed foi omitida silenciosamente.
3. As imagens têm as dimensões esperadas e tamanho diferente de zero.
4. Os vídeos abrem e têm a duração, resolução e taxa de quadros esperadas; o ffprobe é adequado para essa verificação.
5. Os prompts e IDs dos endpoints foram preservados para os assets mantidos.
6. Executar o mesmo manifesto novamente produz skips, e não arquivos duplicados.

O guia da fal enfatiza que a mídia gerada deve permanecer próxima dos metadados em JSON. Essa prática também facilita uma futura migração de provedor: você consegue comparar os registros de saída, requisição e custo, em vez de tentar reconstruir uma execução a partir dos nomes dos arquivos.

Perguntas frequentes

O fal genmedia CLI oferece suporte nativo a centenas de prompts em lote?

Os comandos documentados oferecem execução de modelos, acompanhamento assíncrono de status, saída em JSON, uploads e downloads. Ainda é necessário um leitor de manifestos e um controlador de concorrência para montar um lote retomável com centenas de prompts.

O genmedia é mais barato que o Replicate CLI?

O CLI não determina o preço dos modelos do provedor. Compare os preços do endpoint específico, as configurações de saída, a quantidade de novas tentativas e o comportamento de transferência do seu workload; não é significativo declarar qual ferramenta é “mais barata” entre catálogos de modelos diferentes.

Para seguir pelo caminho mais curto, use o genmedia para descoberta e execução, adicione um wrapper orientado por manifesto para os lotes e escolha um script personalizado somente quando precisar de roteamento entre provedores ou de um estado de jobs centralizado.