13c - API集成编程生成

本文是《AI Agent 实战手册》第 13 章第 3 节。上一节：13b-图像Prompt工程 | 下一节：13d-批量生成自动化

概述

当你需要在应用中自动化生成图像——无论是电商产品图、社交媒体配图还是动态营销素材——就需要通过 API 将图像生成能力集成到代码中。本节覆盖三大主流平台（OpenAI GPT-image-1、Black Forest Labs FLUX.2、Stability AI Stable Diffusion 3.5）的 API 集成，提供 Python 和 TypeScript 双语言完整代码示例，涵盖认证、请求/响应处理、错误处理、速率限制和成本优化。

1. 平台与 API 对比

工具推荐

平台	模型	API 类型	价格（每张图）	最大分辨率	延迟	适用场景
OpenAI	GPT-image-1	REST (Responses API)	$0.02-$0.19（按质量）	1536×1024	10-30s	文字渲染精准、多模态编辑、指令遵循
OpenAI	DALL-E 3	REST (Images API)	$0.04-$0.12	1792×1024	5-15s	通用生成、快速原型
Black Forest Labs	FLUX.2 Pro	REST (异步轮询)	$0.03/百万像素起	~4MP	3-10s	照片级真实感、高保真度
Black Forest Labs	FLUX.2 Klein	REST (异步轮询)	$0.014/百万像素起	~4MP	<1s	实时生成、高吞吐量
Stability AI	SD 3.5 Large	REST (v2beta)	$0.065/张（6.5 credits）	1536×1536	5-15s	开源生态、本地部署可选
Stability AI	Stable Image Ultra	REST (v2beta)	$0.08/张（8 credits）	1536×1536	8-20s	最高质量、照片级输出

认证方式对比

平台	认证方式	API Key 获取	组织验证
OpenAI	Bearer Token (`Authorization: Bearer sk-...`)	platform.openai.com	GPT-image-1 需要组织验证
Black Forest Labs	Header Key (`X-Key: ...`)	api.bfl.ai	不需要
Stability AI	Bearer Token (`Authorization: Bearer sk-...`)	platform.stability.ai	不需要

2. OpenAI GPT-image-1 / DALL-E 3 集成

2.1 环境准备


# Python
pip install openai Pillow
 
# Node.js / TypeScript
npm install openai

环境变量配置：


export OPENAI_API_KEY="sk-proj-..."

2.2 GPT-image-1 生成（Responses API）

GPT-image-1 于 2025 年 4 月发布，使用 OpenAI 的 Responses API（而非传统 Images API），支持多模态输入和精准文字渲染。

Python 示例


import base64
import os
from pathlib import Path
from openai import OpenAI
 
client = OpenAI()  # 自动读取 OPENAI_API_KEY
 
def generate_image_gpt(
    prompt: str,
    quality: str = "medium",  # low | medium | high
    size: str = "1024x1024",  # 1024x1024 | 1536x1024 | 1024x1536 | auto
    output_path: str = "output.png",
) -> str:
    """使用 GPT-image-1 生成图像"""
    result = client.responses.create(
        model="gpt-image-1",
        input=prompt,
        tools=[{"type": "image_generation", "quality": quality, "size": size}],
    )
 
    # 从响应中提取图像数据
    for output in result.output:
        if output.type == "image_generation_call":
            image_bytes = base64.b64decode(output.result)
            Path(output_path).write_bytes(image_bytes)
            print(f"✅ 图像已保存: {output_path}")
            return output_path
 
    raise RuntimeError("未生成图像")
 
 
# 使用示例
generate_image_gpt(
    prompt="一只橘猫坐在堆满书的书桌上，旁边有一杯冒着热气的咖啡，"
           "窗外是雨天，温暖的台灯光照，水彩画风格",
    quality="high",
    size="1536x1024",
    output_path="cat_study.png",
)

TypeScript 示例


import OpenAI from "openai";
import { writeFileSync } from "fs";
 
const openai = new OpenAI(); // 自动读取 OPENAI_API_KEY
 
async function generateImageGPT(
  prompt: string,
  options: {
    quality?: "low" | "medium" | "high";
    size?: "1024x1024" | "1536x1024" | "1024x1536" | "auto";
    outputPath?: string;
  } = {}
): Promise<string> {
  const { quality = "medium", size = "1024x1024", outputPath = "output.png" } = options;
 
  const result = await openai.responses.create({
    model: "gpt-image-1",
    input: prompt,
    tools: [{ type: "image_generation", quality, size }],
  });
 
  // 提取图像数据
  for (const output of result.output) {
    if (output.type === "image_generation_call") {
      const imageBuffer = Buffer.from(output.result, "base64");
      writeFileSync(outputPath, imageBuffer);
      console.log(`✅ 图像已保存: ${outputPath}`);
      return outputPath;
    }
  }
 
  throw new Error("未生成图像");
}
 
