WaveSpeed 批量生成：自信地每天運行 1,000+ 張圖像請求

你好，各位！我是 Dora。這一切始於一個小煩惱：我需要為測試準備幾百個變體圖像，而我常用的單次請求迴圈就像推著一輛輪子卡住的購物車。我一直聽說 WaveSpeed 上的批次生成可以處理大量工作。我不需要花俏的功能，只是想讓工作輕鬆一點。

所以在十二月末和本週的幾次會議中，我在 WaveSpeed 上設置了一個簡單的批次管道，並要求它運行超過 1,000 個圖像請求。沒有英雄氣概，只是穩定的輸送量、清晰的狀態和乾淨的重試。以下是對我有效的方式、遇到的障礙，以及讓成本和錯誤不在我不注意時增加的小選擇。

批次架構概述

生成者／隊列／工作程序／存儲

我故意讓各部分保持簡單。一個小型生成者腳本收集提示詞和元數據，隊列保存作業，無狀態工作程序調用 WaveSpeed 圖像 API，存儲保存結果。每一部分都可以故障，而不會導致整個系統癱瘓。

• 生產者： 讀取包含提示詞和每張圖片設定（模型、尺寸、seed）的 CSV。每一行向佇列寫入一個作業，並附上穩定的作業 ID（輸入的雜湊值）。
• 隊列： 我曾經使用過 Redis Streams，也用過 RabbitMQ。兩者都有效。如果你還沒有運行其中任何一個，Redis 的起點更輕量。
• 工作程序： 容器化進程，拉取作業、調用 WaveSpeed、將結果寫入對象存儲（我使用了 S3）並更新狀態。它們是無狀態的，所以擴展是一個開關，而不是重建。
• 存儲： 一個用於圖像的桶，一個用於 JSON 元數據的桶。按日期和批次 ID 的簡單資料夾保持整潔。

讓我驚訝的是，當我從 100 張擴展到 1,200 張圖像時，代碼改動很少；大多數問題都是關於節奏和防止重複，而不是輸送量。

簡單架構圖

這是我心裡的圖：

生成者 → 隊列 → 工作程序 → WaveSpeed API → 存儲
             ↓                      ↑
          狀態數據庫 ←────────→ 指標/警報

• 狀態數據庫可以是相同的 Redis 或輕量級 Postgres 表。
• 指標在錯誤率或成本變得奇怪時觸發警報。

這不聰明。這就是重點。當 API 返回間歇性的 429/5xx 時，隊列會吸收它。當工作程序在運行中死亡時，另一個工作程序會在可見性超時後接管。

並發策略

安全的並行級別

在我的運行中，我從 5 個工作程序開始，每個都有 2 個進行中的請求。這給了我穩定的 8-10 張圖像/分鐘，而不會觸發限制。提升到 20 個並發請求短暫有效，但之後我看到重試激增。最好的設置不是最快的峰值，而是最平穩的平均值。

如果你想試試：找到使隊列不增長的最小工作程序數。然後逐漸增加。在改動任何東西之前，觀察 p95 延遲和錯誤率 10-15 分鐘。

速率限制意識

WaveSpeed 在文檔中發布速率限制指南，但限制因模型和帳戶而異。我添加了兩個防護措施：

• 客戶端令牌桶： 每個工作程序在調用 API 前獲取令牌。令牌按計劃的有效 RPS 補充。當我更改模型時，我調整了補充率。
• 退避紀律： 提交時的 429，以及輪詢時暫時性的 429/5xx 回應，都會觸發帶有抖動的指數退避，上限 30 秒（若有 Retry-After 則遵循）。這避免了短暫故障後的請求湧入。

我還用模型 + 尺寸標記每個作業，以便在需要時為每個模型設置單獨的並發上限。這不是花哨的，只是一個小的 switch 語句，但它幫助避免了局部熱點。

重試和冪等性

避免重複圖像

我的第一批任務有一個隱蔽的 bug：某個 worker 在 API 呼叫返回之後、寫入儲存之前當機了。佇列重試了該作業，結果我為兩次相同的生成付了費。很不愉快。

為了避免這種情況，我把每個作業拆成兩步：提交任務，然後等待任務完成。提交用的 POST 會建立一個任務（並產生費用），所以 worker 會立即把回傳的任務 ID 儲存到作業記錄中。如果作業被再次領取且已經有任務 ID，worker 就跳過提交，只輪詢該任務。我還讓儲存路徑由作業 ID 確定性地產生，這樣重試時若發現圖片已經寫入，就直接短路並把作業標記為成功。成本很低的修正，卻讓人大大鬆了口氣。

冪等性鍵實現

我使用了正規化後的提示詞 + 模型 + seed + 尺寸 + guidance 的 SHA-256。這個雜湊值會成為：

• 佇列中的作業 ID
• 儲存檔名前綴
• 作業在資料庫中的主鍵

這個鍵是我自己的記帳方式，並不是 API 的功能。WaveSpeed 不會替你對提交請求去重：兩次提交會建立兩個任務，兩個都要付費。儲存在作業記錄中的任務 ID 和先檢查儲存的做法，才是避免重複付費的關鍵。

失敗作業恢復

並非每次失敗都值得重試，提交和輪詢遵循不同的規則：

