Génération en lot sur WaveSpeed : Exécutez 1 000+ demandes d'images quotidiennes en toute confiance

Bonjour à tous ! Je m’appelle Dora. Tout a commencé par une petite contrariété : j’avais besoin de quelques centaines d’images variantes pour un test, et ma boucle habituelle de requêtes uniques me donnait l’impression de pousser un chariot avec une roue bloquée. J’entendais régulièrement dire que la Génération par lots sur WaveSpeed pouvait gérer des volumes importants. Je ne cherchais pas l’extraordinaire. Je voulais juste que le travail soit plus léger.

Alors, au fil de quelques sessions fin décembre et à nouveau cette semaine, j’ai mis en place un simple pipeline de lot sur WaveSpeed et je lui ai demandé de traiter plus de 1 000 requêtes d’images. Rien d’héroïque, juste un débit régulier, des états clairs et des tentatives propres. Voici la structure qui a fonctionné pour moi, les parties qui ont posé problème et les petits choix qui ont empêché les coûts et les erreurs de s’envoler pendant que mon attention était ailleurs.

Aperçu de l’architecture du lot

Producteur / File d’attente / Travailleur / Stockage

J’ai volontairement gardé les pièces simples. Un petit script producteur collecte les invites et les métadonnées, une file d’attente retient les tâches, des travailleurs sans état appellent l’API d’images WaveSpeed, et le stockage récupère les résultats. Chaque partie peut échouer sans faire tomber tout le système.

Producteur : lit un CSV de prompts et de paramètres par image (modèle, taille, seed). Il écrit un job par ligne dans la file avec un ID de job stable (un hash des entrées).
File d’attente : J’ai utilisé Redis Streams une fois et RabbitMQ une autre fois. Les deux ont fonctionné. Si vous n’en exécutez pas déjà, Redis est plus léger pour commencer.
Travailleurs : Processus conteneurisés qui récupèrent les tâches, appellent WaveSpeed, écrivent les résultats dans le stockage d’objets (j’ai utilisé S3) et mettent à jour l’état. Ils sont sans état, donc la mise à l’échelle est un bouton, pas une reconstruction.
Stockage : Un compartiment pour les images, un pour les métadonnées JSON. Des dossiers simples par date et ID de lot gardent tout organisé.

Ce qui m’a surprise, c’est combien peu de code a changé quand j’ai augmenté de 100 à 1 200 images ; la plupart des problèmes concernaient le rythme et la protection contre les doublons, pas le débit.

Diagramme d’architecture simple

Voici l’image que je gardais en tête :

Producteur → File d'attente → Travailleurs → API WaveSpeed → Stockage
             ↓                      ↑
          BD d'état ←────────→ Métriques/Alertes

La BD d’état peut être le même Redis ou une table Postgres légère.
Les Métriques alimentent les alertes quand les taux d’erreur ou les coûts deviennent bizarres.

Ce n’est pas compliqué. C’est le but. Quand l’API retourne des 429/5xx intermittents, la file d’attente les absorbe. Quand un travailleur meurt en cours d’exécution, un autre la reprend après l’expiration de la visibilité.

Stratégie de concurrence

Niveaux de parallélisme sûr

Dans mes exécutions, j’ai commencé avec 5 travailleurs, chacun faisant 2 requêtes en vol. Cela m’a donné un régime stable de 8–10 images/minute sans déclencher de limites. Passer à 20 requêtes simultanées a fonctionné brièvement, puis j’ai vu les tentatives monter en flèche. Le meilleur réglage n’était pas le pic le plus rapide, mais la moyenne la plus plate.

Si vous essayez ceci : trouvez le plus petit nombre de travailleurs qui empêche votre file d’attente de croître. Puis montez graduellement. Observez la latence p95 et les taux d’erreur pendant 10–15 minutes avant de toucher à quoi que ce soit.

Sensibilisation aux limites de débit

WaveSpeed publie des conseils sur les limites de débit dans la documentation, mais les limites varient encore selon le modèle et le compte. J’ai ajouté deux garde-fous :