// 使用示例
generateImageGPT(
  "一只橘猫坐在堆满书的书桌上，温暖的台灯光照，水彩画风格",
  { quality: "high", size: "1536x1024", outputPath: "cat_study.png" }
);

2.3 DALL-E 3 生成（Images API）

DALL-E 3 使用传统的 Images API，接口更简单，适合快速集成。

Python 示例


from openai import OpenAI
import urllib.request
 
client = OpenAI()
 
def generate_image_dalle3(
    prompt: str,
    size: str = "1024x1024",  # 1024x1024 | 1792x1024 | 1024x1792
    quality: str = "standard",  # standard | hd
    output_path: str = "dalle3_output.png",
) -> str:
    """使用 DALL-E 3 生成图像"""
    response = client.images.generate(
        model="dall-e-3",
        prompt=prompt,
        size=size,
        quality=quality,
        n=1,
    )
 
    image_url = response.data[0].url
    revised_prompt = response.data[0].revised_prompt
    print(f"📝 修正后的 prompt: {revised_prompt}")
 
    # 下载图像
    urllib.request.urlretrieve(image_url, output_path)
    print(f"✅ 图像已保存: {output_path}")
    return output_path
 
 
generate_image_dalle3(
    prompt="极简主义风格的山水画，留白大量，只有远山和一叶扁舟",
    size="1792x1024",
    quality="hd",
)

TypeScript 示例


import OpenAI from "openai";
import { writeFileSync } from "fs";
 
const openai = new OpenAI();
 
async function generateImageDalle3(
  prompt: string,
  options: {
    size?: "1024x1024" | "1792x1024" | "1024x1792";
    quality?: "standard" | "hd";
    outputPath?: string;
  } = {}
): Promise<string> {
  const { size = "1024x1024", quality = "standard", outputPath = "dalle3_output.png" } = options;
 
  const response = await openai.images.generate({
    model: "dall-e-3",
    prompt,
    size,
    quality,
    n: 1,
  });
 
  const imageUrl = response.data[0].url!;
  console.log(`📝 修正后的 prompt: ${response.data[0].revised_prompt}`);
 
  // 下载图像
  const imageResponse = await fetch(imageUrl);
  const buffer = Buffer.from(await imageResponse.arrayBuffer());
  writeFileSync(outputPath, buffer);
  console.log(`✅ 图像已保存: ${outputPath}`);
  return outputPath;
}
 
generateImageDalle3(
  "极简主义风格的山水画，留白大量，只有远山和一叶扁舟",
  { size: "1792x1024", quality: "hd" }
);

2.4 OpenAI 错误处理与速率限制


import time
from openai import OpenAI, RateLimitError, APIError, APITimeoutError
 
client = OpenAI()
 
def generate_with_retry(prompt: str, max_retries: int = 3) -> str:
    """带重试和错误处理的图像生成"""
    for attempt in range(max_retries):
        try:
            result = client.responses.create(
                model="gpt-image-1",
                input=prompt,
                tools=[{"type": "image_generation", "quality": "medium"}],
            )
            for output in result.output:
                if output.type == "image_generation_call":
                    return output.result  # base64 数据
            raise RuntimeError("响应中无图像数据")
 
        except RateLimitError as e:
            wait_time = 2 ** attempt * 10  # 指数退避: 10s, 20s, 40s
            print(f"⚠️ 速率限制，等待 {wait_time}s... ({attempt + 1}/{max_retries})")
            time.sleep(wait_time)
 
        except APITimeoutError:
            print(f"⏱️ 请求超时，重试中... ({attempt + 1}/{max_retries})")
            time.sleep(5)
 
        except APIError as e:
            if e.status_code and e.status_code >= 500:
                print(f"🔥 服务器错误 {e.status_code}，重试中...")
                time.sleep(5)
            else:
                raise  # 客户端错误不重试
 
    raise RuntimeError(f"生成失败，已重试 {max_retries} 次")

提示词模板


你是一个产品摄影师。请生成以下产品的商业摄影图：

