AI 工作流实战 - 调用豆包api实现批量生图

一、前言

        豆包大模型（DouBao-SeedDream-4.0）作为字节跳动推出的先进生成式人工智能模型，在视觉内容生成领域展现出卓越的性能。该模型凭借出色的多模态理解能力与创造性输出，能够基于文本描述生成高质量、高精度的视觉内容。然而，其核心价值不仅体现在单次生成任务中，更在于通过API接口实现的批量化生成能力——这一功能将人工智能从辅助工具升级为规模化内容生产的核心驱动力。

二、如何调用api

        打开 火山引擎，选择：产品-精选产品-豆包大模型，即可。你也可以选择其他大语言模型，模型的选用时根据模型的具体价格和你对性能的需求选定的。

        这里我示范的是豆包图像创作模型4.0：doubao-seedream，如果是需要生成视频的，我们需要选择现有的视频模型。

        进入火山方舟的控制台界面，选择右上角的API接入。

        如果你还没有创建过api，则选择创建。如果已经创建好了api，则点击API Key的小眼睛查看密钥（千万不能泄露自己的api，切记！！！）

        创建API后可以查看STEP2的快速接入测试。如果是要批量生图，则在左上角下拉栏选用“文生图-生成-组图” （需要根据你选定的模型跟换你的模型型号和接口地址，api是通用的。）

        这里的 doubao-seedream-4-0-250828 就是模型型号，https://ark.cn-beijing.volces.com/api/v3 是

接口地址。

三、批量生图

        我们需要将代码修改后放入.py文件后在编译器运行（这里的示例编译器为pycharm）。按照模型的原始代码，会给你生成四张图片，你可以通过运行后给出的网址链接进行下载。如果我们想指定下载位置，可以将代码修改为下面这样：（其中./generated_images为你的.py文件所在文件夹的下一级创建的存放图片的文件夹。）

你可能需要修改的内容为：API_KEY，url， model 和 prompt。

import requests
import json
import osAPI_KEY = "换成你自己的api密钥"
url = "https://ark.cn-beijing.volces.com/api/v3/images/generations"payload = {"model": "doubao-seedream-4-0-250828","prompt": "生成一组共4张连贯插画，核心为同一庭院一角的四季变迁，以统一风格展现四季独特色彩、元素与氛围","sequential_image_generation": "auto","sequential_image_generation_options": {"max_images": 4},"response_format": "url","size": "2K","stream": False,"watermark": True
}headers = {"Content-Type": "application/json","Authorization": f"Bearer {API_KEY}"
}# 发请求
resp = requests.post(url, headers=headers, data=json.dumps(payload))
print("Status:", resp.status_code)# 解析响应
jd = resp.json()# 创建保存文件夹
save_dir = "./generated_images"
os.makedirs(save_dir, exist_ok=True)# 下载图片
for i, item in enumerate(jd.get("data", [])):img_url = item.get("url")if not img_url:continueimg_bytes = requests.get(img_url).contentsave_path = os.path.join(save_dir, f"image_{i+1}.jpg")with open(save_path, "wb") as f:f.write(img_bytes)print(f"✅ 已保存: {save_path}")print("所有图片已生成。")

生成效果如下：

如果我们想单次指定多个prompt，并且指定每个prompt生成的图片数，可以使用下面的代码：

（你需要修改的地方为：API_KEY，url， model， prompt 和 IMAGES_PER_PROMPT）

