fal genmedia CLI: очереди пакетной обработки, повторы и контроль расходов

fal genmedia CLI берёт на себя поиск моделей, просмотр схем, асинхронные request ID, скачивание файлов и JSON-квитанции. Но для надёжных пакетных запусков нужен ещё один слой: оркестратор, который ограничивает параллелизм, повторяет временные сбои, продолжает незавершённые задачи и сохраняет ID асинхронных запросов.

Установите CLI один раз и сделайте запуски воспроизводимыми

Для macOS и Linux в README проекта приведены такие команды:

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

В Windows используйте документированный установщик PowerShell:

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

Согласно README genmedia CLI, команда genmedia умеет брать ключ из FAL_KEY, выводить JSON, работать в неинтерактивном режиме и при необходимости запускать фоновую проверку обновлений. В CI передавайте ключ через хранилище секретов раннера, а не прописывайте его в скрипте или логах команд.

Перед пакетным запуском зафиксируйте endpoint и проверьте его актуальный контракт:

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

В руководстве fal предлагается понятная последовательность: models → schema → run → status → download. Не рассчитывайте, что флаги одного video endpoint подойдут другому: workflow skill отдельно советует повторно смотреть схему после ошибки валидации.

Пакет задач — это очередь, а не shell-цикл

Официальные материалы по genmedia описывают асинхронное выполнение, но не обещают встроенную команду формата «прочитать CSV и обработать 500 строк» или автоматическую политику повторов. Рассматривайте CLI как runner, знающий специфику провайдера, а свою обёртку — как контроллер очереди.

В устойчивой записи задачи должны быть как минимум следующие поля:

Для долгих задач image-to-video или video используйте --async. Сразу сохраните возвращённый request_id, а затем опрашивайте статус по ID endpoint и ID запроса:

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

Шаблон {request_id}_{index}.{ext} снижает риск, что результаты двух задач незаметно перезапишут друг друга. JSON лучше хранить рядом с медиафайлом, а не в отдельной временной директории.

Повторяйте только ошибки, которые могут пройти сами

В качестве стартовой политики подойдёт следующее:

1. Повторяйте сетевые тайм-ауты, сбросы соединения, HTTP 429 и временные ответы 5xx.
2. Используйте экспоненциальную задержку: например, 2, 4, 8, 16, затем 32 секунды, с небольшим случайным разбросом.
3. Ограничьте число попыток для каждой строки: например, четырьмя отправками или пятью проверками статуса за заданный период.
4. Не повторяйте ошибки аутентификации 401/403, ошибки валидации схемы 422, отказы по safety-фильтрам и некорректный JSON.
5. Если отправка завершилась тайм-аутом, но запрос мог дойти до провайдера, сначала проверьте сохранённую запись запроса — не создавайте сразу второй платный запрос.

Эта политика должна жить в вашей обёртке: публичный README genmedia документирует команды и этапы жизненного цикла, но не гарантирует автоматические повторы. При 422 прочитайте validation_errors, снова запустите genmedia schema и исправьте указанное поле, а не отправляйте тот же запрос вслепую.

Готовая Python-обёртка для пакетной обработки

Ниже — базовая синхронная обёртка для пакета изображений. Она вызывает subprocess.run со списком аргументов, пропускает уже готовые выходные файлы, ограничивает параллельную работу, повторяет временные ошибки процесса и атомарно записывает манифест. Для долгих video-задач сохраните асинхронную отправку до начала polling: submit -> write endpoint/request_id -> restart -> poll saved request_id -> download. Повторно отправляйте задачу, только если 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))

Для video замените ENDPOINT и добавьте флаги, требуемые схемой конкретного endpoint. В цепочке image-to-video один раз загрузите локальное изображение командой genmedia upload ./frame.png --json, передайте возвращённый URL в video-задачу и сохраните обе записи в манифесте. Эта обёртка не утверждает, что genmedia самостоятельно оценивает или ограничивает максимальные расходы: она лишь предотвращает повторную локальную работу и лимитирует число повторов.

Добавьте проверку бюджета до платной генерации

genmedia pricing <endpoint_id> --json — это справочный запрос, а не резервирование средств и не бюджетный лимит. Выполняйте его перед пакетом, затем рассчитывайте консервативный верхний предел по числу строк, количеству результатов на строку, настройкам разрешения и длительности, а также максимальному количеству повторов.

Практический набор ограничений выглядит так:

Сохраняйте ответ pricing вместе с манифестом: цены моделей и доступность endpoint могут меняться. Если провайдер не показывает сопоставимую единицу измерения, называйте результат лимитом на число запросов, а не счётом.

fal genmedia CLI, Replicate CLI или собственный скрипт

Эти инструменты решают задачи на разных уровнях. Официальный CLI Replicate предлагает команды для запуска и стриминга prediction, просмотра схем моделей, загрузок, обучения и управления моделями. genmedia удобнее, когда работа начинается с поиска endpoint в fal и проходит через очередь fal до скачивания результата.

Рекомендация простая: для исследования используйте genmedia напрямую, а для production-пакета только на fal добавьте небольшую Python-обёртку. К собственной абстракции над провайдерами стоит переходить лишь тогда, когда переключение между ними действительно необходимо, а не потому, что обёртка кажется недостаточно «enterprise».

Храните результаты как записи и проводите QA

В публичном workflow skill советуют вести компактный манифест с целью, ID узла, ID endpoint, ID запроса, URL входных данных, URL результатов, скачанными файлами и заметками о дефектах. Такой подход полезнее папки с файлами, названными генератором.

Перед приёмкой пакета проверьте следующее:

1. У каждой строки со статусом complete есть локальный файл и JSON-квитанция.
2. Ни одна строка со статусом failed не была молча пропущена.
3. У изображений ожидаемые размеры и ненулевой размер файла.
4. Видео открываются и соответствуют ожидаемым длительности, разрешению и частоте кадров; для этой проверки подходит ffprobe.
5. Для сохранённых ассетов записаны prompts и ID endpoint.
6. Повторный запуск того же манифеста пропускает готовые строки, а не создаёт дубликаты файлов.

Руководство fal подчёркивает, что сгенерированные медиа стоит хранить рядом с JSON-метаданными. Это упрощает и будущую миграцию между провайдерами: можно сравнить результаты, запросы и данные о расходах, а не пытаться восстановить запуск по именам файлов.

FAQ

Умеет ли fal genmedia CLI нативно обрабатывать сотни prompts пакетно?

Документированные команды позволяют запускать модели, обрабатывать асинхронные статусы, получать JSON, загружать и скачивать файлы. Но для возобновляемого пакета из сотен prompts всё равно понадобятся чтение манифеста и контроллер параллелизма.

genmedia дешевле Replicate CLI?

CLI не определяет цену модели у провайдера. Сравнивайте стоимость конкретных endpoint, настройки результатов, число повторов и особенности передачи данных для своей нагрузки: вывод «дешевле» на уровне инструмента не имеет смысла для разных каталогов моделей.

Самый короткий путь такой: используйте genmedia для поиска и выполнения задач, добавьте обёртку на основе манифеста для пакетной обработки и выбирайте собственный скрипт только тогда, когда нужны маршрутизация между провайдерами или централизованное состояние задач.