FLUX 1.1 Pro API Python 教程：5 分钟内完成图像生成

3 个关键数字： 生成延迟约 10 秒（相比前代 FLUX.1 Pro 快 6 倍）、每张图像成本约 $0.04、在 GenAI Arena 公开评测中质量得分排名 第一。

---

目录

1. 前提条件
2. 认证与环境配置
3. 基础实现：第一张图像
4. 生产级实现
5. API 参数参考表
6. 错误处理
7. 性能与成本
8. 结论

---

1. 前提条件 {#prerequisites}

在运行任何代码之前，你需要以下内容：

账户与 API Key

• 在 Black Forest Labs API 平台 注册账户
• 生成一个 API key，格式为 bfl-xxxxxxxxxxxxxxxx
• 或者通过 Replicate 使用 FLUX 1.1 Pro（endpoint 不同，本文主要覆盖官方 BFL API）

Python 环境

Python 3.8 或更高版本。使用以下命令安装所需依赖：

# 安装必要的库
pip install requests pillow python-dotenv

# 如果你更倾向于使用官方客户端（可选）
pip install bfl

验证安装

python -c "import requests, PIL; print('依赖安装成功')"

环境变量配置

在项目根目录创建 .env 文件：

# .env
BFL_API_KEY=bfl-your-actual-api-key-here

注意： 永远不要把 API key 硬编码在代码里，也不要提交到版本控制系统。.gitignore 文件里要包含 .env。

---

2. 认证与环境配置 {#auth-setup}

BFL 官方 API 使用 HTTP header 认证，不需要 OAuth 流程。每个请求在 x-key header 中传递 API key。

FLUX 1.1 Pro 的 API 是 异步的：你先提交一个生成任务，拿到 task_id，然后轮询结果端点直到任务完成。这不同于某些同步 API（比如 DALL-E 3），需要你在代码里处理轮询逻辑。

# config.py — 所有配置集中管理
import os
from dotenv import load_dotenv

load_dotenv()  # 从 .env 文件加载环境变量

# BFL 官方 API 端点
BFL_API_BASE = "https://api.bfl.ml"
FLUX_11_PRO_ENDPOINT = f"{BFL_API_BASE}/v1/flux-pro-1.1"
RESULT_ENDPOINT = f"{BFL_API_BASE}/v1/get_result"

# 从环境变量读取 key，避免硬编码
API_KEY = os.getenv("BFL_API_KEY")
if not API_KEY:
    raise EnvironmentError(
        "BFL_API_KEY 未设置。请在 .env 文件中配置，"
        "或通过 export BFL_API_KEY=your_key 设置环境变量。"
    )

# 统一的请求 header
HEADERS = {
    "x-key": API_KEY,
    "Content-Type": "application/json",
    "Accept": "application/json",
}

运行 python config.py 来验证配置是否正确。如果没有报错，说明 API key 已经正确加载。

---

3. 基础实现：第一张图像 {#basic-implementation}

下面是最精简的实现版本。它完成两件事：提交生成任务、轮询直到拿到结果 URL。

# basic_generation.py — 最简单的可运行版本
import time
import requests
from config import FLUX_11_PRO_ENDPOINT, RESULT_ENDPOINT, HEADERS


def generate_image(prompt: str) -> str:
    """
    提交图像生成任务并等待完成。
    返回值：图像下载 URL（有效期约 10 分钟）
    """

    # Step 1: 提交生成任务
    # BFL API 是异步的，这一步只是排队，不会直接返回图像
    response = requests.post(
        FLUX_11_PRO_ENDPOINT,
        headers=HEADERS,
        json={
            "prompt": prompt,
            "width": 1024,   # 默认尺寸，1 megapixel
            "height": 1024,
        }
    )

    # 任务提交失败时立即抛出异常，避免进入无效的轮询循环
    response.raise_for_status()
    task_id = response.json()["id"]
    print(f"任务已提交，task_id: {task_id}")

    # Step 2: 轮询结果
    # BFL 官方文档建议轮询间隔不低于 0.5 秒，避免触发速率限制
    while True:
        time.sleep(1.5)  # 1.5 秒间隔，留有余量

        result_response = requests.get(
            RESULT_ENDPOINT,
            headers=HEADERS,
            params={"id": task_id}
        )
        result_response.raise_for_status()
        result = result_response.json()

        status = result.get("status")
        print(f"当前状态: {status}")

        if status == "Ready":
            # 图像 URL 在 result.sample 字段
            return result["result"]["sample"]

        if status in ("Error", "Failed", "Content Moderated"):
            # 任务失败，不需要继续轮询
            raise RuntimeError(f"生成失败，状态: {status}，详情: {result}")

        # 状态为 "Pending" 或 "Processing" 时继续等待


if __name__ == "__main__":
    image_url = generate_image(
        prompt="A photorealistic red fox sitting in a snowy forest, "
               "golden hour lighting, shallow depth of field"
    )
    print(f"\n图像已生成: {image_url}")

运行方式：

python basic_generation.py

正常情况下约 8–15 秒后你会看到一个图像 URL。URL 是临时的，有效期约 10 分钟，务必在此期间下载。

---

4. 生产级实现 {#production-implementation}

基础版本在 demo 阶段够用，但生产环境需要：超时控制、重试机制、图像本地保存、完整的异常分类处理。

# production_generator.py — 生产可用版本
import time
import uuid
import requests
from pathlib import Path
from dataclasses import dataclass, field
from typing import Optional
from config import FLUX_11_PRO_ENDPOINT, RESULT_ENDPOINT, HEADERS


@dataclass
class GenerationConfig:
    """
    图像生成参数配置。
    所有参数都有合理默认值，只需覆盖你关心的部分。
    """
    prompt: str
    width: int = 1024
    height: int = 1024
    prompt_upsampling: bool = False  # True 时 API 会自动扩展 prompt，增加细节但降低可控性
    safety_tolerance: int = 2        # 0-6，数值越高内容过滤越宽松，默认 2 适合大多数场景
    output_format: str = "jpeg"      # "jpeg" 或 "png"
    seed: Optional[int] = None       # 固定 seed 可复现结果；None 表示随机


@dataclass
class GenerationResult:
    task_id: str
    image_url: str
    local_path: Optional[str] = None
    elapsed_seconds: float = 0.0


class FluxProGenerator:
    """
    FLUX 1.1 Pro 图像生成器。
    包含重试、超时、本地保存功能。
    """

    MAX_POLL_ATTEMPTS = 60    # 最多轮询 60 次
    POLL_INTERVAL = 1.5       # 每次轮询间隔（秒）
    MAX_TIMEOUT = 120         # 整体超时上限（秒）

    def __init__(self, output_dir: str = "./generated_images"):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def submit(self, config: GenerationConfig) -> str:
        """提交生成任务，返回 task_id。"""
        payload = {
            "prompt": config.prompt,
            "width": config.width,
            "height": config.height,
            "prompt_upsampling": config.prompt_upsampling,
            "safety_tolerance": config.safety_tolerance,
            "output_format": config.output_format,
        }
        # seed 为 None 时不传该字段，让 API 使用随机 seed
        if config.seed is not None:
            payload["seed"] = config.seed

        resp = requests.post(
            FLUX_11_PRO_ENDPOINT,
            headers=HEADERS,
            json=payload,
            timeout=30  # 提交请求本身超时 30 秒
        )

        # 细化 HTTP 错误处理（详见 error handling 章节）
        if resp.status_code == 422:
            raise ValueError(f"参数验证失败: {resp.json()}")
        if resp.status_code == 429:
            raise RuntimeError("触发速率限制，请降低请求频率或升级套餐。")
        resp.raise_for_status()

        return resp.json()["id"]

    def poll(self, task_id: str) -> str:
        """轮询直到任务完成，返回图像 URL。"""
        start_time = time.time()

        for attempt in range(self.MAX_POLL_ATTEMPTS):
            # 超时检查：避免无限等待
            elapsed = time.time() - start_time
            if elapsed > self.MAX_TIMEOUT:
                raise TimeoutError(
                    f"任务 {task_id} 超过 {self.MAX_TIMEOUT} 秒未完成。"
                )

            time.sleep(self.POLL_INTERVAL)

            resp = requests.get(
                RESULT_ENDPOINT,
                headers=HEADERS,
                params={"id": task_id},
                timeout=10
            )
            resp.raise_for_status()
            result = resp.json()
            status = result.get("status")

            if status == "Ready":
                return result["result"]["sample"]

            if status == "Content Moderated":
                # 内容审核拦截，这不是网络错误，不应重试
                raise ValueError(
                    "图像被内容审核拦截。请修改 prompt 或降低 safety_tolerance。"
                )

            if status in ("Error", "Failed"):
                raise RuntimeError(
                    f"任务失败: {result.get('error', '未知错误')}"
                )

            # "Pending" / "Processing" 状态继续等待
            if attempt % 5 == 0:  # 每 5 次打印一次，避免日志太多
                print(f"  [{elapsed:.0f}s] 状态: {status}，继续等待...")

        raise TimeoutError(f"轮询次数超过上限 {self.MAX_POLL_ATTEMPTS}。")

    def save_image(self, image_url: str, filename: Optional[str] = None) -> str:
        """下载图像到本地，返回本地文件路径。"""
        if filename is None:
            # 用 uuid 避免文件名冲突
            filename = f"flux_{uuid.uuid4().hex[:8]}.jpg"

        local_path = self.output_dir / filename

        # stream=True 避免大文件一次性载入内存
        resp = requests.get(image_url, stream=True, timeout=60)
        resp.raise_for_status()

        with open(local_path, "wb") as f:
            for chunk in resp.iter_content(chunk_size=8192):
                f.write(chunk)

        return str(local_path)

    def generate(
        self,
        config: GenerationConfig,
        save_locally: bool = True
    ) -> GenerationResult:
        """端到端生成流程：提交 → 轮询 → （可选）保存。"""
        start = time.time()

        print(f"提交生成任务...")
        task_id = self.submit(config)
        print(f"task_id: {task_id}")

        print("等待生成完成...")
        image_url = self.poll(task_id)

        elapsed = time.time() - start
        print(f"生成完成，耗时 {elapsed:.1f} 秒")

        result = GenerationResult(
            task_id=task_id,
            image_url=image_url,
            elapsed_seconds=elapsed
        )

        if save_locally:
            result.local_path = self.save_image(image_url)
            print(f"图像已保存至: {result.local_path}")

        return result


# ── 使用示例 ──────────────────────────────────────────────
if __name__ == "__main__":
    generator = FluxProGenerator(output_dir="./output")

    config = GenerationConfig(
        prompt=(
            "A cinematic portrait of an astronaut standing on Mars, "
            "dust storm in background, warm orange tones, "
            "high detail, 8k resolution"
        ),
        width=1344,
        height=768,          # 16:9 横向构图
        output_format="jpeg",
        seed=42,             # 固定 seed，方便调试时复现结果
        safety_tolerance=2,
    )

    result = generator.generate(config, save_locally=True)

    print(f"\n--- 生成结果 ---")
    print(f"Task ID    : {result.task_id}")
    print(f"图像 URL   : {result.image_url}")
    print(f"本地路径   : {result.local_path}")
    print(f"耗时       : {result.elapsed_seconds:.1f}s")

批量生成示例（并发）：

如果你需要同时生成多张图像，不要顺序等待。先批量提交，再批量轮询：

# batch_generation.py — 并发提交多个任务
import time
import requests
from config import FLUX_11_PRO_ENDPOINT, RESULT_ENDPOINT, HEADERS

def batch_generate(prompts: list[str]) -> list[str]:
    """
    并发提交多个生成任务，然后统一轮询。
    比顺序生成快约 N 倍（N = 任务数量）。
    """
    # Phase 1: 批量提交所有任务
    task_ids = []
    for prompt in prompts:
        resp = requests.post(
            FLUX_11_PRO_ENDPOINT,
            headers=HEADERS,
            json={"prompt": prompt, "width": 1024, "height": 1024},
            timeout=30
        )
        resp.raise_for_status()
        task_ids.append(resp.json()["id"])
        time.sleep(0.2)  # 短暂间隔，避免触发突发速率限制

    print(f"已提交 {len(task_ids)} 个任务")

    # Phase 2: 统一轮询，收集完成的任务
    results = {}
    pending = set(task_ids)

    while pending:
        time.sleep(2.0)  # 批量轮询时间隔稍长，减少 API 调用次数

        for task_id in list(pending):
            resp = requests.get(
                RESULT_ENDPOINT,
                headers=HEADERS,
                params={"id": task_id},
                timeout=10
            )
            result = resp.json()
            if result.get("status") == "Ready":
                results[task_id] = result["result"]["sample"]
                pending.remove(task_id)
                print(f"完成: {task_id[:8]}... ({len(results)}/{len(task_ids)})")

    # 按原始顺序返回 URL
    return [results[tid] for tid in task_ids]


if __name__ == "__main__":
    prompts = [
        "A red apple on a white background, studio lighting",
        "A blue ocean wave at sunset, long exposure photography",
        "A dense green forest with morning fog, aerial view",
    ]
    urls = batch_generate(prompts)
    for i, url in enumerate(urls):
        print(f"图像 {i+1}: {url}")

---

5. API 参数参考表 {#api-parameters}

以下参数适用于 POST /v1/flux-pro-1.1 端点：

参数名	类型	默认值	有效范围	影响内容
prompt	string	—	最大约 10,000 字符	图像内容描述，必填
width	integer	1024	256–1440，须为 32 的倍数	输出图像宽度（像素）
height	integer	1024	256–1440，须为 32 的倍数	输出图像高度（像素）
prompt_upsampling	boolean	false	true / false	开启时 API 自动扩展 prompt，增加细节随机性
seed	integer	null（随机）	0–4294967295	固定 seed 可复现完全相同的图像
safety_tolerance	integer	2	0–6	内容过滤严格程度，0 最严，6 最宽松
output_format	string	"jpeg"	"jpeg" / "png"	输出文件格式；PNG 无损但文件更大

关于分辨率的说明：

• 总像素数（width × height）应保持在约 1 megapixel（1024×1024）附近以获得最佳质量
• 极端宽高比（如 256×1440）会明显影响构图质量
• 常用宽高比参考：1024×1024（1:1）、1344×768（16:9）、768×1344（9:16）

---

6. 错误处理 {#error-handling}

BFL API 使用标准 HTTP 状态码，以下是你在生产环境中最可能遇到的错误：

HTTP 状态码	含义	常见原因	解决方案
401 Unauthorized	API key 无效	key 拼写错误或已过期	检查 .env 中的 key，重新生成
402 Payment Required	账户余额不足	免费额度耗尽	在 BFL 控制台充值
422 Unprocessable Entity	参数验证失败	width 不是 32 的倍数等	检查参数范围，参考上方参数表
429 Too Many Requests	触发速率限制	请求频率过高	添加退避重试，降低并发数
500 Internal Server Error	BFL 服务端错误	偶发性故障	等待后重试，建议指数退避
task.status = "Content Moderated"	内容审核拦截	Prompt 包含违禁内容	修改 prompt；不要尝试绕过
task.status = "Error"	生成任务失败	不确定原因	记录 task_id，用新参数重试

带退避重试的完整错误处理：

# retry_handler.py — 生产级重试逻辑
import time
import requests
from config import FLUX_11_PRO_ENDPOINT, HEADERS


def submit_with_retry(payload: dict, max_retries: int = 3) -> str:
    """
    提交任务，遇到可重试错误时使用指数退避策略。
    不可重试的错误（401、402、422）直接抛出，不浪费重试次数。
    """
    # 这些状态码代表客户端错误，重试没有意义
    NON_RETRYABLE = {401, 402, 403, 422}

    for attempt in range(max_retries):
        try:
            resp = requests.post(
                FLUX_11_PRO_ENDPOINT,
                headers=HEADERS,
                json=payload,
                timeout=30
            )

            if resp.status_code in NON_RETRYABLE:
                # 直接抛出，不进入重试循环
                resp.raise_for_status()

            if resp.status_code == 429:
                # 速率限制：读取 Retry-After header（如果有）
                retry_after = int(resp.headers.get("Retry-After", 10))
                print(f"速率限制，等待 {retry_after} 秒后重试...")
                time.sleep(retry_after)
                continue  # 不计入 attempt 次数

            if resp.status_code >= 500:
                # 服务端错误：指数退避
                wait = (2 ** attempt) * 2  # 2s, 4s, 8s
                print(f"服务端错误 {resp.status_code}，{wait}s 后重试 (attempt {attempt+1}/{max_retries})")
                time.sleep(wait)
                continue

            resp.raise_for_status()
            return resp.json()["id"]

        except requests.exceptions.Timeout:
            wait = (2 ** attempt) * 2
            print(f"请求超时，{wait}s 后重试...")
            time.sleep(wait)

        except requests.exceptions.ConnectionError as e:
            if attempt == max_retries - 1:
                raise RuntimeError(f"连接失败，已重试 {max_retries} 次: {e}")
            time.sleep(5)

    raise RuntimeError(f"提交任务失败，已重试 {max_retries} 次。")

关于 Content Moderated 的说明：

safety_tolerance 设为 6 并不意味着”无限制”。BFL 有硬性内容政策，某些类别的内容无论参数如何设置都会被拦截。遇到这个状态时不要重试相同的 prompt，而是修改内容。

---

7. 性能与成本 {#performance-cost}

以下数据基于官方文档和社区测试（来源：Black Forest Labs 发布公告，GenAI Arena 评测）：

指标	FLUX 1.1 Pro	FLUX.1 Pro（上代）	DALL-E 3（1024×1024）
平均生成时间	~10 秒	~60 秒	~15–30 秒
速度提升	6×（对比上代）	基准	—
每张图像成本（1024×1024）	~$0.04	~$0.055	$0.04–$0.08
最大分辨率	1440×1440	1440×1440	1024×1024
GenAI Arena 排名	#1（公开评测）	前 5	—
异步 API	是	是	否
支持 seed 复现	是	是	否

成本估算（按月）：

使用量（张/月）	月成本（估算）
100 张	~$4
1,000 张	~$40
10,000 张	~$400
50,000 张	~$2,000

何时不应该使用 FLUX 1.1 Pro：

• 你需要实时（< 3 秒）响应的场景（异步 API 不适合实时展示）
• 你需要图像编辑（inpainting / outpainting）功能——FLUX 1.1 Pro 仅支持文生图，不支持图生图
• 你的月预算低于 $20 且需要商业使用——免费额度约 $5，用完即付费
• 你需要多语言 prompt——测试显示英文 prompt 效果显著优于中文 prompt

---

8. 结论 {#conclusion}

本文覆盖了从零到生产可用的完整 FLUX 1.1 Pro Python 集成：环境配置、异步轮询逻辑、参数控制、退避重试和批量并发。上方所有代码块均可直接运行，你只需要替换 BFL_API_KEY。在实际部署时，把 FluxProGenerator 包装成一个单例服务实例，复用 requests.Session() 而不是每次创建新连接，可以将网络开销再降低约 15–20%。

---

数据来源：Black Forest Labs 官方 API 文档、GenAI Arena 公开排行榜（2024 Q4）、Stackademic 技术博客《The Ultimate Guide to Generating Images via API》。

提示： 如果你需要在同一个项目中使用多个 AI 模型，AtlasCloud 提供统一 API 接入 300+ 模型（Kling、Flux、Seedance、Claude、GPT 等），一个 key 全部搞定。新用户首次充值享 25% 赠送（最高 $100）。