Пакетная генерация на WaveSpeed: Запуск 1000+ запросов изображений ежедневно с уверенностью

Привет, ребята! Я Дора. Всё началось с небольшого раздражения: мне нужны были несколько сотен вариантов изображений для тестирования, а мой обычный цикл с одним запросом напоминал катание тачки с заклинившим колесом. Я часто слышал, что пакетная генерация на WaveSpeed может справиться с большими объёмами. Мне не нужны были чудеса. Я просто хотел, чтобы работа казалась лёгче.

Так что в течение нескольких сеансов в конце декабря и снова на этой неделе я настроил простой конвейер пакетной обработки на WaveSpeed и попросил его выполнить более 1000 запросов на генерацию изображений. Ничего героического, только стабильная пропускная способность, ясные состояния и чистые повторы. Ниже описана схема, которая сработала для меня, части, которые создавали препятствия, и небольшие решения, которые удерживали расходы и ошибки от роста, пока я был отвлечён.

Обзор архитектуры пакетной обработки

Производитель / Очередь / Рабочий процесс / Хранилище

Я намеренно сделал части простыми. Небольшой скрипт-производитель собирает подсказки и метаданные, очередь держит задания, без состояния рабочие процессы вызывают API изображений WaveSpeed, а хранилище принимает результаты. Каждая часть может отказать без того, чтобы вся система упала.

Производитель: Читает CSV с подсказками и параметрами для каждого изображения (модель, размер, seed). Он записывает одно задание за строку в очередь с ключом идемпотентности и мягким сроком.
Очередь: Я использовал Redis Streams один раз и RabbitMQ один раз. Оба сработали. Если вы ещё не запускаете ни один из них, Redis легче начать.
Рабочие процессы: Контейнеризованные процессы, которые извлекают задания, вызывают WaveSpeed, записывают результаты в объектное хранилище (я использовал S3) и обновляют статус. Они без состояния, поэтому масштабирование — это просто переключатель, а не перестройка.
Хранилище: Один бакет для изображений, один для метаданных JSON. Простые папки по дате и ID пакета держат всё в порядке.

Меня удивило то, как мало кода изменилось при масштабировании с 100 на 1200 изображений; большинство проблем касались темпа и защиты от дубликатов, а не пропускной способности.

Простая диаграмма архитектуры

Вот картинка, которую я держал в голове:

Производитель → Очередь → Рабочие → WaveSpeed API → Хранилище
                    ↓                      ↑
                 БД состояния ←────────→ Метрики/Оповещения

БД состояния может быть той же Redis или лёгкой таблицей Postgres.
Метрики активируют оповещения, когда коэффициенты ошибок или расходы становятся странными.

Это не умно. И в этом смысл. Когда API возвращает прерывистые 429/5xx, очередь их поглощает. Когда рабочий процесс падает во время выполнения, другой берёт его после истечения времени видимости.

Стратегия параллелизма

Безопасные уровни параллелизма

На моих прогонах я начал с 5 рабочих процессов, каждый с 2 одновременными запросами. Это дало мне стабильные 8–10 изображений/минуту без срабатывания лимитов. Переход на 20 одновременных запросов работал недолго, затем я увидел скачок повторов. Лучшая настройка была не самой быстрой пиковой, а самой ровной средней.

Если вы пытаетесь это сделать: найдите наименьшее количество рабочих, которое предотвращает рост вашей очереди. Затем постепенно увеличивайте. Наблюдайте за p95 латентностью и коэффициентами ошибок в течение 10–15 минут перед любыми изменениями.

Осведомлённость о лимитах скорости

WaveSpeed опубликовывает рекомендации по лимитам скорости в документации, но лимиты всё равно варьируются по модели и аккаунту. Я добавил две защиты:

Клиентский токен-бакет: Каждый рабочий получает токен перед вызовом API. Токены пополняются с эффективной RPS плана. Когда я менял модели, я корректировал скорость пополнения.
Дисциплина отката: 429 и 5xx запускают экспоненциальный откат с jitter, ограниченный 30 секундами. Это предотвращало давку после кратких сбоев.

Я также пометил каждое задание моделью + размером, чтобы можно было установить отдельные пределы параллелизма per модель при необходимости. Это не сложно, просто небольшой switch, но это помогло избежать локальных горячих точек.

Повтор и идемпотентность

Избегание дубликатов изображений

Мой первый пакет имел незаметную ошибку: рабочий процесс упал после того, как API вернул результат, но до того, как он записал в хранилище. Очередь повторила задание, и я в итоге заплатил за две одинаковые генерации. Неприятно.

Чтобы этого избежать, я сделал путь к хранилищу детерминированным из ключа идемпотентности и хеша входных данных. Если повтор находит уже написанное изображение, он прерывается и просто отмечает задание как успешное. Дешёвое исправление, большое облегчение.

Реализация ключа идемпотентности

Я использовал SHA-256 нормализованной подсказки + модели + seed + размера + guidance. Этот хеш становится:

ключом идемпотентности API (отправляется в заголовке или поле payload, в зависимости от SDK)
префиксом имени файла хранилища
первичным ключом БД для задания

Если API WaveSpeed соблюдает ключи идемпотентности (проверьте документацию для вашей конечной точки), повторные вызовы с одним и тем же ключом возвращают одинаковый результат без дополнительных расходов. Если нет, проверка storage-first всё равно предотвращает дубликаты, за которые вы платили бы дважды.