产品：[产品名称]
风格：[简约白底 / 生活场景 / 创意概念]
角度：[正面 / 45度 / 俯拍]
光照：[柔和自然光 / 工作室灯光 / 戏剧性侧光]
背景：[纯白 / 渐变 / 场景化]

要求：高清晰度，产品细节清晰可见，适合电商平台使用。

3. Black Forest Labs FLUX.2 集成

FLUX.2 由 Stable Diffusion 原始团队创建，以照片级真实感和精准文字渲染著称。API 采用异步轮询模式。

3.1 环境准备


# Python
pip install requests Pillow
 
# Node.js / TypeScript
npm install node-fetch  # 或使用内置 fetch (Node 18+)


export BFL_API_KEY="your-bfl-api-key"

3.2 FLUX.2 Pro 生成

Python 示例


import os
import time
import requests
from pathlib import Path
 
BFL_API_KEY = os.environ["BFL_API_KEY"]
BASE_URL = "https://api.bfl.ai/v1"
 
def generate_image_flux(
    prompt: str,
    model: str = "flux-2-pro",  # flux-2-pro | flux-2-klein-9b | flux-2-flex
    width: int = 1024,
    height: int = 1024,
    output_path: str = "flux_output.png",
    poll_interval: int = 2,
    timeout: int = 120,
) -> str:
    """使用 FLUX.2 生成图像（异步轮询模式）"""
    headers = {"X-Key": BFL_API_KEY, "Content-Type": "application/json"}
 
    # 步骤 1: 提交生成请求
    payload = {
        "prompt": prompt,
        "width": width,
        "height": height,
    }
    response = requests.post(
        f"{BASE_URL}/{model}", json=payload, headers=headers
    )
    response.raise_for_status()
    task = response.json()
    task_id = task["id"]
    polling_url = task.get("polling_url", f"{BASE_URL}/get_result?id={task_id}")
    print(f"📤 任务已提交: {task_id}")
 
    # 步骤 2: 轮询结果
    start_time = time.time()
    while time.time() - start_time < timeout:
        result = requests.get(polling_url, headers=headers)
        result.raise_for_status()
        data = result.json()
 
        status = data.get("status")
        if status == "Ready":
            image_url = data["result"]["sample"]
            # 下载图像
            img_response = requests.get(image_url)
            Path(output_path).write_bytes(img_response.content)
            print(f"✅ 图像已保存: {output_path}")
            return output_path
        elif status == "Error":
            raise RuntimeError(f"生成失败: {data.get('error', '未知错误')}")
        else:
            print(f"⏳ 状态: {status}，等待 {poll_interval}s...")
            time.sleep(poll_interval)
 
    raise TimeoutError(f"生成超时（{timeout}s）")
 
 
# 使用示例
generate_image_flux(
    prompt="A professional headshot of a confident business woman, "
           "soft studio lighting, neutral background, sharp focus, 8K",
    model="flux-2-pro",
    width=1024,
    height=1024,
    output_path="headshot.png",
)

TypeScript 示例


import { writeFileSync } from "fs";
 
const BFL_API_KEY = process.env.BFL_API_KEY!;
const BASE_URL = "https://api.bfl.ai/v1";
 
interface FluxTask {
  id: string;
  polling_url?: string;
}
 
interface FluxResult {
  status: "Ready" | "Pending" | "Processing" | "Error";
  result?: { sample: string };
  error?: string;
}
 
async function generateImageFlux(
  prompt: string,
  options: {
    model?: string;
    width?: number;
    height?: number;
    outputPath?: string;
    pollInterval?: number;
    timeout?: number;
  } = {}
): Promise<string> {
  const {
    model = "flux-2-pro",
    width = 1024,
    height = 1024,
    outputPath = "flux_output.png",
    pollInterval = 2000,
    timeout = 120000,
  } = options;
 
  const headers = {
    "X-Key": BFL_API_KEY,
    "Content-Type": "application/json",
  };
 
  // 步骤 1: 提交生成请求
  const submitRes = await fetch(`${BASE_URL}/${model}`, {
    method: "POST",
    headers,
    body: JSON.stringify({ prompt, width, height }),
  });
  if (!submitRes.ok) throw new Error(`提交失败: ${submitRes.status}`);
 
  const task: FluxTask = await submitRes.json();
  const pollingUrl = task.polling_url ?? `${BASE_URL}/get_result?id=${task.id}`;
  console.log(`📤 任务已提交: ${task.id}`);
 
  // 步骤 2: 轮询结果
  const startTime = Date.now();
  while (Date.now() - startTime < timeout) {
    const resultRes = await fetch(pollingUrl, { headers });
    const data: FluxResult = await resultRes.json();
 
    if (data.status === "Ready" && data.result) {
      const imgRes = await fetch(data.result.sample);
      const buffer = Buffer.from(await imgRes.arrayBuffer());
      writeFileSync(outputPath, buffer);
      console.log(`✅ 图像已保存: ${outputPath}`);
      return outputPath;
    } else if (data.status === "Error") {
      throw new Error(`生成失败: ${data.error ?? "未知错误"}`);
    }
 
    console.log(`⏳ 状态: ${data.status}，等待中...`);
    await new Promise((r) => setTimeout(r, pollInterval));
  }
 
  throw new Error(`生成超时（${timeout / 1000}s）`);
}
 
