豆包API批量图生图完整指南

目录

1. 准备工作
2. API获取与配置
3. 批量图生图方法
4. 代码实现示例
5. 高级参数配置
6. 最佳实践与注意事项
7. 常见问题解决

准备工作

1.1 注册火山引擎账号

豆包API通过火山引擎平台提供服务，需要先注册账号：

1. 访问火山引擎官网：https://www.volcengine.com/
2. 使用抖音账号扫码登录或手机号注册
3. 完成实名认证（需要身份证和人脸识别）

1.2 开通相关服务

登录后需要开通豆包大模型服务：

1. 进入火山引擎控制台
2. 选择"产品" → “人工智能” → “豆包大模型”
3. 点击"开通服务"，同意相关协议
4. 开通视觉大模型服务（前200次生成免费）

API获取与配置

2.1 获取API Key

1. 在火山引擎控制台找到"API接入"
2. 点击"创建API Key"
3. 生成后点击小眼睛图标查看完整API Key
4. 重要：妥善保管API Key，不要泄露给他人

2.2 模型选择

目前支持的主要图像创作模型：

模型名称	版本	特点	适用场景
doubao-seedream-4-0-250828	4.0	支持4K多模态生图，推理速度快	商业设计、专业海报
seededit-3.0	3.0	基础图生图功能	简单图像编辑

2.3 API接口信息

• Base URL: https://ark.cn-beijing.volces.com/api/v3/images/generations
• 请求方式: POST
• 认证方式: Bearer Token
• 响应格式: 支持url或b64_json

批量图生图方法

3.1 方法一：单次请求批量生成

使用sequential_image_generation参数实现单次请求生成多张图片：

payload = {
    "model": "doubao-seedream-4-0-250828",
    "prompt": "生成一组共4张连贯插画，核心为同一庭院一角的四季变迁",
    "sequential_image_generation": "auto",
    "sequential_image_generation_options": {"max_images": 4},
    "size": "2K",
    "response_format": "url"
}

3.2 方法二：多prompt循环生成

对多个不同的prompt进行循环调用：

prompts = [
    "春天的樱花盛开场景",
    "夏天的海滩日落风景", 
    "秋天的红叶满山景色",
    "冬天的雪景小屋画面"
]

for prompt in prompts:
    generate_image(prompt)
    time.sleep(1)  # 避免请求过于频繁

3.3 方法三：并发批量生成

使用多线程实现并发请求，提高生成效率：

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=3) as executor:
    executor.map(generate_image, prompts)

代码实现示例

4.1 基础批量生成代码

import requests
import json
import os
import time

class DouBaoImageGenerator:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ark.cn-beijing.volces.com/api/v3/images/generations"
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }
    
    def generate_single_image(self, prompt, model="doubao-seedream-4-0-250828", 
                            size="2K", response_format="url", seed=-1):
        """生成单张图片"""
        payload = {
            "model": model,
            "prompt": prompt,
            "size": size,
            "response_format": response_format,
            "seed": seed
        }
        
        try:
            response = requests.post(
                self.base_url, 
                headers=self.headers, 
                data=json.dumps(payload),
                timeout=180
            )
            
            if response.status_code == 200:
                return response.json()
            else:
                print(f"请求失败: {response.status_code} - {response.text}")
                return None
                
        except Exception as e:
            print(f"生成图片时出错: {str(e)}")
            return None
    
    def generate_batch_images(self, prompts, output_dir="./generated_images", 
                             model="doubao-seedream-4-0-250828", size="2K", 
                             delay=1, max_retries=3):
        """批量生成图片"""
        # 创建输出目录
        os.makedirs(output_dir, exist_ok=True)
        
        results = []
        
        for i, prompt in enumerate(prompts):
            print(f"正在生成第 {i+1}/{len(prompts)} 张图片...")
            
            retries = 0
            while retries < max_retries:
                result = self.generate_single_image(prompt, model, size)
                
                if result and "data" in result:
                    # 下载并保存图片
                    for j, item in enumerate(result["data"]):
                        img_url = item.get("url")
                        if img_url:
                            # 处理转义字符
                            img_url = img_url.replace("\\u0026", "&")
                            
                            try:
                                img_response = requests.get(img_url, timeout=30)
                                if img_response.status_code == 200:
                                    filename = f"image_{i+1}_{j+1}_{int(time.time())}.jpg"
                                    filepath = os.path.join(output_dir, filename)
                                    
                                    with open(filepath, "wb") as f:
                                        f.write(img_response.content)
                                    
                                    print(f"✅ 图片已保存: {filepath}")
                                    results.append(filepath)
                            except Exception as e:
                                print(f"下载图片时出错: {str(e)}")
                    break
                else:
                    retries += 1
                    print(f"重试 {retries}/{max_retries}...")
                    time.sleep(2)
            
            # 避免请求过于频繁
            if i < len(prompts) - 1:
                time.sleep(delay)
        
        return results

