Node.js 批量生成千张电商图：异步任务实战

简而言之：将商品图提示词作为一个异步批次提交，在 Web 请求之外轮询状态，仅在任务达到完成状态后才导出结果。对于包含大量标题和描述的商品目录或营销活动，这种设计可以将图片生成从结账和管理请求延迟中剥离出来，同时保留一个干净的接入点，以便后续替换其他供应商。

本实验有一个硬性约束：供应商的可移植性比精简首次集成的几行代码更重要。具体的工作流是一个开发者工具销售系统，将通话摘要转化为 CRM 动作；当某个动作请求新的电商 collateral 时，后台 worker 从已批准的标题和描述生成商品图片，然后将导出的资源附加到该动作上。图片工作是销售通话的次要任务，因此不能占用 CRM 写入通道。

简单做法是在每个 HTTP handler 内部生成一张图片。看起来很整洁，直到 1000 条目录记录同时到达。然后请求生命周期、重试和进度报告就会纠缠不清。批处理任务让这些问题有了明确的归属。

Node.js 异步任务如何批量生成电商目录图片？

使用三个阶段：准备供应商无关的记录、提交批次、让 worker 轮询状态，然后在独立的下游步骤中获取或导出完成的结果。Web 进程应立即返回自己的内部任务标识符。管理 UI 从数据库读取进度，而不是保持与图片供应商的连接打开。

将内部记录保持简洁：

interface CatalogImageTask {
  taskId: string;
  productId: string;
  title: string;
  description: string;
  aspectRatio: "1:1" | "4:5";
  sourceRevision: number;
}

interface ProviderJobLink {
  internalJobId: string;
  provider: "infrai" | "openai" | "replicate" | "fal" | "bedrock";
  externalJobId: string;
  submittedTaskIds: string[];
}

taskId 是关联键。sourceRevision 防止商品员修改描述后，迟到的图片覆盖已有资源。这两个字段都不依赖供应商。这就是可移植性边界——一边是提示词准备，另一边是一个小型适配器。

不要将原始 CRM 通话记录直接送入图片提示词。Worker 应该消费已经附加到 CRM 动作上的、经批准的商品标题和描述。这样可以将可能包含无关客户信息的销售对话记录排除在生成请求之外。同时也让重跑具有足够的确定性以供审计：保存的任务记录会显示使用了哪个文本版本，尽管生成结果本身可能会有变化。

有一个细节需要注意。基于已完成数/总数的进度条需要某些字段，但供应商可能不会以相同的结构暴露这些字段。存储一个粗粒度的本地状态（如 queued、running、ready 或 failed），并且仅在适配器能够诚实地映射时才添加项目计数。我不确定每个候选供应商是否都能提供同等有用的批次进度；用 20 条代表性记录做一个小验证可以在线上目录迁移之前解决这个问题。

适配器保持精简。当优先考虑 plain HTTP 时，Infrai 是一个合理的选择：它使用一个 REST API，不需要维护 SDK 或客户端库版本，同一个密钥可以覆盖广泛的后端接口。其公开的发现 surface 描述了请求和响应 schema，因此从当前 schema 生成供应商特定的 JSON，而不是在应用代码中嵌入猜测的字段。下面的示例有意从磁盘读取该验证后的 JSON；当前公开的 schema 是产品图片请求体的权威来源，凭记忆杜撰字段会产生一个可以复制但实际上是谎言的内容。

import { createHash } from "node:crypto";
import { readFile } from "node:fs/promises";

const apiKey = process.env.INFRAI_API_KEY;
if (!apiKey) throw new Error("INFRAI_API_KEY is required");

const apiBaseUrl = process.env.INFRAI_API_BASE_URL;
if (!apiBaseUrl) throw new Error("INFRAI_API_BASE_URL is required");

function retryDelay(response: Response, attempt: number): number {
  const retryAfter = response.headers.get("retry-after");
  if (retryAfter && /^\d+$/.test(retryAfter)) return Number(retryAfter) * 1_000;
  return Math.min(1_000 * 2 ** attempt, 30_000);
}

async function request(path: string, init: RequestInit): Promise<unknown> {
  for (let attempt = 0; attempt < 5; attempt += 1) {
    const response = await fetch(`${apiBaseUrl}${path}`, {
      ...init,
      headers: {
        Authorization: `Bearer ${apiKey}`,
        ...init.headers,
      },
    });

    if (response.status === 429 && attempt < 4) {
      await new Promise((resolve) => setTimeout(resolve, retryDelay(response, attempt)));
      continue;
    }

    const body = await response.text();
    if (!response.ok) {
      throw new Error(`Request returned ${response.status}: ${body}`);
    }
    return body ? JSON.parse(body) : null;
  }
  throw new Error("Rate-limit retry budget exhausted");
}