// 使用示例
generateImageFlux(
  "A professional headshot, soft studio lighting, neutral background, 8K",
  { model: "flux-2-pro", width: 1024, height: 1024, outputPath: "headshot.png" }
);

3.3 FLUX 错误处理


import requests
from requests.exceptions import HTTPError
 
def flux_generate_safe(prompt: str, max_retries: int = 3) -> str:
    """带错误处理的 FLUX 生成"""
    headers = {"X-Key": BFL_API_KEY, "Content-Type": "application/json"}
 
    for attempt in range(max_retries):
        try:
            resp = requests.post(
                f"{BASE_URL}/flux-2-pro",
                json={"prompt": prompt, "width": 1024, "height": 1024},
                headers=headers,
                timeout=30,
            )
            resp.raise_for_status()
            return resp.json()["id"]
 
        except HTTPError as e:
            status = e.response.status_code
            if status == 402:
                raise RuntimeError("❌ 余额不足，请充值") from e
            elif status == 429:
                wait = 2 ** attempt * 5
                print(f"⚠️ 速率限制，等待 {wait}s...")
                time.sleep(wait)
            elif status >= 500:
                print(f"🔥 服务器错误，重试中...")
                time.sleep(3)
            else:
                raise
        except requests.Timeout:
            print(f"⏱️ 请求超时，重试中...")
 
    raise RuntimeError("生成失败，已达最大重试次数")

4. Stability AI（Stable Diffusion 3.5）集成

Stability AI 提供 REST API，使用 credit 计费系统（1 credit = $0.01）。2025 年 4 月起，SD 3.0 已自动升级至 SD 3.5。

4.1 环境准备


# Python
pip install requests Pillow
 
# Node.js / TypeScript
npm install node-fetch form-data  # 或使用内置 fetch


export STABILITY_API_KEY="sk-..."

4.2 Stable Image Ultra 生成

Stability AI 的 v2beta API 使用 multipart/form-data 格式提交请求，响应直接返回图像二进制数据。

Python 示例


import os
import requests
from pathlib import Path
 
STABILITY_API_KEY = os.environ["STABILITY_API_KEY"]
BASE_URL = "https://api.stability.ai/v2beta"
 
def generate_image_stability(
    prompt: str,
    model: str = "ultra",  # ultra | core | sd3.5-large | sd3.5-medium
    aspect_ratio: str = "1:1",  # 1:1 | 16:9 | 9:16 | 4:3 | 3:4 | 21:9 等
    negative_prompt: str = "",
    seed: int = 0,  # 0 = 随机
    output_path: str = "stability_output.png",
) -> str:
    """使用 Stability AI 生成图像"""
    endpoint_map = {
        "ultra": f"{BASE_URL}/stable-image/generate/ultra",
        "core": f"{BASE_URL}/stable-image/generate/core",
        "sd3.5-large": f"{BASE_URL}/stable-image/generate/sd3",
        "sd3.5-medium": f"{BASE_URL}/stable-image/generate/sd3",
    }
    url = endpoint_map.get(model, endpoint_map["ultra"])
 
    headers = {
        "Authorization": f"Bearer {STABILITY_API_KEY}",
        "Accept": "image/*",  # 直接返回图像二进制
    }
 
    form_data = {
        "prompt": (None, prompt),
        "aspect_ratio": (None, aspect_ratio),
        "output_format": (None, "png"),
    }
    if negative_prompt:
        form_data["negative_prompt"] = (None, negative_prompt)
    if seed > 0:
        form_data["seed"] = (None, str(seed))
    if model.startswith("sd3.5"):
        form_data["model"] = (None, model.replace(".", "-"))
 
    response = requests.post(url, headers=headers, files=form_data)
 
    if response.status_code == 200:
        Path(output_path).write_bytes(response.content)
        print(f"✅ 图像已保存: {output_path}")
        print(f"💰 消耗 credits: {response.headers.get('x-credits-consumed', 'N/A')}")
        return output_path
    else:
        raise RuntimeError(
            f"生成失败 [{response.status_code}]: {response.json().get('message', response.text)}"
        )
 
 
