大規模に画像生成を自動化する方法

複数の異なるAPIを管理せずに大規模な画像生成を自動化する最も簡単な方法は、ワークフローとモデル提供元を分離することです。すべての画像リクエストを1つのキューに入れ、各ジョブを現在のモデルIDへルーティングし、互換のあるリクエストは1つの CometAPI キーと OpenAI 互換のベースURL https://api.cometapi.com/v1. 経由で送信します。

このチュートリアルでは、そのパイプラインを Python で構築します。JSON Lines キューから製品・広告・コンテンツの各ジョブを受け取り、モデルを選択し、同時実行数を制限し、一時的な失敗をリトライし、URL または base64 のいずれかの結果を保存し、ジョブごとの使用量と推定コストを記録します。この記事をコードリファレンス化しないために、例では本番で必須の要素をコンパクトなブロックにまとめ、ワークフローを簡単にテストできるようにしています。

統一イメージAPIが一括生成を簡素化する方法

最終的に、ワークフローは次のようになります。

jobs.jsonl → bounded worker pool → CometAPI /v1/images/generations → object storage → manifest.jsonl

キューとストレージ層はあなたの環境のままです。画像モデルを切り替える際は、認証方式や主要なリクエスト経路ではなく、変更するのは model の値です。これが統一イメージAPIの実利的な利点です。モデル選択が1つのパイプライン内のルーティング判断となり、プロバイダごとに個別の統合を行う必要がなくなります。

画像生成を自動化するには何が必要ですか？

Python 3.10 以降、requests パッケージ、CometAPI キー、書き込み可能な出力先、そして少なくとも1つの現行イメージモデルIDが必要です。

唯一の依存関係をインストールします:

pip install requests

キーはサーバー側に設定し、ブラウザコードやリポジトリには決して含めないでください:

export COMETAPI_KEY="your-key"

ベースURLは https://api.cometapi.com/v1 で、互換のあるテキストから画像のジョブは POST /images/generations を使用します。デプロイ前に live model catalog で各モデルを検証してください。カタログは、現行ID、サポートされるエンドポイント、機能、価格メタデータを、認可ヘッダーなしで返します。

2026年8月20日現在、ライブカタログには次の2つの有用なルートが記載されています:

この表は出発点であり、両モデルが同等の機能を持つという主張ではありません。サイズ、品質、フォーマット、参照画像サポート、応答の挙動はモデルごとに異なります。オプションパラメータを渡す前に、モデルのレコードとリンクされたドキュメントを確認してください。

Python でバッチ画像生成ワークフローを構築する方法

1. 各ジョブに永続的なIDを付与する

1行につき1つのJSONオブジェクトを使用すれば、キュー、データベースエクスポート、スプレッドシートのジョブが同じワーカーに供給できます:

{"id":"sku-1001","kind":"product","prompt":"Studio product photo of a ceramic coffee dripper on a warm neutral background"}
{"id":"campaign-204","kind":"ad","prompt":"Editorial summer travel image, vivid natural light, wide composition, no text"}
{"id":"blog-088","kind":"content","prompt":"Minimal illustration of a developer automating a creative workflow, no text"}

ID は出力ファイル名およびマニフェストのキーになります。本番では、これを冪等性キーとして使用し、再処理の際は既に成功済みとマークされたIDをスキップします。

2. ジョブタイプでルーティングし、ライブカタログで検証する

この例では、製品向けの作業を gpt-image-2 に、広告やコンテンツ向けの作業を doubao-seedream-4-5-251128 にマッピングします。ジョブは自身の model フィールドでこの選択を上書きできます。起動時にワーカーは公開カタログをダウンロードし、記載がなくなったIDは拒否します。

これは、アプリ全体にプロバイダ固有のSDKをハードコーディングするより安全です。自分のプロンプトに対して品質・レイテンシ・価格を評価した後、1つのマッピングを変更するだけでルートを切り替えられます。

3. バッチ全体を起動するのではなく同時実行数を制限する

ワーカーは4つの同時リクエストから開始します。この数は保守的なアプリ設定であり、普遍的なサービス上限ではありません。自分のアカウントでレイテンシと 429 応答を測定し、意図的に MAX_WORKERS を上げ下げしてください。

408、429、および 5xx 応答のみを指数バックオフとジッターでリトライします。認証エラー、無効なモデルID、未対応パラメータは即時に失敗させます。同じ不正なリクエストをリトライしても遅延を増やすだけだからです。

4. 保存前に結果を正規化する

画像モデルは常に同じコンテナを返すとは限りません。ドキュメント化された GPT Image の応答は data[0].b64_json を含み、他の互換モデルは data[0].url を返す場合があります。ワーカーは両方を扱い、一時ファイルに画像を書き込み、ダウンロードまたはデコードが成功した後にのみリネームします。