Seau de jetons côté client : Chaque travailleur acquiert un jeton avant d’appeler l’API. Les jetons se reconstituent au RPS effectif du plan. Quand j’ai changé de modèle, j’ai ajusté la reconstitution.
Discipline de backoff : un 429 à la soumission, ainsi que les réponses 429/5xx transitoires pendant l’interrogation, déclenchent un backoff exponentiel avec jitter, plafonné à 30 secondes (en respectant Retry-After s’il est présent). Cela a évité les effets de ruée après de courtes pannes.

J’ai également marqué chaque tâche avec modèle + taille pour pouvoir définir des plafonds de concurrence distincts par modèle si nécessaire. Ce n’est pas sophistiqué, juste une petite déclaration switch, mais cela a aidé à éviter les points chauds localisés.

Tentatives et idempotence

Éviter les images dupliquées

Mon premier batch avait un bug discret : un worker a planté après le retour de l’appel API, avant d’écrire dans le stockage. La file a relancé le job et j’ai fini par payer deux générations identiques. Pas drôle.

Pour l’éviter, j’ai découpé chaque job en deux étapes : soumettre la tâche, puis l’attendre. Le POST de soumission crée une tâche (et un débit), donc le worker enregistre aussitôt l’ID de tâche renvoyé dans la ligne du job. Si un job est repris et possède déjà un ID de tâche, le worker saute la soumission et interroge seulement cette tâche. J’ai aussi rendu le chemin de stockage déterministe à partir de l’ID du job : si une reprise trouve l’image déjà écrite, elle court-circuite et marque simplement le job comme réussi. Correctif peu coûteux, grand soulagement.

Implémentation de la clé d’idempotence

J’ai utilisé un SHA-256 du prompt normalisé + modèle + seed + taille + guidance. Ce hash devient :

l’ID du job dans la file
le préfixe du nom de fichier dans le stockage
la clé primaire du job en base de données

Cette clé relève de ma propre comptabilité, pas d’une fonctionnalité de l’API. WaveSpeed ne dédoublonne pas les soumissions à votre place : deux soumissions créent deux tâches, et vous payez les deux. L’ID de tâche enregistré dans la ligne du job et la vérification préalable du stockage évitent de payer deux fois.

Récupération des tâches échouées

Tous les échecs ne méritent pas une nouvelle tentative, et la soumission et l’interrogation suivent des règles différentes :

Soumission : envoyez-la une seule fois. Ne réessayez que si la requête a été clairement rejetée avant qu’une tâche existe, comme un 429 (backoff, puis nouvel envoi). Ne réessayez jamais après un 5xx, un délai dépassé ou une connexion coupée : la tâche peut déjà exister. Marquez plutôt le job needs_check et consultez l’historique (History) de votre tableau de bord WaveSpeed avant de décider de soumettre à nouveau.
Interrogation : tolérez les 429, 500, 502, 503, 504 transitoires et les erreurs réseau avec un backoff (en respectant Retry-After), puis continuez jusqu’à ce que la tâche soit completed, failed, cancelled, timeout ou deleted.
Ne pas réessayer : les 4xx avec erreurs de validation, paramètres manquants ou entrée manifestement invalide

Je plafonne les nouvelles soumissions après un 429 à 5, avec backoff exponentiel. Ensuite, le job atterrit dans une file de lettres mortes avec la charge utile de l’erreur. Une fois par jour, je trie les jobs de la DLQ : certains reçoivent une entrée corrigée et sont remis en file, d’autres sont archivés avec une note. Les jobs en needs_check se règlent à la main, et chaque erreur journalisée inclut l’ID de tâche quand il y en a un. Cela a maintenu mon taux d’échec global sous 1,5 % sur un lot de 1 200 images.

Gestion de l’état des tâches

Statut : pending/running/success/failed

J’ai essayé plusieurs modèles d’états. Le plus simple est resté :

pending : en file, pas encore pris par un worker
running : pris par un worker avec une expiration de bail
success : image et métadonnées écrites, contrôles réussis
failed : terminal, avec code d’erreur, ID de tâche et horodatage de la dernière tentative
needs_check : le résultat de la soumission est inconnu (5xx, délai dépassé, pas de réponse), donc une tâche peut exister ; en attente d’un examen manuel

J’ai ajouté un champ task_id, qui rend la reprise sûre, ainsi que deux champs optionnels utiles : attempt_count et last_response_code. Ils ont rendu les tableaux de bord plus lisibles et le débogage moins hasardeux.