# 使用示例
generate_image_stability(
    prompt="A serene Japanese zen garden with raked sand patterns, "
           "moss-covered stones, and a single cherry blossom tree, "
           "morning mist, photorealistic",
    model="ultra",
    aspect_ratio="16:9",
    negative_prompt="cartoon, anime, low quality, blurry",
    output_path="zen_garden.png",
)

TypeScript 示例


import { writeFileSync } from "fs";
 
const STABILITY_API_KEY = process.env.STABILITY_API_KEY!;
const BASE_URL = "https://api.stability.ai/v2beta";
 
async function generateImageStability(
  prompt: string,
  options: {
    model?: "ultra" | "core" | "sd3.5-large" | "sd3.5-medium";
    aspectRatio?: string;
    negativePrompt?: string;
    seed?: number;
    outputPath?: string;
  } = {}
): Promise<string> {
  const {
    model = "ultra",
    aspectRatio = "1:1",
    negativePrompt = "",
    seed = 0,
    outputPath = "stability_output.png",
  } = options;
 
  const endpointMap: Record<string, string> = {
    ultra: `${BASE_URL}/stable-image/generate/ultra`,
    core: `${BASE_URL}/stable-image/generate/core`,
    "sd3.5-large": `${BASE_URL}/stable-image/generate/sd3`,
    "sd3.5-medium": `${BASE_URL}/stable-image/generate/sd3`,
  };
 
  const formData = new FormData();
  formData.append("prompt", prompt);
  formData.append("aspect_ratio", aspectRatio);
  formData.append("output_format", "png");
  if (negativePrompt) formData.append("negative_prompt", negativePrompt);
  if (seed > 0) formData.append("seed", seed.toString());
 
  const response = await fetch(endpointMap[model], {
    method: "POST",
    headers: {
      Authorization: `Bearer ${STABILITY_API_KEY}`,
      Accept: "image/*",
    },
    body: formData,
  });
 
  if (!response.ok) {
    const error = await response.json();
    throw new Error(`生成失败 [${response.status}]: ${error.message ?? "未知错误"}`);
  }
 
  const buffer = Buffer.from(await response.arrayBuffer());
  writeFileSync(outputPath, buffer);
  console.log(`✅ 图像已保存: ${outputPath}`);
  console.log(`💰 消耗 credits: ${response.headers.get("x-credits-consumed") ?? "N/A"}`);
  return outputPath;
}
 
// 使用示例
generateImageStability(
  "A serene Japanese zen garden, morning mist, photorealistic",
  { model: "ultra", aspectRatio: "16:9", outputPath: "zen_garden.png" }
);

4.3 Stability AI 错误处理


def stability_generate_safe(prompt: str, max_retries: int = 3) -> bytes:
    """带错误处理的 Stability AI 生成"""
    headers = {
        "Authorization": f"Bearer {STABILITY_API_KEY}",
        "Accept": "image/*",
    }
 
    for attempt in range(max_retries):
        try:
            resp = requests.post(
                f"{BASE_URL}/stable-image/generate/ultra",
                headers=headers,
                files={"prompt": (None, prompt), "output_format": (None, "png")},
                timeout=60,
            )
 
            if resp.status_code == 200:
                return resp.content
            elif resp.status_code == 402:
                raise RuntimeError("❌ Credits 不足，请充值")
            elif resp.status_code == 429:
                wait = 2 ** attempt * 5
                print(f"⚠️ 速率限制，等待 {wait}s...")
                time.sleep(wait)
            elif resp.status_code >= 500:
                print(f"🔥 服务器错误，重试中...")
                time.sleep(3)
            else:
                error_msg = resp.json().get("message", resp.text)
                raise RuntimeError(f"请求失败 [{resp.status_code}]: {error_msg}")
 
        except requests.Timeout:
            print(f"⏱️ 请求超时，重试中... ({attempt + 1}/{max_retries})")
 
    raise RuntimeError("生成失败，已达最大重试次数")

5. 统一封装：多平台图像生成服务

提示词模板


请为 [业务场景] 生成一张图像：