Восстановление при отказе задания

Не каждый отказ заслуживает ещё один попытку. Моё правило:

Повтор: 429, 5xx, сетевые тайм-ауты, «модель занята» или преходящие ошибки хранилища
Не повторять: 4xx с ошибками валидации, отсутствующие параметры или явно неправильные входные данные

Я ограничиваю повторы на 5 с экспоненциальным откатом. После этого задание попадает в очередь мертвых писем с полезной нагрузкой ошибки. Один раз в день я сортирую задания DLQ: некоторые исправляются с новыми входными данными и переставляются, другие архивируются с примечанием. Это удерживало мой общий коэффициент отказов ниже 1,5% для прогона из 1200 изображений.

Управление состоянием задания

Статус: ожидающий/выполняющийся/успех/отказ

Я пробовал несколько форм состояния. Самая простая прижилась:

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

Я добавил два дополнительных поля, которые окупились: attempt_count и last_response_code. Они сделали панели мониторинга более понятными и отладку менее гадательной.

Обработка тайм-аута задания

Имеет значение два тайм-аута:

Тайм-аут аренды: Если рабочий процесс падает во время выполнения, задание должно вернуться к ожидающему после N секунд. Я использовал 120 сек.
Тайм-аут API: Если WaveSpeed не отвечает в течение N секунд, прерваться и повторить с откатом. Я использовал 60 сек за вызов.

Когда API медленный, эти два могут конфликтовать. Чтобы избежать дублирования работы, я отмечаю выполняющийся → ожидающий только после истечения аренды и остановки сердцебиения рабочего. Сердцебиения были просто обновлением Redis hash каждые 10 секунд. Если сердцебиение свежее, я продлевал аренду.

Мониторинг и оповещения

Отслеживание коэффициента ошибок

Я наблюдал три числа во время прогонов:

error_rate_5m: скользящая 5-минутная доля неудачных попыток
p95_latency: per модель, per размер
retry_depth: сколько заданий находятся на попытке ≥ 2

Если error_rate_5m > 5% в течение 10 минут, я автоматически сокращаю параллелизм вдвое и отправляю себе записку. Большинство скачков улеглись в течение пяти минут без ручного вмешательства.

Оповещения о скачках расходов

Расходы могут расти. Я записывал:

cost_per_image: сообщается WaveSpeed, если доступно, иначе оценивается из плана
duplicate_prevented: счёт коротких замыканий хранилища
total_estimated_cost: совокупная

Когда cost_per_image скачкал более чем на 30% от среднего за последний час, я паузировал приём новых заданий и позволял очереди истощиться. Дважды это поймало непреднамеренные изменения параметров (больше размеры, другая модель) до того, как счёт вырос. Тихие защиты вроде этой стоят своих нескольких строк кода.

Эталонная реализация

Псевдокод Python
Ниже описана схема, которую я использовал. Это не полный код, только скелет:

# producer.py
for row in csv_rows:
    key = hash_inputs(row)
    job = { "id": key, "inputs": row, "deadline": now+6*3600 }
    queue.push(job)

# worker.py
while True:
    job = queue.lease(timeout=120)
    if not job:
        sleep(1)
        continue
    try:
        record_heartbeat(job.id)
        resp = wavespeed.generate_image(inputs=job.inputs, idempotency_key=job.id, timeout=60)
        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 Retryable as e:
        mark_retry(job.id, e)
        backoff_sleep(job.attempt)
    except Fatal as e:
        mark_failed(job.id, e)
    finally:
        queue.release(job)

Псевдокод Node.js

// producer.mjs
for (const row of rows) {
    const key = hashInputs(row);
    queue.push({ id: key, inputs: row, deadline: Date.now() + 6 * 3600e3 }); // 6 hours
}


// worker.mjs
while (true) {
    const job = await queue.lease(120); // lease timeout in seconds

    if (!job) { // 原来的 ".job" 改为 "!job"
        await delay(1000);
        continue;
    }

    try {
        await heartbeat(job.id);

        const resp = await wavespeed.generateImage(
            { ...job.inputs, idempotencyKey: job.id },
            { timeout: 60000 } // 60 seconds
        );

        const path = makePath(job.id, job.inputs);

        if (!(await storage.exists(path))) { // 原来的 ".(await ...)" 修正
            await storage.write(path, resp.image);
            await storage.write(path + '.json', resp.metadata);
        }

        await markSuccess(job.id);
    } catch (e) {
        if (isRetryable(e)) {
            await markRetry(job.id, e);
        } else {
            await markFailed(job.id, e);
        }
    } finally {
        await queue.release(job);
    }
}

Если вы уже выполняете пакетные задания где-то в другом месте, это будет знакомо. WaveSpeed не требовал переосмысления, просто несколько осторожных защит. Это то, что я хотел.

Последняя заметка из моих прогонов: самые гладкие пакеты были теми, которые я едва смотрел. Не потому, что это было «установи и забудь», а потому, что система говорила мне, когда ей нужно внимание, и молчала, когда ей оно не нужно. Это кажется правильным видом скорости.

А как у вас? Вы недавно занимались пакетной обработкой изображений с WaveSpeed? Какая ваша оптимальная точка для параллелизма (я постоянно около 8–10)? Или вы столкнулись с какими-нибудь хитрыми ошибками (вроде дублирующихся начислений)? Не стесняйтесь делиться своей настройкой, подводными камнями или советами в комментариях!