做批量生图这个事我是从给一批电商商品做展示图开始入坑的。当时手里有一百多个SKU每个都要出三四张不同角度的图按常规思路“调个API、复制粘贴跑一下”单张调用确实没问题但一旦到了几十上百张就发现流程完全不是那么回事要么跑一半报错中断、要么API限流被卡住、要么生成的图片格式不统一还要一张张手动处理。后来我把DALL·E和Gemini都接进了Python脚本做成了一条从prompt清单到成品图片自动落盘的完整流水线整个过程才算真正跑顺。这篇文章就是把整套方案从头到尾梳理一遍包括选型原因、代码实现、踩坑记录和常见问题排查。适合有Python基础、准备把AI生图从“偶尔玩一两张”升级到“批量生产”的开发者参考也适合那些正在搭建内容素材流水线的团队拿来直接用。1. 批量生图的整体设计与思路拆解1.1 从单张到批量你到底需要解决哪些问题很多人以为批量生图就是把单张调用的代码套一层for循环实测下来根本不是这么简单。单张调用时你只需要关心“这1张图能不能成功生成”但批量场景下的核心问题是如何在有限的时间和预算内可靠地生成大量图片并且在失败时不至于前功尽弃。我拆解了一下批量生图至少要解决五个层面的问题并发调度图片生成接口的响应时间通常是5到20秒串行跑100张图意味着要等半小时甚至更久。必须引入并发但又不能无脑并发因为API有频率限制。失败恢复批量任务跑半小时如果中途某一张图因为内容审核或临时网络问题报错整个流程就中断这是一种很糟糕的体验。需要设计重试机制并且让失败任务不阻塞后续任务。成本控制生图接口按张数或按token计费批量跑之前最好先算出预算不然月底账单能吓你一跳。结果管理生成的图片需要按规则命名、按目录归档最好还能生成一份包含prompt、模型、出图参数、状态的结果清单方便溯源。断点续跑如果跑到一半进程崩了已生成的图不应该重复生成得能跳过已完成项接着跑。这些需求其实就是一条流水线的组成部分。我自己搭的方案是把整条链路分成四层输入层用CSV或JSON维护prompt清单调度层负责并发控制和重试执行层封装不同模型的调用逻辑输出层统一做图片保存、命名和日志记录。每一层之间只通过任务字典传递数据这样很方便做单元测试也方便以后加入更多模型。1.2 为什么是Python DALL·E / Gemini 双模型并行选Python不用多说做脚本、做数据处理、做自动化Python的生态最顺手。关键是“为什么同时接DALL·E和Gemini而不是只用一个”。最开始我也只接了DALL·E但用了一段时间后发现两个问题一是DALL·E的输出风格比较“固定”所有图都带着同一种审美倾向做内容矩阵的时候图片风格太单一二是某些prompt在DALL·E上的内容审核比较严格偶尔会直接拒绝生成但同一个prompt放到Gemini上能正常出图。后来又试了Gemini发现它的图片生成模型在文字渲染和画面层次感上也有自己的特点。把两个模型都接入其实是在给自己多一个选择余地。从工程角度看双模型并行还有几个实际好处互备某一个模型服务出现故障或限流严格时另一个模型可以顶上流水线不至于停摆。成本优化两个模型的价格和计费方式不一样低预算任务可以优先走便宜的那个。风格差异化不同内容账号或不同栏目可以固定用不同模型形成视觉风格上的区分。当然双模型接入也意味着代码层要多做一层适配。同一张prompt、同样的尺寸参数两个模型的请求体结构、返回格式完全不同所以我在执行层做了一层薄薄的封装把模型差异隔离在内部调度层完全不关心底层是哪个模型。2. 环境准备与最早的“单张调用”2.1 环境搭建和API接入这一步比较简单但也是很多人容易出岔子的地方。我建议先建一个新的虚拟环境再安装两个官方SDK。python -m venv .venv source .venv/bin/activate pip install openai google-generativeai httpx python-dotenvAPI Key的管理是一个我一直强调的点永远不要硬编码在代码里。写死在脚本里的Key一旦被提交到代码仓库基本等于公开了。我把Key放在项目的.env文件里用python-dotenv读取from dotenv import load_dotenv import os load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY) GEMINI_API_KEY os.getenv(GEMINI_API_KEY)然后分别初始化客户端。两个模型接口风格差异比较大一边是OpenAI风格一边是Google风格建议分模块封装from openai import OpenAI openai_client OpenAI(api_keyOPENAI_API_KEY)from google import genai gemini_client genai.Client(api_keyGEMINI_API_KEY)有两点值得提一下。新版OpenAI SDK的openai包已经集成了图片生成能力不需要额外安装独立的images库。Gemini这边图片生成推荐用的模型是gemini-2.0-flash-exp-image-generation或者通过Imagen API调用imagen-3.0-generate-002目前是通过官方SDK访问的。2.2 跑通第一张图的完整代码先别急着写流水线第一步一定是验证“单张能不能跑通”。这一步排查的是API Key、网络连通性、参数格式这些基础问题如果连单张都跑不通后面所有工程化设计都是空中楼阁。我按DALL·E先来因为OpenAI SDK的返回结构比较直观response openai_client.images.generate( modelgpt-image-1, prompt一只橘猫坐在窗台上背景是黄昏的城市天际线摄影风格高细节, size1024x1024, )关键点在于结果提取。gpt-image-1这个模型的返回对象是data[0]这个对象里既可能直接带有b64_json字段Base64编码的图片二进制也可能只有url字段。实测下来部分模型默认返回Base64部分返回可下载的临时URL。处理方式就是两者都判断import base64 import httpx image_data response.data[0] if image_data.b64_json: img_bytes base64.b64decode(image_data.b64_json) elif image_data.url: img_bytes httpx.get(image_data.url).content else: raise ValueError(无法从响应中提取图片数据) with open(first_image.png, wb) as f: f.write(img_bytes)Gemini侧的调用代码风格完全不同没有client.images.generate这种“统一入口”而是通过models.generate_content传图片生成模型gemini_response gemini_client.models.generate_content( modelgemini-2.0-flash-exp-image-generation, contents一只橘猫坐在窗台上背景是黄昏的城市天际线摄影风格高细节, ) image_bytes gemini_response.candidates[0].content.parts[0].inline_data.data with open(first_gemini_image.png, wb) as f: f.write(image_bytes)Gemini的返回格式较复杂图片内容通常在candidates[0].content.parts[0].inline_data.data这个路径下。如果返回的是纯文本而非图片说明模型没有正确触发图片生成模式。跑通这两段代码后我建议立刻做一个验证多试几种尺寸、几种风格的prompt确认账号权限和模型可用性。这一步能帮你提前发现很多问题比如某些账号对特定尺寸不支持或者某些关键词会触发内容审核这类问题越早暴露越好。3. 从单张到批量流水线的关键演进3.1 同步批处理的简单实现与隐藏问题单张调用没问题后最自然的想法就是循环调用。比如先准备一批prompt然后挨个跑prompts [prompt 1, prompt 2, prompt 3, ...] for i, prompt in enumerate(prompts): response openai_client.images.generate( modelgpt-image-1, promptprompt, size1024x1024, ) # 保存图片...这段代码能跑但它有几个非常现实的毛病。第一慢。每张图请求发出后要等模型生成完毕按平均8秒算100张图就是800多秒还不包括下载图片的时间和网络延迟。第二一挂全挂。如果第50张图触发了内容审核抛异常整个脚本就中断前面49张白跑了。第三没有恢复能力。即使你手动改了代码跳过失败项这些失败的任务也没有记录下次还得人工排查哪几张没生成。所以同步循环只适合演示不适合真实批产。做批量流水线必须从“循环调用”升级到“任务调度”。3.2 并发批处理用线程池提升吞吐提升吞吐最直接的方式是并发。Python里做并发推荐concurrent.futures.ThreadPoolExecutor因为图片生成这类任务是IO密集型——大部分时间都在等API响应线程池足够胜任而且写起来比asyncio直观很多。from concurrent.futures import ThreadPoolExecutor, as_completed import json tasks [ {id: i, prompt: prompt, model: dalle, size: 1024x1024} for i, prompt in enumerate(prompts) ] results [] def process_task(task): try: img_bytes generate_image(task) return {task_id: task[id], status: success, data: img_bytes} except Exception as e: return {task_id: task[id], status: failed, error: str(e)} with ThreadPoolExecutor(max_workers5) as executor: futures [executor.submit(process_task, task) for task in tasks] for future in as_completed(futures): result future.result() results.append(result)这里给ThreadPoolExecutor的max_workers参数设了一个经验值5原因是大多数生图API账号的并发限制通常在每分钟20到40次请求左右5到8个并发线程是比较安全的起始值。但这一步只解决了“快”的问题还没解决“稳”的问题。线程池里的某个任务失败了虽然不会影响其他任务的执行但失败的任务怎么重试怎么记录仍需要额外设计。3.3 任务队列真正的流水线架构我把“真正的批量流水线”定义为三个能力任务可描述、进度可追踪、失败可重试。三者缺一不可。任务可描述是指每个任务里不只有prompt还应该包含模型选择、尺寸、输出路径、命名规则等完整信息。我习惯用一个字典来承载task { id: sku_1024_main, prompt: 商品主体居中的电商白底图左侧留白放文案, model: gemini, size: 1024x1024, output_path: ./output/sku_1024/main.png, retry_count: 0, max_retries: 3, }进度可追踪是指每个任务的状态能够在运行过程中被实时查看。我在调度层维护一个状态字典每个任务在“排队、运行中、成功、失败”之间流转最终汇总成一份results.json。失败可重试是最关键的一环。单个任务在执行中可能遇到网络超时、临时限流、服务端错误等可以自动恢复的问题这类情况应该自动重试。我封装了一个带重试的调度循环import time def run_with_retry(task, max_retries3, backoff_base2): for attempt in range(max_retries): try: return generate_image(task) except RateLimitError: wait_time backoff_base ** attempt print(f任务 {task[id]} 触发限流等待 {wait_time}s 后重试...) time.sleep(wait_time) except APIConnectionError: wait_time backoff_base ** attempt print(f任务 {task[id]} 网络连接失败等待 {wait_time}s 后重试...) time.sleep(wait_time) raise RuntimeError(f任务 {task[id]} 重试 {max_retries} 次后仍失败)这样100张图的任务即使有十几张中途报错只要不是永久性错误都能在自动重试后完成最后只有极少数真正失败的任务需要人工介入。4. 成本控制、限流与工程化加固4.1 限流与重试策略优雅地面对429跑批量任务遇到最多的就是HTTP 429请求过多。不少人的第一反应是“并发调低一点”其实这只是治标。做完一次大规模跑批后我的体会是限流处理同时做三件事第一合理控制并发。不要一上来就开10个线程建议根据API文档给的速率限制反推并发数。比如文档写“每分钟40次”那每秒最多0.67次保守起见max_workers设为5单张平均约8秒的话5并发折算下来每分钟约37次请求刚好卡在阈值内。第二对429做指数退避重试。指数退避的核心是“失败后等待时间随尝试次数翻倍”避免所有线程在同一时刻同时重试、形成“惊群效应”。第三区分错误类型。429是限流可以重试401是Key错误重试一万次也没用400如果是参数错误需要修改代码而不是重试。所以我在异常处理时会对错误类型做个分支判断直接决定这个任务是重试还是标记为永久失败。4.2 结构化输出与缓存机制让流水线可断点续跑批量任务最怕的是“跑了一半进程崩了从头再来”。针对这个问题我设计了一个简单的断点续跑机制核心思想就是“先查缓存再决定是否生成”。每成功生成一张图片我就把这条任务标记为已完成写入一个completed.json文件。下次启动脚本时先加载这个文件凡是已经存在的任务ID直接跳过def load_completed(pathcompleted.json): if not os.path.exists(path): return set() with open(path, r, encodingutf-8) as f: return set(json.load(f)) completed load_completed() for task in all_tasks: if task[id] in completed: print(f跳过已完成任务: {task[id]}) continue run_with_retry(task) # 成功后把任务ID写入 completed 集合这个机制在批量任务里价值很高。假设你有300张图要生成跑到第270张时断网或断电重启脚本后前270张一键跳过成本和时间都能省下来。除了断点续跑命名规范也要提前设计。我曾经吃过“图片文件乱命名”的亏生成完找不到哪张图对应哪个prompt。后来统一用“任务ID_模型_尺寸”的规则来命名例如sku_1024_main_dalle_1024x1024.png并在结果清单里记录原始prompt和生成时间。4.3 预算估算与成本控制方案图片生成并不是免费的批量之前先算好成本是个好习惯。不同模型计费方式不同有的按张数计费、有的按输入输出token综合计费价格也会随尺寸和画质变化。以我当时常用的配置为例一张中等尺寸的图单张成本大约在3分钱到1毛钱人民币之间浮动具体看模型和参数。如果你做的是小规模测试这个成本可能不太看得出来但一旦上百张、上千张费用就显著了。我给自己定了一个流程先拿5到10张图做小样测试确认prompt质量和风格满意根据小样测试估算总成本超出预算就精简prompt数量或用低价模型跑一部分设置每日上限用脚本控制每天最多只跑多少张避免半夜挂机跑过头定期导出账单按项目、按模型、按任务维度核算成本。还有一个省钱技巧是“结果复用”。有些图只是微调了尺寸或构图如果模型API支持基于已有图片做变体就比单独从这个prompt重新生成要省不少。这个功能各家模型叫法不同具体看API文档里是否提供edit或variation接口。5. 常见问题与调试实录5.1 高频报错速查表批量生图跑到现在我把遇到过的高频错误整理成了一张表遇到问题照着排查能省很多时间错误类型典型现象可能原因解决方案HTTP 401请求未授权API Key错误或过期检查环境变量再重新生成KeyHTTP 400参数不合法尺寸值不支持、提示词为空、内容审核拒绝核对文档支持的参数范围调整prompt措辞HTTP 404模型不存在模型名拼写错误或账号无权限确认已开通对应模型权限HTTP 429请求频率超限并发数过高或触发分钟级限制降低并发、等待一分钟、加指数退避HTTP 500服务端内部错误API服务暂时不稳定按退避策略重试多次失败就大睡一觉再跑超时异常请求长时间不返回网络问题或服务端处理慢设置超时参数失败自动重试返回内容为空响应里没有图片数据触发内容审核、模型输出了文本检查响应全文调整prompt或换模型5.2 我踩过的几个坑以下这些坑每一行都是真实经历按“受害程度”排序分享出来。第一个坑是并发数开太高导致被限流一整天。刚开始做批量时我把线程数调到20跑了几分钟就撞上429然后整批任务都开始反复重试。后来我一查API文档才发现每分钟请求上限远低于我的设置。现在我的做法是先跑10张做“探针测试”统计出单请求的平均耗时再按并发数 每分钟配额 / (60 / 平均耗时)这个公式反推合理线程数。平均耗时8秒时每分钟大约能跑7.5张按这个结果设置并发数就非常稳。第二个坑是不同模型的返回结构字段名不一样写代码时不能想当然。DALL·E返回的图片信息在data[0].b64_json或data[0].urlGemini则藏在candidates[0].content.parts[0].inline_data.data这俩结构差异很大所以我把两种模型的解析逻辑分别封装成独立的函数并在单元测试里做了Mock验证避免解析时报KeyError。第三个坑是内容审核触发时服务端返回的还是200一不注意会把错误响应当成正常数据。有一次批量跑了几十张图代码没有报错但实际生成的是一段“该内容违反了政策”的文本说明。排查了很久才发现接口返回200并不等于成功。现在的做法是在解析响应时校验关键字段是否存在并且把响应原文的status、safety_ratings等信息记录到日志里便于事后定位。第四个坑是图片下载不设超时导致任务卡死。从url字段下载图片时如果网络异常httpx.get()可能长时间挂起。后来所有网络请求我都设置了超时img_bytes httpx.get(image_data.url, timeout10).content第五个坑是中文prompt在部分模型上出现乱码或理解偏移。同样的中文描述在两个模型上的生成效果差异不小Gemini的中文理解能力相对强一些DALL·E上某些中文长句容易被“过度解读”。我的应对策略是在prompt里同时给出一段英文关键词作辅助引导生成效果稳定很多。6. 批量流水线的进阶扩展方向如果上面的方案你已经跑通了日常的出图效率至少翻了几倍。但批量生图这件事本质上是可以向“自动化内容工厂”方向延伸的。我目前比较推荐的下一个方向是把prompt清单的管理从代码里抽出来放到一个简单的配置文件或表格里。非技术人员只需要维护表格里的prompt文本、模型偏好、尺寸流水线本身完全不用动。这样一来你写的就不再是一个“我给你100个prompt你帮我跑100张图”的一次性脚本而是一个可以反复使用、随时增减任务的生产工具。再往下走可以把流水线加上一个简单的Web界面或者接入任务队列服务让批量任务能定时触发、定时出结果。做到这一步批量生图就真正从“脚本工具”变成了“生产系统”。当然“生产系统”比脚本多出来的复杂度主要是运维层面的任务状态要可视化、失败告警要通知到你、历史记录要可查询。这些每一样都需要额外开发但每一步都能实打实地省下人力。先从小处着手单张跑通、并发提速、重试兜底、缓存续跑一步步做扎实后面的扩展才不会摇摇欲坠。我自己做完这套流水线后最大的感受是批量生图的核心难点不在“调接口”而在“工程化”。把任务管理、异常处理、成本控制这三件事想清楚整个流程才真正靠谱。希望这份踩坑总结能帮你少走一些弯路。