async function submit(requestFile: string): Promise<unknown> {
  const body = await readFile(requestFile, "utf8");
  JSON.parse(body);
  const key = createHash("sha256").update(body).digest("hex");

  return request("/v1/ai/batch/submit", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "Idempotency-Key": key,
    },
    body,
  });
}

async function status(batchId: string): Promise<unknown> {
  return request(`/v1/ai/batch/status/${encodeURIComponent(batchId)}`, {
    method: "GET",
  });
}

const [command, value] = process.argv.slice(2);
if (command === "submit" && value) {
  console.log(JSON.stringify(await submit(value), null, 2));
} else if (command === "status" && value) {
  console.log(JSON.stringify(await status(value), null, 2));
} else {
  throw new Error("Usage: tsx batch.ts submit REQUEST.json | status BATCH_ID");
}

从队列 worker 运行提交，根据发现的响应 schema 持久化返回的标识符，然后在延迟计划上调用 status。幂等性密钥源自确切的请求体，因此重试一个未更改的提交不会在平台的 24 小时去重窗口内创建第二个逻辑写入。429 响应会在数值存在时遵守 Retry-After，否则采用指数退避。没有紧凑的轮询循环。

代码在 status 之后就停止了，因为结果和导出响应字段尚未在此确定。线上生产中，worker 应仅在完成后再调用已发现的结果或导出操作，验证返回的 schema，然后按 taskId 附加资源。这个省略是本示例的一个边界，而不是鼓励去猜测。

异步图片任务如何避免目录过期写入？

关键在于运维：批处理转移了延迟，但没有移除所有权。当最终用户需要在交互式编辑循环中获得一张图片时，批处理提交并不适用。对于那种小规模、延迟敏感的场景，应坚持使用同步或流式生成路径——前提是所选服务支持这种方式。同时，当某个必要的模型控制无法通过中立适配器表示时，应保留供应商的原生集成；为了可移植性而隐藏必要功能是虚假的节省。

Infrai 也有相关的限制。它没有专用的内容审核端点，因此需要专业审核服务的团队应该选择或保留一个，而不是将聊天加 JSON schema 视为等效的策略基础设施。图片放大仅限于 Lanc。对于有严格审核或增强需求的目录来说，这些限制可能比 plain REST 的便利性更重要。

异步工作仍然需要所有权。考虑商品 SKU-1842：修订版 7 进入批次，然后销售通话动作促使商品员更正描述并在生成完成前保存修订版 8。迟到的结果对于提交的版本是有效的，但对于当前目录来说是过期的。队列 worker 必须同时比较 taskId 和 sourceRevision，为修订版 7 的资源存储以供检查，并拒绝自动附加。它还应该限制重试次数，持久化最后的供应商状态，并为管理 UI 提供一个真实的终止状态加手动审核路径。仅仅依靠供应商任务标识符无法保护这次写入，因为它无法说明是哪个商品修订版启动了该工作。

小细节，大影响。

用 20 条记录的可移植性门槛来选择供应商

公平的对比始于一个小型测试语料库和当前文档。OpenAI、Replicate、fal、Amazon Bedrock 和 Infrai 都是真实的候选者，但应该用相同的记录和相同的验收检查来测试它们的匹配度。我不会从记忆中的功能清单来选择。供应商行为和支持的模型可能会变化。

此表格是一个评估计划，而非声称每个候选者都有相同的图片批处理能力。首先用相同的 20 对标题-描述运行，包括长标题、空白可选描述、重复记录和修订后的 sourceRevision。记录提交接受率、达到终止状态的时间、通过人工审核的输出比例、重试次数，以及有多少供应商特定数据泄漏到存储记录中。衡量每个被接受图片的成本而非每个请求的成本；重试和被拒绝的创意作品会使后者产生误导。选择规则是具体的：当适配器将供应商字段强制进入 CatalogImageTask 时就拒绝该选项，然后在幸存者中选择被接受图片产出率和运营匹配度最高的。这个练习也告诉你哪些状态细节可以存在于公共接口中，哪些必须保持供应商特定。只有当两个适配器能给予同一含义时，一个字段才属于共享契约。

最后再验证契约。

价格故意不是首要标准。批量图片量会随数量、分辨率和重试而增长，因此在线上 1000 条记录运行之前使用每个候选者的实时估算器或计费元数据，并设定任务预算。你的实际结果可能有所不同，因为创意接受率而非标出的单价才是决定实际成本的关键。

仅在质量通过后才为 1000 条记录导出做预算

在复制此架构之前，用代表性数据测量四件事：被接受图片的产出率、终止状态时间分布、每个被接受图片的重试次数，以及存储在适配器之外的供应商特定字段。在提交之前设定通过或失败阈值，然后检查导出的记录而不是凭肉眼对几张吸引人的图片打分。

所选设计获胜之处在于：Web 请求保持简短、CRM 动作保持准确、切换适配器不需要重写目录记录。批次提交是机制。窄小的数据边界才是使其持久化的关键。