# 使用示例
if __name__ == "__main__":
    # 配置API密钥（建议使用环境变量）
    API_KEY = "your_api_key_here"
    
    # 初始化生成器
    generator = DouBaoImageGenerator(API_KEY)
    
    # 批量生成的prompt列表
    prompts = [
        "梦幻的城堡坐落在云端，周围有彩虹和飞翔的独角兽，手绘水彩风格",
        "未来科技城市夜景，高楼大厦灯火通明，飞行器在空中穿梭，赛博朋克风格",
        "宁静的日式庭院，有樱花树、石灯笼和小池塘，水墨画风格",
        "可爱的卡通小猫在花园里追蝴蝶，色彩鲜艳，儿童插画风格",
        "壮观的雪山日出景象，山峰巍峨，云海翻腾，写实摄影风格"
    ]
    
    # 生成图片
    generated_files = generator.generate_batch_images(
        prompts,
        output_dir="./my_batch_images",
        size="2K",
        delay=2  # 每张图片之间的延迟时间
    )
    
    print(f"\n批量生成完成！共生成 {len(generated_files)} 张图片")

4.2 高级批量生成代码（支持并发和进度跟踪）

import requests
import json
import os
import time
import re
import pathlib
import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed

class AdvancedDouBaoGenerator:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ark.cn-beijing.volces.com/api/v3/images/generations"
        self.headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }
        self.session = requests.Session()
        self.session.headers.update(self.headers)
    
    def _safe_filename(self, text, max_length=40):
        """生成安全的文件名"""
        # 移除特殊字符
        filename = re.sub(r'[^\w\s-]', '', text.strip())
        # 替换空格为下划线
        filename = re.sub(r'\s+', '_', filename)
        # 限制长度
        if len(filename) > max_length:
            filename = filename[:max_length]
        return filename or f"image_{int(time.time())}"
    
    def _download_image(self, url, save_path):
        """下载图片"""
        try:
            # 处理转义字符
            url = url.replace("\\u0026", "&")
            response = self.session.get(url, timeout=60)
            response.raise_for_status()
            
            with open(save_path, "wb") as f:
                f.write(response.content)
            return True
        except Exception as e:
            print(f"下载失败 {save_path}: {str(e)}")
            return False
    
    def generate_with_retry(self, prompt, model="doubao-seedream-4-0-250828", 
                          size="2K", seed=-1, max_retries=3):
        """带重试机制的图片生成"""
        for attempt in range(max_retries):
            try:
                payload = {
                    "model": model,
                    "prompt": prompt,
                    "size": size,
                    "response_format": "url",
                    "seed": seed
                }
                
                response = self.session.post(
                    self.base_url,
                    data=json.dumps(payload),
                    timeout=180
                )
                
                if response.status_code == 200:
                    return response.json()
                elif response.status_code == 429:
                    print(f"请求过于频繁，等待 {2**attempt} 秒后重试...")
                    time.sleep(2**attempt)
                else:
                    print(f"请求失败 {response.status_code}: {response.text}")
                    if attempt < max_retries - 1:
                        time.sleep(1)
                        
            except Exception as e:
                print(f"生成失败 (尝试 {attempt+1}/{max_retries}): {str(e)}")
                if attempt < max_retries - 1:
                    time.sleep(1)
        
        return None
    
    def batch_generate_concurrent(self, prompts, output_dir="./batch_output", 
                                 model="doubao-seedream-4-0-250828", size="2K",
                                 concurrency=3, delay_between_requests=0.5):
        """并发批量生成图片"""
        start_time = time.time()
        output_path = pathlib.Path(output_dir)
        output_path.mkdir(parents=True, exist_ok=True)
        
        results = {
            "success": [],
            "failed": [],
            "total": len(prompts)
        }
        
        def process_prompt(index, prompt):
            """处理单个prompt"""
            try:
                print(f"正在处理第 {index+1}/{len(prompts)}: {prompt[:50]}...")
                
                # 生成图片
                result = self.generate_with_retry(prompt, model, size)
                
                if result and "data" in result:
                    base_filename = self._safe_filename(prompt)
                    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
                    
                    for img_idx, item in enumerate(result["data"]):
                        img_url = item.get("url")
                        if img_url:
                            filename = f"{base_filename}_{timestamp}_{img_idx+1}.jpg"
                            save_path = output_path / filename
                            
                            if self._download_image(img_url, save_path):
                                results["success"].append(str(save_path))
                                print(f"✅ 成功: {filename}")
                            else:
                                results["failed"].append(prompt)
                else:
                    results["failed"].append(prompt)
                    print(f"❌ 失败: {prompt[:50]}...")
                
                # 控制请求频率
                time.sleep(delay_between_requests)
                
            except Exception as e:
                results["failed"].append(prompt)
                print(f"⚠️  错误: {str(e)}")
        
        # 使用线程池并发处理
        with ThreadPoolExecutor(max_workers=concurrency) as executor:
            futures = [executor.submit(process_prompt, i, prompt) 
                      for i, prompt in enumerate(prompts)]
            
            for future in as_completed(futures):
                future.result()
        
        # 生成报告
        end_time = time.time()
        duration = end_time - start_time
        
        print(f"\n" + "="*50)
        print(f"批量生成完成！")
        print(f"总耗时: {duration:.2f} 秒")
        print(f"成功: {len(results['success'])} 张")
        print(f"失败: {len(results['failed'])} 张")
        print(f"成功率: {len(results['success'])/len(prompts)*100:.1f}%")
        print("="*50)
        
        return results

