fal genmedia CLI: Batch Kuyrukları, Yeniden Denemeler ve Maliyet Kontrolü

fal genmedia CLI; model keşfi, şema inceleme, asenkron istek kimlikleri, indirme işlemleri ve JSON makbuzlarını zaten yönetiyor. Eksik kalan katman ise eşzamanlılığı sınırlayan, geçici hataları yeniden deneyen, tamamlanan işleri atlayarak kaldığı yerden devam eden ve asenkron işlerde kullanılan istek kimliklerini kaydeden bir orkestratör.

Bir kez kurun, her çalıştırmayı tekrarlanabilir hâle getirin

macOS ve Linux için projenin README dosyasında şu kurulum adımları yer alıyor:

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

Windows tarafında belgelenen PowerShell yükleyicisi kullanılıyor:

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

genmedia CLI README dosyasına göre genmedia; FAL_KEY, JSON çıktısı, etkileşimsiz kurulum ve isteğe bağlı arka plan güncelleme denetleyicisini destekliyor. CI ortamında anahtarı bir betiğe veya komut günlüğüne yazmak yerine çalıştırıcının gizli değişken deposu üzerinden sağlayın.

Batch çalıştırmadan önce endpoint’i sabitleyin ve güncel sözleşmesini inceleyin:

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 rehberinin önerdiği pratik sıra şu: models → schema → run → status → download. Bir video endpoint’inde çalışan seçeneklerin başka bir endpoint’te de geçerli olduğunu varsaymayın. İş akışı becerisi, doğrulama hatası alındığında şemanın yeniden kontrol edilmesini özellikle öneriyor.

Batch işini bir shell döngüsü değil, kuyruk olarak tasarlayın

Resmî genmedia kaynakları asenkron çalıştırmayı belgeliyor; ancak “bu CSV dosyasını oku ve 500 satırı çalıştır” şeklinde yerleşik bir komut ya da otomatik bir yeniden deneme politikası vaat etmiyor. CLI’ı sağlayıcıya özgü çalıştırıcı, kendi sarmalayıcınızı da kuyruğu yöneten katman olarak konumlandırın.

Dayanıklı bir iş kaydında en az şu alanlar bulunmalı:

Uzun süren görselden videoya veya video işlerinde --async kullanın. Dönen request_id değerini hemen kaydedin, ardından hem endpoint kimliği hem de istek kimliğiyle durum sorgulayın:

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} kalıbı, iki işin fark edilmeden aynı dosyanın üzerine yazma riskini azaltır. JSON dosyasını ayrı bir geçici klasöre değil, medya dosyasının yanına koyun.

Yalnızca toparlanabilecek hataları yeniden deneyin

Başlangıç için uygulanabilir bir politika şöyle olabilir:

1. Ağ zaman aşımı, bağlantı sıfırlanması, HTTP 429 ve geçici 5xx yanıtlarını yeniden deneyin.
2. Üstel geri çekilme kullanın; örneğin 2, 4, 8, 16 ve ardından 32 saniye bekleyin, ayrıca küçük bir rastgele sapma ekleyin.
3. Satır başına deneme sayısını sınırlayın; örneğin dört gönderim veya belirli bir zaman aralığında beş durum sorgusu.
4. 401/403 kimlik doğrulama hatalarını, 422 şema doğrulama hatalarını, güvenlik reddetmelerini ve hatalı JSON’u yeniden denemeyin.
5. Gönderim zaman aşımına uğradığında isteğin sağlayıcıya ulaşmış olma ihtimali varsa ikinci ve ücretli bir istek oluşturmadan önce kaydedilmiş istek kaydını inceleyin.

Bu politika sizin sarmalayıcınıza ait olmalı. Genel kullanıma açık genmedia README’si komutları ve yaşam döngüsü işlemlerini belgeliyor; garantili bir otomatik yeniden deneme davranışı tanımlamıyor. 422 hatasında validation_errors alanını okuyun, genmedia schema komutunu yeniden çalıştırın ve körlemesine tekrar göndermek yerine belirtilen alanı düzeltin.

Kopyalayıp kullanabileceğiniz bir Python batch sarmalayıcısı

Aşağıdaki sarmalayıcı, senkron bir görsel batch işlemi için başlangıç noktasıdır. Argüman listesiyle subprocess.run kullanır, tamamlanmış çıktı yollarını atlar, eşzamanlı iş sayısını sınırlar, geçici süreç hatalarını yeniden dener ve atomik bir manifest yazar. Uzun video işlerinde asenkron gönderimi polling işleminden önce kalıcı hâle getirin: submit -> write endpoint/request_id -> restart -> poll saved request_id -> download; yalnızca ortada bir istek kimliği yoksa yeniden gönderim yapın.

