电商团队做商品图这件事，最怕的不是没图，而是图多了之后失控。一个 SKU 动辄要出主图、场景图、细节图、卖点图、尺寸对比图，几十个 SKU 叠起来就是几百上千张。早期我们靠人工一张张跑，提示词复制粘贴、生成完手动改名、失败了自己重试，一天下来人累得够呛，产出还不稳定。后来我把这套流程拆成了三块：提示词模板化、队列调度、幂等命名，才算真正把批量生成跑顺。这篇就聊聊我在 AI 商品图批量生成上的完整实践，包括为什么这么设计、每一步怎么落地、踩过哪些坑，以及哪些参数是真正影响成败的关键。不管你是刚接触批量出图的运营，还是想把这套流程工程化的开发者，都能直接抄作业。
1. 为什么商品图批量生成必须走工程化路线
1.1 手工出图的三个死结
先说清楚问题，不然工程化就是空谈。手工出图在 SKU 少的时候还能撑，一旦上量，三个死结立刻暴露。
第一个是提示词漂移。同一个商品，今天写"白色陶瓷马克杯，简约北欧风，纯白背景"，明天可能写成"白色杯子，北欧简约，白底"。人脑觉得差不多，模型出来的图风格却差很远。批量场景下，这种不一致会直接毁掉店铺的视觉统一性。
第二个是并发失控。很多人一开始用脚本 for 循环挨个调接口，看着能跑，但一旦某个请求卡住，整个循环就堵死。更糟的是有些平台对并发有限制，你一股脑发出去，轻则限流重则封禁，排查起来非常痛苦。
第三个是文件命名混乱。生成出来的图默认叫output_1.png、output_2.png，你根本不知道哪张对应哪个 SKU、哪个角度。等到要上传到后台时，只能一张张点开看，几百张图能看瞎眼。
这三个死结的本质，是把"生成"当成了单次动作，而不是一条流水线。工程化的核心，就是把这条流水线拆成可复用、可重试、可追溯的环节。
1.2 工程化要解决的四个核心诉求
拆解下来，批量生成真正要满足的诉求有四个：
- 一致性：同一批商品的提示词结构必须统一，风格才能统一。
- 可控性：并发数、重试次数、超时时间都要能配置，不能任由脚本乱跑。
- 可追溯：每张图从哪来、用的什么提示词、对应哪个 SKU，必须能查。
- 可重入：任务失败或中断后，重跑不能产生重复文件，也不能漏掉。
这四个诉求，正好对应标题里的三个关键词：模板化解决一致性，队列调度解决可控性，幂等命名解决可追溯和可重入。下面逐个展开。
2. 提示词模板化：把"灵感"变成"参数"
2.1 模板化的本质是变量抽取
很多人以为模板化就是写个字符串拼接，其实没那么简单。模板化的本质，是把提示词里会变的部分和不变的部分彻底分开。
以商品图为例，一段完整的提示词通常包含这些维度：
| 维度 | 是否变化 | 举例 |
|---|---|---|
| 主体 | 变 | 马克杯、帆布包、蓝牙耳机 |
| 材质 | 变 | 陶瓷、帆布、金属 |
| 颜色 | 变 | 纯白、米色、哑光黑 |
| 风格 | 相对固定 | 简约北欧、日系清新 |
| 背景 | 相对固定 | 纯白背景、浅灰渐变 |
| 光影 | 固定 | 柔和顶光、无硬阴影 |
| 画质词 | 固定 | 高清、商业摄影、细节清晰 |
把"变"的抽成变量，"不变"的固化成模板骨架，提示词就从一段自由文本变成了一个可填充的结构。这样做的好处是，同一批商品只要换变量值，风格骨架完全一致，出图自然统一。
2.2 模板结构怎么设计才不僵化
模板设计最容易走极端：要么太死，所有商品套一个模板，出来的图千篇一律；要么太活，变量多到跟没模板一样。我的经验是分层设计：
- 基础层：所有商品共用的画质词、光影词、负面词，写死在模板里。
- 品类层：按品类（杯具、服饰、数码）定义各自的风格和背景，一个品类一套。
- 单品层：只填主体、材质、颜色这些真正因商品而异的值。
这样三层下来，改动成本很低。要调整整体画质，改基础层；要新增一个品类，加一套品类层；单品层就是一张表格，运营填就行。
一个简化的模板骨架长这样（伪代码，实际用字符串模板即可）：
BASE = "商业产品摄影，{subject}，{material}材质，{color}，{style}风格，{background}，柔和顶光，无硬阴影，高清细节，8k" NEGATIVE = "文字，水印，多余物体，变形，模糊，低质量" def build_prompt(item): return BASE.format( subject=item["subject"], material=item["material"], color=item["color"], style=item["style"], background=item["background"], )注意负面词单独抽出来，因为不同平台对负面词的写法不一样，有的用negative_prompt字段，有的要拼在提示词里加权重。抽出来之后适配成本低很多。
2.3 变量表怎么维护才不出错
模板化之后，真正的"数据源"就变成了那张变量表。这张表我建议用 CSV 或表格维护，一行一个 SKU，列就是各个变量。这里有几个实操心得：
- 变量值要收敛。颜色别写"白色""纯白""奶白"三种，统一成一种，否则风格又漂了。可以维护一个枚举字典，填表时只允许选。
- 必填项要校验。生成前先跑一遍校验，缺字段的直接拦下来，别等到调接口才报错。
- 留一列备注。有些商品有特殊要求，比如"杯子要带把手朝右"，这种没法模板化的，单独备注，人工处理。
提示：变量表是整条流水线的源头，源头脏了后面全脏。宁可花时间把表规范好，也别指望在生成环节补救。
3. 队列调度：让几百个任务稳稳跑完
3.1 为什么不能用 for 循环硬跑
前面说过 for 循环的问题，这里展开讲。假设你有 500 个 SKU，每个要生成 4 张图，就是 2000 次调用。用 for 循环串行跑，假设每次 10 秒，那就是 20000 秒，五个多小时，中间任何一次网络抖动都可能让整个脚本崩掉，而且崩了之后你不知道跑到哪了。
改成并发呢？直接开 2000 个线程，平台限流立刻教你做人。所以核心矛盾是：既要快，又不能把平台打爆，还要能断点续跑。这正是队列调度要解决的。
3.2 队列的三层结构
我用的队列结构分三层：
- 任务队列：每个"SKU + 角度"是一个任务，入队时带上所有元数据（SKU、角度、提示词、目标文件名）。
- 工作池：固定数量的 worker 从队列取任务执行，worker 数量就是并发上限。
- 结果队列：worker 执行完把结果（成功/失败、文件路径、耗时）放回结果队列，由单独的消费者落库或写日志。
这样设计的好处是，并发数由 worker 数量控制，想调就调；任务和结果分离，方便统计和重试。
一个基于 Python 队列的简化实现：
import queue import threading task_q = queue.Queue() result_q = queue.Queue() def worker(worker_id): while True: task = task_q.get() if task is None: break try: path = generate_image(task) # 调接口生成 result_q.put({"task": task, "status": "ok", "path": path}) except Exception as e: result_q.put({"task": task, "status": "fail", "error": str(e)}) finally: task_q.task_done() # 启动固定数量 worker workers = [threading.Thread(target=worker, args=(i,)) for i in range(4)] for w in workers: w.start()worker 数量我一般设 3 到 5，具体看平台限流策略。宁可慢一点稳一点，也别追求极限并发。
3.3 重试与退避策略
失败是常态，关键是怎么重试。我的策略是有限重试 + 指数退避：
- 每个任务最多重试 3 次。
- 第一次失败等 2 秒，第二次等 4 秒，第三次等 8 秒。
- 三次都失败，标记为"永久失败"，进人工处理队列。
为什么要退避？因为很多失败是瞬时的（网络抖动、平台临时限流），立刻重试大概率还是失败，等一会儿反而能过。指数退避比固定间隔更合理，因为它给平台留了恢复时间。
import time def generate_with_retry(task, max_retry=3): for attempt in range(max_retry): try: return generate_image(task) except Exception as e: if attempt == max_retry - 1: raise time.sleep(2 ** (attempt + 1))3.4 断点续跑怎么实现
队列跑一半崩了，重跑时不能从头再来。实现断点续跑的关键是任务状态持久化。每完成一个任务，就把状态写进一个记录文件或数据库。重跑时先读状态，已经成功的任务直接跳过。
最简单的做法是用一个 JSON 文件记录已完成的任务 ID：
import json import os STATE_FILE = "done_tasks.json" def load_done(): if os.path.exists(STATE_FILE): with open(STATE_FILE) as f: return set(json.load(f)) return set() def mark_done(task_id, done_set): done_set.add(task_id) with open(STATE_FILE, "w") as f: json.dump(list(done_set), f)任务 ID 用"SKU + 角度"拼出来，天然唯一。这样即使中途崩了，重跑时读一下状态文件，已经完成的直接跳过，只跑剩下的。
注意：状态文件要定期备份，别让它成为单点。生产环境建议直接上 SQLite 或轻量数据库，比 JSON 文件可靠。
4. 幂等命名：让每张图都能被追溯
4.1 命名混乱的代价
前面提过默认命名的坑，这里说具体代价。假设你生成完 500 张图，文件名是output_1.png到output_500.png。现在运营说"把 A 商品的主图换一下"，你得先找到 A 商品对应哪几张图，然后一张张点开确认哪张是主图。500 张图，找一次十分钟，一天找十次就是一百分钟。这还只是找图，如果生成过程中断重跑，产生了重复文件，你还得去重。
命名混乱的代价，本质是把机器该干的活推给了人。幂等命名就是把这部分活还给机器。
4.2 幂等命名的设计原则
幂等命名的核心是：同样的输入，永远产生同样的文件名。这样重跑时，同名文件直接覆盖或跳过，不会产生重复。
命名结构我建议包含这几个部分：
{sku}_{angle}_{prompt_hash}.pngsku：商品编号，唯一标识。angle：角度或用途，如 main、scene、detail。prompt_hash：提示词的哈希值，前 8 位即可。
为什么要加 prompt_hash？因为同一个 SKU 的同一个角度，提示词可能改过。加了 hash，改提示词后生成的是新文件，不会覆盖旧图，方便对比。如果提示词没变，hash 一样，文件名一样，天然幂等。
import hashlib def make_filename(sku, angle, prompt): h = hashlib.md5(prompt.encode()).hexdigest()[:8] return f"{sku}_{angle}_{h}.png"4.3 幂等和断点续跑的配合
幂等命名和断点续跑是绝配。断点续跑靠状态文件判断"任务是否完成"，但状态文件可能丢或损坏。有了幂等命名，即使状态文件没了，重跑时发现目标文件已存在，也可以直接跳过，相当于多了一层保险。
判断逻辑很简单：
import os def should_skip(sku, angle, prompt, output_dir): filename = make_filename(sku, angle, prompt) return os.path.exists(os.path.join(output_dir, filename))这样即使状态文件丢了，重跑也不会重复生成，最多是重新校验一遍文件是否存在，成本极低。
4.4 元数据落库，方便后续检索
光有文件名还不够，最好把每个任务的元数据也落库，包括 SKU、角度、提示词、生成时间、文件路径、状态。这样后续要检索"某个 SKU 的所有图"或者"某个提示词生成了哪些图"，一条 SQL 就搞定。
我用的是 SQLite，轻量够用：
import sqlite3 conn = sqlite3.connect("images.db") conn.execute(""" CREATE TABLE IF NOT EXISTS images ( id INTEGER PRIMARY KEY, sku TEXT, angle TEXT, prompt TEXT, filename TEXT, status TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """)有了这张表，运营要查图、要统计、要重生成，都有据可依。
5. 三块拼起来：一次完整的批量生成流程
5.1 从变量表到成图的完整链路
把三块拼起来，完整流程是这样的：
- 读变量表：加载 CSV，校验必填字段。
- 构建任务：每个 SKU 按角度展开成多个任务，每个任务生成提示词和幂等文件名。
- 过滤已完成：读状态文件 + 检查文件是否存在，跳过已完成的。
- 入队：剩余任务入任务队列。
- worker 消费：固定并发数执行，带重试和退避。
- 结果落库：成功写元数据，失败进重试或人工队列。
- 收尾：输出统计报告，成功多少、失败多少、耗时多少。
这条链路跑通之后，一次几百个 SKU 的批量生成，基本可以无人值守。
5.2 关键参数怎么定
几个关键参数的取值经验：
| 参数 | 建议值 | 说明 |
|---|---|---|
| worker 数量 | 3-5 | 看平台限流，宁少勿多 |
| 单任务超时 | 60-120 秒 | 生成类接口普遍较慢 |
| 最大重试次数 | 3 | 再多收益递减 |
| 退避基数 | 2 秒 | 指数退避起点 |
| 状态落盘频率 | 每任务完成即写 | 别攒批，崩了会丢 |
这些值不是拍脑袋，是踩坑踩出来的。比如 worker 数量，我一开始设 10，结果平台限流，一半任务失败，改成 4 之后稳定多了。
5.3 实测中的意外情况
跑通之后还是会遇到意外，分享几个我遇到的：
- 提示词超长被截断：有些平台对提示词长度有限制，超了会静默截断，出来的图跟预期不符。解决办法是生成前校验长度，超长的拆分或精简。
- 同名文件被覆盖：如果 prompt_hash 算错（比如用了随机数），同名文件会互相覆盖。一定要用确定性哈希。
- 状态文件并发写冲突：多个 worker 同时写状态文件会冲突。解决办法是加锁，或者干脆每个 worker 写自己的状态文件，最后合并。
- 平台返回成功但图是坏的：有些接口返回 200 但图是空白或损坏。解决办法是生成后做一次校验，比如检查文件大小和尺寸。
提示：批量生成最怕"静默失败"，即看起来成功实际有问题。一定要加校验环节，宁可多花几秒，也别让坏图混进去。
6. 几个容易被忽略的实操细节
6.1 提示词版本管理
提示词改来改去，很容易忘了哪版效果好。我的做法是给提示词模板加版本号，每次改动记录变更原因。配合 prompt_hash，能清楚知道每张图用的是哪版提示词。这样对比效果时，不会张冠李戴。
6.2 失败任务的分类处理
失败不是一种，要分类：
- 瞬时失败（网络、限流）：重试能解决。
- 参数失败（提示词违规、字段缺失）：重试没用，要改参数。
- 平台失败（接口变更、服务异常）：要人工介入。
分类之后，重试策略才能对症下药，而不是无脑重试。
6.3 成本控制
批量生成是要花钱的，尤其是按次计费的平台。几个省钱技巧：
- 先小批量试跑：新模板先跑 5 个 SKU，确认效果再全量。
- 复用已有图：提示词没变的，直接跳过，别重复生成。
- 失败快速止损：连续失败超过阈值，暂停队列，别继续烧钱。
6.4 日志要打全
日志是排查问题的命根子。每个任务至少记录：任务 ID、提示词、开始时间、结束时间、状态、错误信息。日志格式建议结构化（JSON），方便后续分析。别用 print 打日志，量大了根本看不过来。
7. 写在最后的一点个人体会
这套流程我从最初的手工出图，到脚本硬跑，再到现在的模板化 + 队列 + 幂等命名，前后迭代了好几轮。最大的体会是：批量生成的核心不是"生成"，而是"管理"。生成本身调个接口就完事，真正花时间的是怎么保证一致性、怎么控制并发、怎么追溯和重跑。
模板化让风格统一，队列调度让过程可控，幂等命名让结果可追溯。三块缺一不可。如果你现在还在手工出图，建议先从模板化入手，把变量表规范起来，这一步收益最直接。等量上来了，再补队列和幂等命名。
最后分享一个小技巧：变量表里可以加一列"优先级"，紧急的 SKU 标 high，队列调度时优先处理。这样运营临时要图，不用等整批跑完，插个队就行。这个改动很小，但实际用起来体验提升很明显。