主题：[具体描述]
风格：[照片级真实 / 插画 / 水彩 / 3D渲染]
宽高比：[1:1 / 16:9 / 9:16 / 4:3]
质量要求：[标准 / 高清]
预算限制：[每张不超过 $X]

请根据预算和质量要求自动选择最合适的平台和模型。

Python 统一封装


from dataclasses import dataclass
from enum import Enum
from typing import Optional
 
class Platform(Enum):
    OPENAI_GPT = "openai_gpt"
    OPENAI_DALLE = "openai_dalle"
    FLUX = "flux"
    STABILITY = "stability"
 
@dataclass
class GenerationResult:
    platform: Platform
    output_path: str
    cost_estimate: float  # 美元
    generation_time: float  # 秒
 
class ImageGenerationService:
    """多平台图像生成统一服务"""
 
    def __init__(self):
        self.openai_client = OpenAI() if os.getenv("OPENAI_API_KEY") else None
        self.bfl_key = os.getenv("BFL_API_KEY")
        self.stability_key = os.getenv("STABILITY_API_KEY")
 
    def generate(
        self,
        prompt: str,
        platform: Platform = Platform.FLUX,
        output_path: str = "output.png",
        **kwargs,
    ) -> GenerationResult:
        """统一生成接口"""
        start = time.time()
 
        if platform == Platform.OPENAI_GPT:
            generate_image_gpt(prompt, output_path=output_path, **kwargs)
            cost = {"low": 0.02, "medium": 0.07, "high": 0.19}.get(
                kwargs.get("quality", "medium"), 0.07
            )
        elif platform == Platform.OPENAI_DALLE:
            generate_image_dalle3(prompt, output_path=output_path, **kwargs)
            cost = 0.04 if kwargs.get("quality") != "hd" else 0.08
        elif platform == Platform.FLUX:
            generate_image_flux(prompt, output_path=output_path, **kwargs)
            cost = 0.03  # FLUX.2 Pro 基础价格
        elif platform == Platform.STABILITY:
            generate_image_stability(prompt, output_path=output_path, **kwargs)
            cost = 0.08  # Ultra 价格
        else:
            raise ValueError(f"不支持的平台: {platform}")
 
        elapsed = time.time() - start
        return GenerationResult(platform, output_path, cost, elapsed)
 
    def auto_select(
        self,
        prompt: str,
        max_cost: float = 0.10,
        priority: str = "quality",  # quality | speed | cost
    ) -> Platform:
        """根据预算和优先级自动选择平台"""
        if priority == "cost":
            if max_cost >= 0.03 and self.bfl_key:
                return Platform.FLUX
            elif max_cost >= 0.04:
                return Platform.OPENAI_DALLE
            return Platform.STABILITY
        elif priority == "speed":
            return Platform.FLUX  # FLUX Klein 亚秒级
        else:  # quality
            if max_cost >= 0.19:
                return Platform.OPENAI_GPT
            elif max_cost >= 0.08:
                return Platform.STABILITY
            return Platform.FLUX
 
 
# 使用示例
service = ImageGenerationService()
platform = service.auto_select("产品摄影", max_cost=0.10, priority="quality")
result = service.generate(
    "专业产品摄影，白色背景，柔和灯光",
    platform=platform,
    output_path="product.png",
)
print(f"平台: {result.platform.value}, 耗时: {result.generation_time:.1f}s, "
      f"预估成本: ${result.cost_estimate:.2f}")

实战案例：构建图像生成微服务

以下是一个基于 FastAPI 的图像生成微服务，支持多平台切换、请求队列和成本追踪。

架构概览


客户端 → FastAPI 网关 → 平台路由器 → OpenAI / FLUX / Stability
                ↓
          Redis 队列（可选）
                ↓
          对象存储（S3/R2）

完整实现（Python + FastAPI）


"""图像生成微服务 - main.py"""
import asyncio
import base64
import os
import time
import uuid
from contextlib import asynccontextmanager
from pathlib import Path
 
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from openai import AsyncOpenAI
 
# ── 配置 ──────────────────────────────────────────
UPLOAD_DIR = Path("generated_images")
UPLOAD_DIR.mkdir(exist_ok=True)
 