import os
import re
import json
import time
import base64
import math
import queue
import shutil
import random
import string
import pathlib
import datetime as dt
from typing import List, Dict, Any, Tuple, Optional
import requests
from concurrent.futures import ThreadPoolExecutor, as_completedAPI_KEY = "换成你自己的api密钥" 
MODEL_ID = "doubao-seedream-4-0-250828"  # 控制台显示的模型ID
BASE_URL = "https://ark.cn-beijing.volces.com/api/v3/images/generations"# 生成尺寸：可用 "1K"、"2K" 或 "1024x1024"/"2048x2048"
SIZE = "2K"# 多 prompt 并发数（建议先用 1~3，稳定后再加）
CONCURRENCY = 3# 速率限制（每次HTTP请求之间的最小间隔秒—针对单进程）
MIN_INTERVAL_SEC = 0.25# 最大重试次数
MAX_RETRY = 3# 超时时间
TIMEOUT_SEC = 180
# ======================================def now_ts() -> str:return dt.datetime.now().strftime("%Y%m%d_%H%M%S")def safe_slug(text: str, maxlen: int = 40) -> str:"""把 prompt 变成安全短文件名片段。"""s = re.sub(r"\s+", "_", text.strip())s = re.sub(r"[^\w\-_.]", "", s)if len(s) > maxlen:s = s[:maxlen]return s or "prompt"def ensure_dir(p: pathlib.Path):p.mkdir(parents=True, exist_ok=True)def normalize_size(sz: str) -> str:s = sz.replace("*", "x").lower()if s in ("1024x1024", "1k"):return "1K"if s in ("2048x2048", "2k"):return "2K"return sz def call_images_api_once(prompt: str, n: int, size: str, stream: bool = False) -> List[bytes]:"""调一次文生图。返回本次拿到的若干张图片（二进制）。Seedream 单次最多 4 张；如果外层要更多，请循环调用这个函数。"""headers = {"Content-Type": "application/json","Authorization": f"Bearer {API_KEY}",}payload = {"model": MODEL_ID,"prompt": prompt,"sequential_image_generation": "auto","sequential_image_generation_options": {"max_images": max(1, min(4, int(n)))},  # 单次最多 4"response_format": "url",  # 推荐 url，拿回再下载"size": normalize_size(size),"stream": bool(stream),    # 测试/批处理建议 False，便于解析"watermark": True}# 速率限制（简单的全局节流）time.sleep(MIN_INTERVAL_SEC)resp = requests.post(BASE_URL, headers=headers, data=json.dumps(payload), timeout=TIMEOUT_SEC)if resp.status_code != 200:raise RuntimeError(f"HTTP {resp.status_code}: {resp.text}")jd = resp.json()images: List[bytes] = []# 优先解析 data[].url / data[].b64_jsonif isinstance(jd, dict) and isinstance(jd.get("data"), list):for item in jd["data"]:if not isinstance(item, dict):continueif "b64_json" in item and item["b64_json"]:try:images.append(base64.b64decode(item["b64_json"]))except Exception:passelif "url" in item and item["url"]:try:img_bytes = requests.get(item["url"], timeout=TIMEOUT_SEC).contentimages.append(img_bytes)except Exception as e:print("[warn] fetch url error:", e)# 兼容 result.images / result.image_listif not images and isinstance(jd.get("result"), dict):arr = jd["result"].get("images") or jd["result"].get("image_list")if isinstance(arr, list):for b in arr:if isinstance(b, str):try:images.append(base64.b64decode(b))except Exception:passif not images and isinstance(jd, dict) and jd.get("error"):raise RuntimeError(f"API error: {jd['error']}")return imagesdef generate_for_prompt(prompt: str, total_images: int, out_dir: pathlib.Path) -> Tuple[str, List[pathlib.Path]]:saved_paths: List[pathlib.Path] = []remaining = int(total_images)call_index = 0while remaining > 0:batch_n = min(4, remaining)call_index += 1last_err = Nonefor attempt in range(1, MAX_RETRY + 1):try:imgs = call_images_api_once(prompt, batch_n, SIZE, stream=False)if not imgs:raise RuntimeError("no image returned in this call")for img in imgs[:batch_n]:fname = f"image_{len(saved_paths)+1:03d}.jpg"fpath = out_dir / fnamefpath.write_bytes(img)saved_paths.append(fpath)break  except Exception as e:last_err = esleep_s = 0.6 * attemptprint(f"[retry] prompt='{prompt[:18]}...' call#{call_index} attempt={attempt}/{MAX_RETRY} err={e} -> sleep {sleep_s:.1f}s")time.sleep(sleep_s)if last_err and len(saved_paths) == 0 and call_index == 1:print(f"[error] prompt='{prompt[:40]}...' failed: {last_err}")breakremaining = total_images - len(saved_paths)return prompt, saved_pathsdef main():if not API_KEY or API_KEY.startswith("YOUR_"):raise SystemExit("请先把 API_KEY 填写为你的 Ark API Key 再运行。")# 1) 在这里维护你的多段 prompt。每一行一个，请将prompt替换为你自己的。prompts: List[str] = ["prompt1","prompt2","prompt3"]IMAGES_PER_PROMPT = 3  # 控制每个 prompt 要生成几张# 你也可从文件读取：每行一个 prompt# with open("prompts.txt", "r", encoding="utf-8") as f:#     prompts = [line.strip() for line in f if line.strip()]if not prompts:print("没有可用的 prompt。")return# 2) 输出根目录root = pathlib.Path("./generated") / now_ts()ensure_dir(root)print(f"总计 {len(prompts)} 个 prompt；每个生成 {IMAGES_PER_PROMPT} 张；并发={CONCURRENCY}；输出根目录：{root.resolve()}")# 3) 并发跑每个 promptfutures = []with ThreadPoolExecutor(max_workers=max(1, CONCURRENCY)) as ex:for i, p in enumerate(prompts, 1):subdir = root / f"{i:03d}_{safe_slug(p)}"ensure_dir(subdir)futures.append(ex.submit(generate_for_prompt, p, IMAGES_PER_PROMPT, subdir))total_saved = 0for fu in as_completed(futures):prompt, paths = fu.result()print(f"[done] '{prompt[:32]}...' -> {len(paths)} 张，目录：{paths[0].parent if paths else '无'}")total_saved += len(paths)print(f"✅ 全部完成！共保存 {total_saved} 张图片。输出目录：{root.resolve()}")if __name__ == "__main__":main()

        其中你有多少行prompt就代表有几条任务（注意要一行一条），其中 IMAGES_PER_PROMPT=3代表每条任务重复三次（生成三张图）。【注：如果你要快速找到某行代码的位置，可以在编译器中使用快捷键 Ctrl+f 进行查找。】

        我给出三条示例prompt（电商场景下生成手机主图、场景图和细节图 -- 虚构ai提示词，仅供参考）：