# 使用示例
if __name__ == "__main__":
    # 从环境变量获取API密钥（推荐）
    import os
    API_KEY = os.getenv("DOUBAO_API_KEY", "your_api_key_here")
    
    generator = AdvancedDouBaoGenerator(API_KEY)
    
    # 批量生成示例
    fashion_prompts = [
        "优雅的女性穿着红色晚礼服，在豪华宴会厅中，金色灯光，高端时尚摄影风格",
        "年轻男性穿着休闲牛仔套装，在街头咖啡馆前，自然光线，生活方式摄影",
        "时尚模特穿着复古风格连衣裙，在老城区街道上，胶片摄影效果",
        "商务人士穿着西装，在现代办公楼前，专业商业摄影风格",
        "运动爱好者穿着运动服装，在健身房内，动感摄影，高能量氛围",
        "情侣穿着情侣装，在公园中漫步，温馨浪漫的摄影风格",
        "儿童穿着可爱的卡通服装，在游乐场中玩耍，童真摄影风格",
        "老年人穿着舒适的休闲服装，在花园中阅读，温暖感人的摄影风格"
    ]
    
    # 并发批量生成
    results = generator.batch_generate_concurrent(
        fashion_prompts,
        output_dir="./fashion_batch",
        size="2K",
        concurrency=2,  # 并发数，根据API限制调整
        delay_between_requests=1
    )
    
    # 保存生成报告
    with open("./batch_report.txt", "w", encoding="utf-8") as f:
        f.write(f"批量生成报告 - {datetime.datetime.now()}\n")
        f.write(f"总数量: {results['total']}\n")
        f.write(f"成功数量: {len(results['success'])}\n")
        f.write(f"失败数量: {len(results['failed'])}\n")
        f.write("\n成功生成的图片:\n")
        for path in results['success']:
            f.write(f"- {path}\n")
        if results['failed']:
            f.write("\n失败的prompt:\n")
            for prompt in results['failed']:
                f.write(f"- {prompt}\n")