本番では、ローカルの output/ ディレクトリを S3、R2、GCS などのオブジェクトストアに置き換えてください。プロバイダホストのURLを、保持ポリシーが明示的にそう述べていない限り恒久的なストレージとして扱わないでください。

5. 使用量、試行回数、推定コストを記録する

各結果は、ジョブID、モデル、保存パス、ステータス、およびライブカタログが十分な価格データを提供する場合は推定USDコストを含むコンパクトなマニフェスト行になります。失敗したジョブは、バッチから消えるのではなくエラーを保持します。

バッチ画像生成のための完全な Python スクリプト

以下を batch_image_pipeline.py として保存し、同じ場所に jobs.jsonl を置き、python3 batch_image_pipeline.py を実行します。

import base64, json, os, random, time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import requests

BASE_URL = "https://api.cometapi.com/v1"
KEY = os.environ["COMETAPI_KEY"]
WORKERS = int(os.getenv("MAX_WORKERS", "4"))
OUT = Path("output")
ROUTES = {
    "product": "gpt-image-2",
    "ad": "doubao-seedream-4-5-251128",
    "content": "doubao-seedream-4-5-251128",
}

catalog = requests.get("https://api.cometapi.com/api/models", timeout=30)
catalog.raise_for_status()
CATALOG = {model["id"]: model for model in catalog.json()["data"]}

def generate(job):
    model = job.get("model", ROUTES[job["kind"]])
    if model not in CATALOG:
        raise ValueError(f"Unknown model: {model}")
    payload = {"model": model, "prompt": job["prompt"], "n": 1}
    if model == "gpt-image-2":
        payload.update(quality="low", size="1024x1024", output_format="jpeg")

    for attempt in range(4):
        response = requests.post(
            f"{BASE_URL}/images/generations",
            headers={"Authorization": f"Bearer {KEY}"},
            json=payload,
            timeout=180,
        )
        if response.status_code not in {408, 429} and response.status_code < 500:
            break
        time.sleep(2**attempt + random.random())
    response.raise_for_status()
    body = response.json()
    item = body["data"][0]

    if item.get("b64_json"):
        data = base64.b64decode(item["b64_json"])
        extension = body.get("output_format", "png")
    else:
        download = requests.get(item["url"], timeout=120)
        download.raise_for_status()
        data = download.content
        extension = {"image/png": "png", "image/webp": "webp"}.get(
            download.headers.get("content-type"), "jpg"
        )
    path = OUT / f"{job['id']}.{extension}"
    path.write_bytes(data)

    price, usage = CATALOG[model].get("pricing") or {}, body.get("usage", {})
    cost = price.get("per_request")
    if cost is None and price.get("input") is not None:
        cost = (usage.get("input_tokens", 0) * price["input"] +
                usage.get("output_tokens", 0) * price["output"]) / 1_000_000
    return {"id": job["id"], "model": model, "path": str(path),
            "estimated_usd": cost * price.get("ratio", 1) if cost is not None else None}

def safe_generate(job):
    try:
        return {"status": "success", **generate(job)}
    except Exception as error:
        return {"id": job["id"], "status": "failed", "error": str(error)}

OUT.mkdir(exist_ok=True)
jobs = [json.loads(line) for line in Path("jobs.jsonl").read_text().splitlines() if line]
with ThreadPoolExecutor(max_workers=WORKERS) as pool:
    results = list(pool.map(safe_generate, jobs))
with (OUT / "manifest.jsonl").open("w") as manifest:
    manifest.writelines(json.dumps(result) + "\n" for result in results)

このスクリプトは実行時に最新のカタログを使用します。2つのフォールバックマッピングは、2026年8月20日に検証された例です。別の日付でコードを公開またはデプロイする前に必ず再確認してください。

バッチ画像生成ワークフローのテスト方法

まずはジョブ1件・ワーカー1つから始めます:

MAX_WORKERS=1 python3 batch_image_pipeline.py

成功した GPT Image の応答は次の構造になります:

{
  "created": 1776841943,
  "output_format": "jpeg",
  "quality": "low",
  "size": "1024x1024",
  "usage": {
    "input_tokens": 16,
    "output_tokens": 208,
    "total_tokens": 224
  },
  "data": [{"b64_json": "<base64-image-data>"}]
}

ワーカーは画像をデコードし、output/<job-id>.jpeg に書き出し、output/manifest.jsonl に成功行を追加します。モデルがURLを返した場合は、ワーカーがそれをダウンロードし、同じマニフェスト形式でローカルパスを保存します。

コードはローカルで構文チェック済みです。実際の生成呼び出しには CometAPI キーが必要なので、同時実行数を増やす前に1ジョブのスモークテストを実行してください。

バッチ画像生成の費用はいくらかかりますか？