主图：

一款未来风格的智能手机（iPhone 17）的官方产品渲染图，纯白色无缝背景。产品占据画面约70%，呈精确的3/4视角，完美展示其全新的钛合金直角边框与背部哑光AG玻璃质感。摄像头模组为创新的一体式水晶设计，略微凸起，内部镜头结构清晰可见。屏幕极大且边框极窄，屏幕处于点亮状态，清晰显示着iOS系统的默认主屏幕界面，包括应用图标、时间小组件以及顶部的信号、电量状态栏，所有UI元素都锐利无比，以展现其超视网膜XDR屏幕的极致分辨与色彩。光线为专业的摄影棚布光，正面柔和均匀，在钛合金边框上形成一道纤细的连续高光，凸显其精密做工。无任何阴影、倒影或多余元素，纯粹聚焦于产品本身。超高分辨率，8K画质，商业级摄影，焦点锐利，细节必须分毫毕现。--ar 1:1 --style raw

场景图：

iPhone 17 深海蓝标准版放在机场休息室原木桌台，机身侧面 Type-C 接口连接移动硬盘（传输 4K 视频图标亮屏显示），旁边有 AirPods Pro 3（正通过手机反向充电）、护照和咖啡杯，清晨自然光从舷窗斜射，机身反射桌面纹理，暖色调氛围，像素1024×768

细节图：

动态的3D渲染图，特写展示一部深空黑色的iPhone 17的侧面角度，突出其极其纤薄又坚固的机身轮廓。背景是柔和的浅灰色渐变。极简主义的信息图元素优雅地悬浮在手机周围的留白区域：一个水滴图标及文字“Ultra WaterGuard: IP68”，一个闪电穿透电池的图标及文字“A19 Pro: 能效提升40%”，以及一个箭头指向手机边框及文字“航天级钛合金”。整体风格是干净、明亮、科技感十足的信息图表。构图带有倾角以营造动感。所有文字和图标都清晰可读，与产品专业地融为一体。整体观感具有说服力，高科技，并能瞬间传递核心信息。--ar 1:1

运行效果如下：

生图效果如下：

一些细节：