Gestion du délai d’expiration de la tâche

Deux délais comptent :

Délai de bail : si un worker meurt en cours d’exécution, le job doit revenir à pending après N secondes. J’ai utilisé 120 s. Un job repris avec un ID de tâche enregistré reprend l’interrogation ; un job qui a commencé une soumission sans jamais enregistrer d’ID de tâche passe en needs_check, pas dans une nouvelle soumission.
Délai de requête : si WaveSpeed ne répond pas à une requête en N secondes, abandonnez cette requête. J’ai utilisé 60 s par appel. Pour une interrogation, c’est juste une erreur transitoire à réessayer. Pour une soumission, cela signifie que le résultat est inconnu, donc le job passe en needs_check. Il n’y a pas de limite globale sur la tâche elle-même : certains jobs durent longtemps, donc je continue d’interroger jusqu’à un statut final.

Quand l’API est lente, les deux peuvent se heurter. Pour éviter le travail en double, je ne repasse running → pending qu’après l’expiration du bail et l’arrêt du heartbeat du worker. Les heartbeats n’étaient qu’une mise à jour de hash Redis toutes les 10 secondes. Si le heartbeat est récent, je prolonge le bail.

Surveillance et alertes

Suivi du taux d’erreur

J’ai observé trois chiffres pendant les exécutions :

error_rate_5m : proportion roulante sur 5 minutes des tentatives échouées
p95_latency : par modèle, par taille
retry_depth : combien de tâches sont à tentative ≥ 2

Si error_rate_5m > 5% pendant 10 minutes, j’ai automatiquement réduit de moitié la concurrence et me suis envoyé une note. La plupart des pics se sont résorbés en cinq minutes sans bidouille manuelle.

Alertes de pics de coûts

Les coûts peuvent s’insinuer. J’ai enregistré :

cost_per_image : signalé par WaveSpeed si disponible, sinon estimé à partir du plan
duplicate_prevented : nombre de jobs repris depuis un ID de tâche enregistré ou court-circuités par la vérification du stockage
total_estimated_cost : cumulatif

Quand cost_per_image a grimpé de plus de 30% par rapport à la moyenne de la dernière heure, j’ai suspendu la prise de nouvelles tâches et laissé la file d’attente se vider. Deux fois, cela a attrapé des changements de paramètres involontaires (tailles plus grandes, modèle différent) avant que la facture ne dérive. Des garde-fous silencieux comme celui-ci valent bien leurs quelques lignes de code.

Implémentation de référence

Pseudo-code Python Voici la forme que j’ai utilisée. Ce n’est pas du code complet, juste le squelette :

# producer.py
for row in csv_rows:
    key = hash_inputs(row)  # stable job ID
    queue.push({"id": key, "inputs": row})

# worker.py
while True:
    job = queue.lease(timeout=120)
    if not job:
        sleep(1)
        continue
    try:
        record_heartbeat(job.id)

        # Step 1: submit once. A saved task ID means the submit already happened.
        if not job.task_id:
            job.task_id = submit_once(job)
            save_task_id(job.id, job.task_id)

        # Step 2: poll until a terminal status.
        resp = poll_until_terminal(job.task_id)

        path = storage_path(job.id, job.inputs)
        if not storage.exists(path):
            storage.write(path, resp.image)
            storage.write(path + ".json", resp.metadata)
        mark_success(job.id)
    except SubmitUnconfirmed as e:  # 5xx, timeout, dropped connection: a task may exist
        mark_needs_check(job.id, e)  # do NOT resubmit
    except Fatal as e:  # rejected input, failed/cancelled task, 429s exhausted
        mark_failed(job.id, e, task_id=job.task_id)
    finally:
        queue.release(job)


def submit_once(job):
    for attempt in range(5):
        try:
            return wavespeed.submit(job.model, job.inputs, timeout=60).task_id  # one POST
        except RateLimited as e:  # 429: rejected, no task was created
            backoff_sleep(attempt, e.retry_after)
        # 5xx / timeout / no response raises SubmitUnconfirmed and is never retried
    raise Fatal("rate limited on submit")