高级参数配置

5.1 尺寸配置

支持的尺寸规格：

比例	推荐尺寸	适用场景
1:1	1328x1328, 2048x2048	社交媒体头像、正方形海报
16:9	1664x936, 2048x1152	横版海报、视频封面
9:16	936x1664, 1152x2048	竖版海报、手机壁纸
3:4	1104x1472	传统照片比例

5.2 高级参数详解

advanced_payload = {
    "model": "doubao-seedream-4-0-250828",
    "prompt": "精美的风景照片",
    "size": "2K",
    "negative_prompt": "模糊, 低质量, 变形, 文字",  # 避免的内容
    "cfg_scale": 8.5,  # 描述词关联度，7-10为宜
    "denoising_strength": 0.7,  # 重绘幅度，图生图时使用
    "seed": 12345,  # 固定种子可生成相同图片
    "steps": 30,  # 生成步数，20-35为宜
    "watermark": False,  # 是否添加水印
    "stream": False  # 是否流式返回
}

5.3 图生图参数配置

image_to_image_payload = {
    "model": "doubao-seedream-4-0-250828",
    "prompt": "将这张图片转换为梵高风格的油画",
    "image": "https://example.com/input.jpg",  # 参考图片URL
    "size": "2K",
    "denoising_strength": 0.6,  # 控制变化程度
    "controlnet": {
        "type": "edge",  # 边缘检测
        "strength": 0.8
    }
}

最佳实践与注意事项

6.1 API调用最佳实践

6.1.1 错误处理策略

def safe_api_call(generator, prompt, max_attempts=3):
    for attempt in range(max_attempts):
        try:
            result = generator.generate_single_image(prompt)
            if result:
                return result
        except Exception as e:
            print(f"尝试 {attempt+1} 失败: {str(e)}")
            if attempt < max_attempts - 1:
                time.sleep(2 ** attempt)  # 指数退避
    return None

6.1.2 速率控制

import time
import random

class RateLimiter:
    def __init__(self, max_requests_per_minute=60):
        self.max_requests = max_requests_per_minute
        self.requests = []
        self.minute_seconds = 60
    
    def wait_if_needed(self):
        now = time.time()
        # 清理1分钟前的请求记录
        self.requests = [req_time for req_time in self.requests 
                        if now - req_time < self.minute_seconds]
        
        if len(self.requests) >= self.max_requests:
            sleep_time = self.minute_seconds - (now - self.requests[0]) + 1
            print(f"速率限制，等待 {sleep_time:.1f} 秒")
            time.sleep(sleep_time)
        
        self.requests.append(time.time())

6.2 安全考虑

6.2.1 API密钥管理

# 推荐的API密钥管理方式
import os
from dotenv import load_dotenv

# 从.env文件加载
load_dotenv()
API_KEY = os.getenv("DOUBAO_API_KEY")

# 或者从环境变量获取
API_KEY = os.environ.get("DOUBAO_API_KEY")

6.2.2 输入验证

def validate_prompt(prompt):
    """验证prompt的安全性和有效性"""
    if not isinstance(prompt, str) or len(prompt) == 0:
        raise ValueError("prompt不能为空字符串")
    
    # 检查长度限制
    if len(prompt) > 2000:
        raise ValueError("prompt长度不能超过2000字符")
    
    # 检查敏感内容（示例）
    sensitive_patterns = ["色情", "暴力", "政治"]
    for pattern in sensitive_patterns:
        if pattern in prompt:
            raise ValueError("prompt包含不适当内容")
    
    return True

6.3 性能优化

6.3.1 连接池管理

import requests