価格はモデルレートが変動するため、タイムスタンプを付ける必要があります。2026年8月20日現在、live CometAPI model catalog には次の基本価格フィールドと 0.8 の課金比率が返されました。

• gpt-image-2: 入力100万トークンあたり $5、出力100万トークンあたり $30。記載の比率を適用すると、有効レートはそれぞれ $4 と $24/百万トークン。
• doubao-seedream-4-5-251128: リクエストあたり $0.04。記載の比率を適用すると、$0.032/リクエスト。

CometAPI pricing guide は、公式価格のあるモデルに対するトークン課金と、リクエスト単価のモデルに対するコールベース課金を説明しています。スクリプトは実行時にカタログを読み、同じルールを使用します。

token cost = ratio × (input tokens × input rate + output tokens × output rate) / 1,000,000

request cost = ratio × per-request price

たとえば、上のドキュメント化された GPT Image 応答は、入力16トークン・出力208トークンを報告しています。8月20日のカタログ値を用いると、その例の推定額は約 $0.005056 になります。実際の合計は、モデル、品質、サイズ、プロンプト、リトライ、応答の usage によって変動します。固定の画像単価ではなく、API応答とアカウントの使用状況ダッシュボードを請求記録として扱ってください。

失敗した作業にも予算を割り当ててください。未確認のタイムアウト後のリトライは、2つ目の課金対象結果を生む可能性がありますし、技術的に成功した画像が審査に通らなくてもコストは発生します。APIコストと合格率の両方を追跡しましょう。

effective cost per accepted image = total batch spend / approved images

一般的な画像生成APIのエラーと対処方法

すべてのエラーをリトライしないでください。恒久的な 400 リクエストは無効のままですし、無制限の 429 リトライループはトラフィックスパイクをバックログに変えてしまいます。

大規模な本番画像生成のベストプラクティス

複数ワーカーを扱う場合は、JSON Lines から耐久性のあるキューへ移行します。可視性タイムアウトは最大生成時間より長く設定し、画像とマニフェストの保存後にのみジョブをアックし、試行し尽くしたジョブはレビュー用にデッドレタキューへ送ります。

任意の制御はモデル固有の設定に保持します。共有ペイロードには model、prompt、n: 1 のような共通フィールドのみを含め、quality、size、output_format は選択したモデルのドキュメントで確認できたときにのみ追加します。フォールバックルーティングを追加するなら、同じタスクをサポートするモデルを選び、そのモデル向けにペイロードを組み直してください。プロバイダ固有のオプションを盲目的に再送しないこと。

APIキーはシークレットマネージャに保存し、プロンプト入力を制限し、方針に従って生成アセットをスキャンし、プロバイダのURLを長期のプロダクト記録に残さないでください。ジョブID、モデルID、レイテンシ、試行回数、使用量、ストレージパス、審査結果、カタログのスナップショット日付をログに残します。これらのフィールドにより、見かけの単価ではなく、承認画像あたりのコストでモデルを比較できます。

最後に、予算のガードレールを設定します。最大バッチサイズ、ジョブごとのリトライ上限、1日の支出アラート、承認率が低下した際の停止条件などです。質の低いプロンプトを速く回すことは最適化ではありません。

大規模な画像生成自動化に関するFAQ

複数のAPIを管理せずに、大規模な画像生成を自動化する最も簡単な方法は何ですか？

1つのキューとストレージワークフローを使用し、互換のある画像リクエストを1つの CometAPI キーと https://api.cometapi.com/v1/images/generations. 経由で送ります。モデルIDはルーティング層で変更し、認証やプロバイダSDKを個別に維持しないでください。

1つのリクエストで複数の画像モデルに同時生成させられますか？

この例ではジョブあたり1モデルです。ファンアウトはアプリケーションのワークフローです。IDとモデル値を変えたジョブを複製し、保存した出力を比較してください。これにより、コストと審査ステータスを各モデルにひも付けたままにできます。

同時実行数はいくつにすべきですか？

すべてのアカウントとモデルに共通の数値はありません。まずは4ワーカー程度の小さなプールから始め、レイテンシと 429 応答を監視し、エビデンスに基づいてチューニングしてください。

返されたURLと画像そのもの、どちらを保存すべきですか？

画像そのものを自前のオブジェクトストレージに保存してください。返却URLは一時的な場合があり、GPT Image モデルはURLではなく base64 コンテンツを返すことがあります。

最も安いモデルはどう選べばいいですか？

呼び出し単価ではなく、承認画像あたりのコストを計算してください。トークンまたはリクエスト課金、リトライ、失敗したダウンロード、却下されたアセット、後処理、人手によるレビューを含めます。公開・デプロイ当日に live model catalog を再確認してください。

エンドポイントと応答フォーマットはどこで確認すべきですか？

CometAPI Quick Start、model catalog documentation、image generation reference、pricing guide を参照してください。