def poll_until_terminal(task_id):
    attempt = 0
    while True:
        try:
            r = wavespeed.get_result(task_id, timeout=60)
        except TransientError as e:  # 429/500/502/503/504 or network error
            attempt += 1
            backoff_sleep(attempt, e.retry_after)
            continue
        if r.status == "completed":
            return r
        if r.status in ("failed", "cancelled", "timeout", "deleted"):
            raise Fatal(f"task {task_id} {r.status}: {r.error}")
        sleep(3)

Pseudo-code Node.js

// producer.mjs
for (const row of rows) {
    const key = hashInputs(row); // stable job ID
    queue.push({ id: key, inputs: row });
}

// worker.mjs
async function submitOnce(job) {
    for (let attempt = 0; attempt < 5; attempt++) {
        try {
            const resp = await wavespeed.submit(job.model, job.inputs, { timeout: 60000 }); // one POST
            return resp.taskId;
        } catch (e) {
            // 5xx / timeout / no response is a SubmitUnconfirmed error: never retried
            if (!(e instanceof RateLimited)) throw e;
            await backoffSleep(attempt, e.retryAfter); // 429: rejected, no task was created
        }
    }
    throw new Fatal("rate limited on submit");
}

async function pollUntilTerminal(taskId) {
    let attempt = 0;
    while (true) {
        let r;
        try {
            r = await wavespeed.getResult(taskId, { timeout: 60000 });
        } catch (e) {
            if (!(e instanceof TransientError)) throw e; // 429/500/502/503/504 or network error
            await backoffSleep(++attempt, e.retryAfter);
            continue;
        }
        if (r.status === "completed") return r;
        if (["failed", "cancelled", "timeout", "deleted"].includes(r.status)) {
            throw new Fatal(`task ${taskId} ${r.status}: ${r.error}`);
        }
        await delay(3000);
    }
}

while (true) {
    const job = await queue.lease(120); // lease timeout in seconds
    if (!job) {
        await delay(1000);
        continue;
    }

    try {
        await heartbeat(job.id);

        // A saved task ID means the submit already happened.
        if (!job.taskId) {
            job.taskId = await submitOnce(job);
            await saveTaskId(job.id, job.taskId);
        }

        const resp = await pollUntilTerminal(job.taskId);

        const path = makePath(job.id, job.inputs);
        if (!(await storage.exists(path))) {
            await storage.write(path, resp.image);
            await storage.write(path + '.json', resp.metadata);
        }

        await markSuccess(job.id);
    } catch (e) {
        if (e instanceof SubmitUnconfirmed) {
            await markNeedsCheck(job.id, e); // a task may exist: do NOT resubmit
        } else {
            await markFailed(job.id, e, job.taskId);
        }
    } finally {
        await queue.release(job);
    }
}

Recommandations de configuration

Commencez petit : 5–10 requêtes simultanées, puis montez graduellement. Observez p95 et error_rate_5m, pas juste le débit.
Configurations séparées par modèle : la concurrence, le délai d’expiration et les attentes de coût changent avec le modèle et la taille.
Protection contre les doublons partout : un ID de job stable, l’ID de tâche enregistré juste après la soumission, un chemin de stockage déterministe et une table de jobs clé par la même valeur.
Battements de cœur et baux : ils sonnent pointilleux, mais ils vous sauvent des doublons fantômes.
Tableaux de bord simples : 6–8 panneaux suffisent — longueur de la file d’attente, succès/min, erreurs/min, p95, profondeur de tentative et coûts.

Si vous exécutez déjà des tâches par lots ailleurs, cela vous semblera familier. WaveSpeed n’a pas nécessité une refonte, juste quelques garde-fous soignés. C’est ce que je voulais.

Une dernière note de mes exécutions : les lots les plus fluides étaient ceux que je regardais à peine. Non pas parce que c’était « réglé et oublié », mais parce que le système me disait quand il avait besoin d’attention et restait silencieux quand ce n’était pas le cas. Cela semble être la bonne sorte de vitesse.

Et vous ? Avez-vous traité des images par lots avec WaveSpeed récemment ? Quel est votre point optimal pour la concurrence (je suis régulièrement autour de 8–10 en ce moment) ? Ou avez-vous rencontré des bugs sournois (comme des frais dupliqués) ? N’hésitez pas à partager votre configuration, vos pièges ou vos conseils dans les commentaires !