openai_client = AsyncOpenAI()
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用生命周期管理"""
    app.state.http_client = httpx.AsyncClient(timeout=120)
    app.state.total_cost = 0.0
    app.state.total_images = 0
    yield
    await app.state.http_client.aclose()
 
app = FastAPI(title="图像生成微服务", version="1.0.0", lifespan=lifespan)
 
# ── 数据模型 ──────────────────────────────────────
class GenerateRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=4000)
    platform: str = Field(default="flux", pattern="^(openai|flux|stability)$")
    quality: str = Field(default="medium")
    aspect_ratio: str = Field(default="1:1")
 
class GenerateResponse(BaseModel):
    image_id: str
    image_url: str
    platform: str
    cost_estimate: float
    generation_time: float
 
class StatsResponse(BaseModel):
    total_images: int
    total_cost: float
    average_cost: float
 
# ── API 端点 ──────────────────────────────────────
@app.post("/generate", response_model=GenerateResponse)
async def generate_image(req: GenerateRequest):
    """生成图像的统一端点"""
    start = time.time()
    image_id = str(uuid.uuid4())[:8]
    output_path = UPLOAD_DIR / f"{image_id}.png"
 
    try:
        if req.platform == "openai":
            cost = await _generate_openai(req.prompt, req.quality, output_path)
        elif req.platform == "flux":
            cost = await _generate_flux(req.prompt, output_path)
        elif req.platform == "stability":
            cost = await _generate_stability(
                req.prompt, req.aspect_ratio, output_path
            )
        else:
            raise HTTPException(400, f"不支持的平台: {req.platform}")
    except Exception as e:
        raise HTTPException(500, f"生成失败: {str(e)}")
 
    elapsed = time.time() - start
    app.state.total_cost += cost
    app.state.total_images += 1
 
    return GenerateResponse(
        image_id=image_id,
        image_url=f"/images/{image_id}.png",
        platform=req.platform,
        cost_estimate=cost,
        generation_time=round(elapsed, 2),
    )
 
@app.get("/stats", response_model=StatsResponse)
async def get_stats():
    """获取使用统计"""
    total = app.state.total_images
    return StatsResponse(
        total_images=total,
        total_cost=round(app.state.total_cost, 4),
        average_cost=round(app.state.total_cost / max(total, 1), 4),
    )
 
# ── 平台实现 ──────────────────────────────────────
async def _generate_openai(prompt: str, quality: str, path: Path) -> float:
    result = await openai_client.responses.create(
        model="gpt-image-1",
        input=prompt,
        tools=[{"type": "image_generation", "quality": quality, "size": "1024x1024"}],
    )
    for output in result.output:
        if output.type == "image_generation_call":
            path.write_bytes(base64.b64decode(output.result))
            return {"low": 0.02, "medium": 0.07, "high": 0.19}.get(quality, 0.07)
    raise RuntimeError("OpenAI 未返回图像")
 
async def _generate_flux(prompt: str, path: Path) -> float:
    client = app.state.http_client
    headers = {"X-Key": os.environ["BFL_API_KEY"], "Content-Type": "application/json"}
 
    resp = await client.post(
        "https://api.bfl.ai/v1/flux-2-pro",
        json={"prompt": prompt, "width": 1024, "height": 1024},
        headers=headers,
    )
    resp.raise_for_status()
    task = resp.json()
    polling_url = task.get("polling_url", f"https://api.bfl.ai/v1/get_result?id={task['id']}")
 
    for _ in range(60):
        result = await client.get(polling_url, headers=headers)
        data = result.json()
        if data["status"] == "Ready":
            img = await client.get(data["result"]["sample"])
            path.write_bytes(img.content)
            return 0.03
        elif data["status"] == "Error":
            raise RuntimeError(data.get("error", "FLUX 生成错误"))
        await asyncio.sleep(2)
    raise TimeoutError("FLUX 生成超时")
 
async def _generate_stability(prompt: str, aspect_ratio: str, path: Path) -> float:
    client = app.state.http_client
    headers = {
        "Authorization": f"Bearer {os.environ['STABILITY_API_KEY']}",
        "Accept": "image/*",
    }
    # httpx multipart 格式
    data = {"prompt": prompt, "aspect_ratio": aspect_ratio, "output_format": "png"}
    resp = await client.post(
        "https://api.stability.ai/v2beta/stable-image/generate/ultra",
        data=data,
        headers=headers,
    )
    if resp.status_code == 200:
        path.write_bytes(resp.content)
        return 0.08
    raise RuntimeError(f"Stability 错误: {resp.text}")

启动与测试


# 安装依赖
pip install fastapi uvicorn httpx openai
 
# 启动服务
uvicorn main:app --host 0.0.0.0 --port 8000
 
# 测试请求
curl -X POST http://localhost:8000/generate \
  -H "Content-Type: application/json" \
  -d '{"prompt": "一只可爱的柴犬戴着太阳镜", "platform": "flux"}'
 
# 查看统计
curl http://localhost:8000/stats

案例分析

平台路由：通过 platform 参数灵活切换，生产环境可根据成本/质量需求自动路由
异步处理：使用 httpx.AsyncClient 和 asyncio 实现非阻塞 I/O，单实例可处理高并发
成本追踪：每次生成记录预估成本，便于预算监控
扩展方向：可接入 Redis 队列实现异步任务、S3/R2 存储替代本地文件、添加 API Key 认证和速率限制

避坑指南

❌ 常见错误

忽略 GPT-image-1 的组织验证要求
- 问题：直接调用 GPT-image-1 API 返回 403，因为该模型需要组织级别验证
- 正确做法：在 OpenAI 平台完成组织验证后再使用；开发阶段可先用 DALL-E 3
FLUX 轮询间隔过短导致被限流
- 问题：每 100ms 轮询一次结果，触发 429 错误
- 正确做法：轮询间隔至少 1-2 秒，使用指数退避策略；始终使用响应中返回的 polling_url
Stability AI 请求格式错误
- 问题：使用 JSON body 而非 multipart/form-data，返回 400 错误
- 正确做法：Stability v2beta API 要求 multipart/form-data 格式，同时设置 Accept: image/* 头以直接获取图像二进制
未处理临时 URL 过期
- 问题：FLUX 和 DALL-E 返回的图像 URL 有过期时间（通常 10-60 分钟），保存 URL 而非图像数据导致后续访问失败
- 正确做法：生成后立即下载图像并存储到自己的对象存储（S3/R2/OSS），不要依赖临时 URL
未设置请求超时
- 问题：图像生成可能因服务端问题长时间无响应，阻塞整个应用
- 正确做法：所有 HTTP 请求设置合理超时（提交请求 30s，轮询 120s），超时后重试或降级
成本失控——未设置预算上限
- 问题：批量生成时未限制并发数和总量，一夜之间产生高额账单
- 正确做法：实现每日/每月预算上限、并发限制、单次批量上限；使用 OpenAI 的 Usage Tier 和 Spending Limit 功能

✅ 最佳实践

使用环境变量管理 API Key，绝不硬编码在代码中；生产环境使用密钥管理服务（AWS Secrets Manager、Vault）
实现三级降级策略：首选平台 → 备选平台 → 缓存/占位图，确保服务可用性
缓存相同 prompt 的结果：对于重复请求，使用 prompt hash 作为缓存键，避免重复生成和计费
记录每次生成的元数据：prompt、平台、模型、参数、成本、耗时，便于后续分析和优化
使用 seed 参数保证可复现性：调试和 A/B 测试时固定 seed 值，确保相同 prompt 生成一致结果

参考来源

OpenAI: New GPT Image Model in the API （2025-04-23）
OpenAI Images and Vision Documentation （2025-04）
Black Forest Labs: FLUX API Pricing （2025）
Black Forest Labs: API Integration Guide （2025）
FLUX.2 Explained: Advanced Image Generation （2025-11）
Stability AI: API Pricing Update （2025-07）
Stability AI: Release Notes - SD 3.5 Migration （2025-04-17）
Stable Image Ultra Overview （2025）
GPT-image-1 Pricing Guide 2025 （2025-04）

📖 返回总览与导航 | 上一节：13b-图像Prompt工程 | 下一节：13d-批量生成自动化

13c - API集成编程生成

概述

1. 平台与 API 对比

工具推荐

认证方式对比

2. OpenAI GPT-image-1 / DALL-E 3 集成

2.1 环境准备

2.2 GPT-image-1 生成（Responses API）

Python 示例

TypeScript 示例

2.3 DALL-E 3 生成（Images API）

Python 示例

TypeScript 示例

2.4 OpenAI 错误处理与速率限制

提示词模板

3. Black Forest Labs FLUX.2 集成

3.1 环境准备

3.2 FLUX.2 Pro 生成

Python 示例

TypeScript 示例

3.3 FLUX 错误处理

4. Stability AI（Stable Diffusion 3.5）集成

4.1 环境准备

4.2 Stable Image Ultra 生成

Python 示例

TypeScript 示例

4.3 Stability AI 错误处理

5. 统一封装：多平台图像生成服务

提示词模板

Python 统一封装

实战案例：构建图像生成微服务

架构概览

完整实现（Python + FastAPI）

启动与测试

案例分析

避坑指南

❌ 常见错误

✅ 最佳实践

相关资源与延伸阅读

参考来源