• 提交： 只送出一次。僅當請求在任務建立之前就被明確拒絕時才重試，例如 429（退避後重新送出）。遇到 5xx、逾時或連線中斷時絕不重試：任務可能已經存在。應把作業標記為 needs_check，並在決定是否重新提交之前，先查看 WaveSpeed 控制台中的歷史記錄（History）。
• 輪詢： 用退避容忍暫時性的 429、500、502、503、504 和網路錯誤（遵循 Retry-After），然後持續輪詢，直到任務變為 completed、failed、cancelled、timeout 或 deleted。
• 不要重試： 因驗證錯誤、缺少參數或明顯錯誤的輸入導致的 4xx

429 之後的提交重試我限制為最多 5 次，並使用指數退避。之後作業連同錯誤內容進入死信佇列。我每天處理一次 DLQ 中的作業：有些修正輸入後重新入列，有些附上備註封存。needs_check 的作業由人工處理，每筆記錄的錯誤都會在有任務 ID 時帶上它。這使得 1,200 張圖片的一次執行中，整體失敗率維持在 1.5% 以下。

作業狀態管理

狀態：pending/running/success/failed

我嘗試過幾種狀態結構，最簡單的那種保留了下來：

• pending： 已入列，尚未被 worker 領取
• running： 已被 worker 領取，帶有租約到期時間
• success： 圖片和中繼資料已寫入，檢查通過
• failed： 終態，附帶錯誤碼、任務 ID 和最後一次嘗試的時間戳
• needs_check： 提交結果未知（5xx、逾時、無回應），因此任務可能已存在；等待人工檢查

我新增了一個 task_id 欄位，它讓恢復變得安全，另外還有兩個很有用的選用欄位：attempt_count 和 last_response_code。它們讓儀表板更易讀，也讓除錯少了很多猜測。

作業超時處理

有兩個逾時很重要：

• 租約逾時： 如果 worker 在執行中途掛掉，作業應在 N 秒後回到 pending。我用的是 120 秒。已儲存任務 ID 的作業被重新領取後會繼續輪詢；開始了提交但從未儲存任務 ID 的作業會進入 needs_check，而不是發起新的提交。
• 請求逾時： 如果 WaveSpeed 在 N 秒內沒有回應單一請求，就中止該請求。我用的是每次呼叫 60 秒。對輪詢來說，這只是又一個需要重試的暫時性錯誤；對提交來說，則意味著結果未知，所以作業進入 needs_check。任務本身沒有總時限：有些作業執行時間很長，所以我會一直輪詢到終態。

當 API 變慢時，這兩者可能互相衝突。為了避免重複工作，我只在租約過期且 worker 的心跳停止之後，才把 running 改回 pending。心跳只是每 10 秒更新一次 Redis 雜湊。如果心跳是新的，我就延長租約。

監控和警報

錯誤率追蹤

我在運行期間觀察了三個數字：

• error_rate_5m：失敗嘗試的 5 分鐘滾動比例
• p95_latency：每個模型、每個尺寸
• retry_depth：有多少作業處於嘗試 ≥ 2 的狀態

如果 error_rate_5m > 5% 持續 10 分鐘，我自動將並發減半並向自己發送一條消息。大多數尖峰在 5 分鐘內穩定下來，無需手動調整。

成本尖峰警報

成本可能會增加。我記錄了：

• cost_per_image： WaveSpeed 報告的（如果可用），否則從計劃估計
• duplicate_prevented： 透過已儲存的任務 ID 恢復或被儲存檢查短路的作業數量
• total_estimated_cost： 累計

當 cost_per_image 相比上一小時的平均值跳升超過 30% 時，我暫停新作業進入並讓隊列清空。兩次，這在賬單漂移之前捕捉到了無意中的參數更改（更大的尺寸、不同的模型）。像這樣的安靜防護欄值得花代碼。

---

參考實現

Python 偽代碼
以下是我使用的形式。這不是完整的代碼，只是骨架：

# 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)

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);
    }
}

配置建議

• 小規模開始： 5-10 個並發請求，然後逐漸增加。觀察 p95 和 error_rate_5m，而不只是輸送量。
• 每個模型單獨配置： 並發、超時和成本預期隨著模型和尺寸而改變。
• 處處防重複： 穩定的作業 ID、提交後立即儲存的任務 ID、確定性的儲存路徑，以及以同一個值為鍵的作業表。
• 心跳和租用： 它們聽起來很麻煩，但它們可以保護你免受虛幻重複。
• 簡單儀表板： 6-8 個面板就夠了 — 隊列長度、成功/分鐘、錯誤/分鐘、p95、重試深度和成本。

如果你已經在其他地方運行批次作業，這將感到熟悉。WaveSpeed 不需要重新思考，只需要幾個仔細的防護措施。那就是我想要的。

我的運行中的最後一個說明：最順利的批次是我幾乎沒有看管的。不是因為它是「設置後忘記」，而是因為系統告訴我什麼時候需要注意，什麼時候保持安靜。那感覺像是正確的速度。

你呢？你最近在用 WaveSpeed 批量處理圖像嗎？你的並發甜蜜點是多少（我現在一直在 8-10 左右）？或者你遇到過任何隱蔽的漏洞（例如重複收費）？請在評論中分享你的設置、陷阱或提示！