# 使用连接池提高性能
session = requests.Session()
adapter = requests.adapters.HTTPAdapter(
    pool_connections=10,  # 连接池大小
    pool_maxsize=20,      # 最大连接数
    max_retries=3         # 重试次数
)
session.mount('https://', adapter)

6.3.2 异步处理

import asyncio
import aiohttp

async def async_generate_image(session, prompt, api_key):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {api_key}"
    }
    
    payload = {
        "model": "doubao-seedream-4-0-250828",
        "prompt": prompt,
        "size": "2K"
    }
    
    async with session.post(
        "https://ark.cn-beijing.volces.com/api/v3/images/generations",
        headers=headers,
        json=payload
    ) as response:
        return await response.json()

async def batch_async_generate(prompts, api_key):
    async with aiohttp.ClientSession() as session:
        tasks = [async_generate_image(session, prompt, api_key) 
                for prompt in prompts]
        results = await asyncio.gather(*tasks)
        return results

常见问题解决

7.1 认证问题

问题: 401 Unauthorized
解决方案:

# 检查API密钥
def verify_api_key(api_key):
    if not api_key or len(api_key) < 32:
        raise ValueError("无效的API密钥")
    
    # 测试请求
    test_prompt = "测试图片生成"
    generator = DouBaoImageGenerator(api_key)
    result = generator.generate_single_image(test_prompt, size="1024x1024")
    
    if not result:
        raise Exception("API密钥验证失败")
    return True

7.2 速率限制

问题: 429 Too Many Requests
解决方案:

class SmartRateLimiter:
    def __init__(self, initial_delay=1, max_delay=30, backoff_factor=2):
        self.delay = initial_delay
        self.max_delay = max_delay
        self.backoff_factor = backoff_factor
    
    def handle_429(self, response):
        # 从响应头获取重试时间
        retry_after = response.headers.get("Retry-After")
        if retry_after:
            sleep_time = int(retry_after)
        else:
            sleep_time = min(self.delay, self.max_delay)
        
        print(f"触发速率限制，等待 {sleep_time} 秒")
        time.sleep(sleep_time)
        
        # 指数退避
        self.delay *= self.backoff_factor
        return sleep_time

7.3 超时问题

问题: Request Timeout
解决方案:

def reliable_request(url, payload, headers, max_retries=3, timeout=180):
    for attempt in range(max_retries):
        try:
            response = requests.post(
                url,
                headers=headers,
                json=payload,
                timeout=timeout,
                allow_redirects=True
            )
            response.raise_for_status()
            return response
        except requests.exceptions.Timeout:
            print(f"请求超时，重试 {attempt+1}/{max_retries}")
            if attempt == max_retries - 1:
                raise
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {str(e)}")
            if attempt == max_retries - 1:
                raise

7.4 生成质量问题

问题: 生成的图片质量不佳
解决方案:

def optimize_prompt(prompt):
    """优化prompt以提高生成质量"""
    # 添加风格词
    style_enhancers = [
        "高清细节",
        "专业摄影", 
        "8K分辨率",
        "超写实",
        "大师级作品"
    ]
    
    # 添加构图词
    composition_enhancers = [
        "黄金比例构图",
        "完美光影",
        "专业打光",
        "电影级画质"
    ]
    
    # 随机选择增强词
    import random
    selected_style = random.choice(style_enhancers)
    selected_composition = random.choice(composition_enhancers)
    
    return f"{prompt}，{selected_style}，{selected_composition}"

总结

豆包API提供了强大的批量图生图功能，通过合理的代码设计和参数配置，可以实现高效的图片批量生成。关键要点包括：

1. 正确配置API环境：获取有效的API Key并妥善管理
2. 选择合适的生成策略：根据需求选择单次批量、循环生成或并发生成
3. 优化参数设置：合理配置尺寸、风格、质量等参数
4. 实施错误处理：添加重试机制、速率控制和超时处理
5. 确保安全性：保护API密钥，验证输入内容
6. 监控性能：跟踪生成进度，优化生成效率