#!/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 için ENDPOINT değerini değiştirin ve endpoint’in şemasına özgü seçenekleri ekleyin. Görselden videoya uzanan bir zincirde yerel görseli genmedia upload ./frame.png --json ile bir kez yükleyin, dönen URL’yi video işine aktarın ve her iki kaydı da manifestte tutun. Bu sarmalayıcı, genmedia’nın kendisinin maksimum harcamayı tahmin ettiğini veya uyguladığını iddia etmiyor; yalnızca yerel işlerin yinelenmesini önlüyor ve yeniden denemeleri sınırlıyor.

Ücretli üretimden önce maliyet kapısı koyun

genmedia pricing <endpoint_id> --json bir fiyat sorgusudur; rezervasyon veya bütçe limiti değildir. Batch başlamadan önce bu komutu kullanın. Ardından satır sayısı, satır başına çıktı adedi, çözünürlük ve süre ayarları ile maksimum yeniden deneme sayısını temel alarak temkinli bir üst sınır hesaplayın.

Uygulanabilir bir kontrol tablosu şöyle:

Model fiyatları ve endpoint erişilebilirliği değişebileceğinden fiyat yanıtını manifestle birlikte saklayın. Sağlayıcı karşılaştırılabilir bir birim sunmuyorsa sonucu fatura değil, istek sayısı üst sınırı olarak etiketleyin.

fal genmedia CLI mı, Replicate CLI mı, özel betik mi?

Bu araçlar problemin farklı katmanlarını çözüyor. Replicate’ın resmî CLI’ı; tahminleri çalıştırma ve akış olarak aktarma, model şemalarını inceleme, yüklemeler, eğitim ve model yönetimi için komutlar sunuyor. İşiniz fal endpoint’lerini keşfetmekle başlayıp medyayı fal’ın kuyruk ve indirme yaşam döngüsünden geçirmekle devam ediyorsa genmedia daha uygun.

Benim önerim net: Keşif için doğrudan genmedia kullanın; yalnızca fal üzerinde çalışan üretim batch’leri için küçük bir Python sarmalayıcısı ekleyin. Sağlayıcı değiştirmek gerçekten bir gereksinim hâline gelmedikçe, bir sarmalayıcı daha “kurumsal” görünüyor diye özel sağlayıcı soyutlamasına geçmeyin.

Çıktıları dosya değil kayıt olarak ele alın, ardından kalite kontrol yapın

Herkese açık iş akışı becerisi; hedefi, node ID’sini, endpoint ID’sini, istek ID’sini, girdi URL’lerini, çıktı URL’lerini, indirilen dosyaları ve kusur notlarını içeren kompakt bir manifest öneriyor. Bu yapı, oluşturulmuş isimlere sahip dosyalarla dolu bir klasörden çok daha kullanışlıdır.

Batch’i kabul etmeden önce şu kontrolleri yapın:

1. complete durumundaki her satırın yerel bir dosyası ve JSON makbuzu bulunuyor mu?
2. Hiçbir failed satırı sessizce atlanmış mı?
3. Görseller beklenen boyutlara ve sıfırdan büyük dosya boyutuna sahip mi?
4. Videolar açılıyor mu; beklenen süre, çözünürlük ve kare hızına sahip mi? Bu kontrol için ffprobe uygundur.
5. Sakladığınız varlıkların prompt’ları ve endpoint kimlikleri korunuyor mu?
6. Aynı manifest yeniden çalıştırıldığında yinelenen dosyalar üretmek yerine mevcut işler atlanıyor mu?

fal rehberi, oluşturulan medyanın JSON meta verilerinin yakınında tutulmasını vurguluyor. Bu yaklaşım ileride sağlayıcı değişikliğini de kolaylaştırır: bir çalıştırmayı yalnızca dosya adlarından yeniden kurmaya çalışmak yerine çıktı, istek ve maliyet kayıtlarını karşılaştırabilirsiniz.

Sık sorulan sorular

fal genmedia CLI yüzlerce prompt’u yerleşik olarak batch hâlinde çalıştırabiliyor mu?

Belgelenen komutlar model çalıştırma, asenkron durum takibi, JSON çıktısı, yüklemeler ve indirmeler sunuyor. Ancak yüzlerce prompt’tan oluşan, kaldığı yerden devam edebilen bir batch için yine de manifest okuyucu ve eşzamanlılık denetleyicisi gerekiyor.

genmedia, Replicate CLI’dan daha ucuz mu?

CLI, sağlayıcının model fiyatını belirlemez. Kullandığınız iş yükü için ilgili endpoint fiyatlarını, çıktı ayarlarını, yeniden deneme sayısını ve veri aktarımı davranışını karşılaştırın; farklı model katalogları arasında araç düzeyinde “daha ucuz” sonucu anlamlı değildir.

En kısa yol şu: Keşif ve çalıştırma için genmedia kullanın, batch işlemleri için manifest tabanlı bir sarmalayıcı ekleyin ve yalnızca birden fazla sağlayıcı arasında yönlendirme ya da merkezi iş durumu gerektiğinde özel bir betiğe geçin.