WaveSpeed 批量生成：自信运行 1000+ 日均图像请求

你好！我是Dora。这一切始于一个小烦恼：我需要几百张变体图像来进行测试，而我通常的单请求循环感觉就像推着一辆轮子卡住的购物车。我一直听说WaveSpeed上的批量生成可以处理大量工作。我不需要什么花哨的功能，我只是想让工作感觉更轻松。

所以在12月下旬和本周的几个会话中，我在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：有多少作业在attempt ≥ 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左右）？或者你遇到过任何狡猾的错误（比如重复收费）？欢迎在评论中分享你的设置、陷阱或提示！