Nano Banana 2の503「Model is Overloaded」対処法：画像生成の再試行を安全に制限する

Nano Banana 2で503「Model is Overloaded」が返ったら、まず画像生成のリクエストに対する503応答を確認し、間隔を空けて、回数と総待ち時間に上限を設けて再試行します。上限を超えたら、その場で送り続けず、処理を保留するかキューに戻します。HTTP 200が返った場合も、画像データを取り出せて初めて画像生成の成功と判断してください。

現在のNano Banana 2のモデルIDはgemini-3.1-flash-imageです。以下はGemini APIのネイティブなgenerateContentを使う画像生成・編集向けの対処法です。Geminiアプリや中継サービスの503を、そのまま同じ原因と決めつけないでください。Googleのモデル情報には画像・テキスト出力とBatch対応が記載されています。

まず、再試行してよい503かを確認する

503は、サービスが一時的な過負荷やメンテナンスなどでリクエストを処理できないことを示すHTTPステータスです。それだけでは、世界中の全ユーザーが影響を受けていることや、特定のGPU不足までは分かりません。中継サーバー自身が503を返す場合もあります。HTTPの503の定義に照らして、まず応答を返したサービスを切り分けます。

ログには、呼び出し先、APIの種類、モデルID、HTTPステータス、エラー本文、発生時刻、経過時間、取得できたリクエストIDを残します。APIキーや入力画像をエラーログに含める必要はありません。PythonのGoogle Gen AI SDKでは、エラー文字列に「503」が含まれるかではなく、errors.APIErrorの整数のcodeで判定できます。SDKのエラー実装を参照してください。

Googleは一時的な429や503についてバックオフとジッターを案内していますが、すべての429が待てば解消するわけではありません。トラブルシューティングと、実際のエラー本文・使用量を併せて確認します。また、Interactions APIのエラー一覧にある文字列のservice_unavailableを、ネイティブなgenerateContentのSDK例へ混ぜないでください。

Pythonで画像生成・編集の再試行に上限を設ける

次の例では、初回を含む最大4回、待機を含む全体の時間枠を90秒とします。これは動作を説明するための設定値で、Googleの推奨値や「90秒で復旧する」という予測ではありません。利用者が待てる時間と実際の生成時間を見て調整してください。

再試行はアプリ側だけが担当します。SDKにはHttpRetryOptions(attempts=1)を指定し、SDK内部と外側の再試行が重ならないようにします。Googleの説明には「SDKが自動で4回再試行する」とありますが、2026年10月6日に確認したPython SDKのmainでは、retry_optionsがNoneなら再試行しない実装です。使用中の公開版とmainが同じとも限りません。既定値を推測せず、設定を明示する理由はここにあります。SDKのHTTPオプションと再試行実装では、attemptsは初回を含み、HttpOptions.timeoutの単位はミリ秒です。

以下は、google-genaiとPillowを利用できる既存のPythonプロジェクトへ組み込む関数です。make_clientには既存のAPIキーを引数として渡し、そのクライアントをgenerate_imageへ渡します。認証設定の変更やキーの切り替えは503対処の前提ではありません。編集する場合だけ、読み取り可能なPNG・JPEG・WebPをedit_pathに指定します。

import asyncio
import io
import math
import random
import time
import uuid
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path

from google import genai
from google.genai import errors, types
from PIL import Image

MODEL = "gemini-3.1-flash-image"


def make_client(api_key: str):
    return genai.Client(
        api_key=api_key,
        http_options=types.HttpOptions(
            retry_options=types.HttpRetryOptions(attempts=1)
        ),
    )


def retry_after_seconds(error):
    headers = getattr(getattr(error, "response", None), "headers", {}) or {}
    value = headers.get("Retry-After") or headers.get("retry-after")
    if not value:
        return 0.0
    value = value.strip()
    if value.isascii() and value.isdigit():
        return float(value)
    try:
        when = parsedate_to_datetime(value)
        if when.tzinfo is None:
            when = when.replace(tzinfo=timezone.utc)
        seconds = (when - datetime.now(timezone.utc)).total_seconds()
        return max(0.0, seconds) if math.isfinite(seconds) else 0.0
    except (TypeError, ValueError, OverflowError):
        return 0.0


def save_image(response, folder, job_id):
    for candidate in response.candidates or []:
        content = candidate.content
        for part in (content.parts if content else []) or []:
            data = part.inline_data
            if not data or not data.mime_type or not data.mime_type.startswith("image/"):
                continue
            if not data.data:
                continue
            # SDKのinline_data.dataはバイト列。実際にデコードできるか確認する。
            try:
                with Image.open(io.BytesIO(data.data)) as image:
                    image.load()
                    decoded = image.copy()
            except (OSError, ValueError):
                continue
            folder.mkdir(parents=True, exist_ok=True)
            output = folder / f"{job_id}.png"
            decoded.save(output, format="PNG")
            return str(output)
    return None


async def generate_image(
    client, prompt: str, *, edit_path=None, output_dir="generated",
    max_attempts=4, budget_seconds=90.0,
):
    if max_attempts < 1 or budget_seconds <= 0:
        raise ValueError("試行回数と時間枠は正の値にしてください")
    job_id = uuid.uuid4().hex
    parts = [types.Part.from_text(text=prompt)]
    if edit_path is not None:
        path = Path(edit_path)
        mime = {".png": "image/png", ".jpg": "image/jpeg",
                ".jpeg": "image/jpeg", ".webp": "image/webp"}.get(path.suffix.lower())
        if mime is None:
            raise ValueError("入力はPNG・JPEG・WebPにしてください")
        parts.append(types.Part.from_bytes(data=path.read_bytes(), mime_type=mime))
    contents = [types.Content(role="user", parts=parts)]
    deadline = time.monotonic() + budget_seconds
    attempts = 0

    def result(state, reason, **extra):
        return {"job_id": job_id, "state": state, "reason": reason,
                "attempts": attempts, **extra}

    for attempt in range(max_attempts):
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            return result("deferred", "time_budget")
        attempts += 1
        config = types.GenerateContentConfig(
            response_modalities=["TEXT", "IMAGE"],
            image_config=types.ImageConfig(aspect_ratio="16:9", image_size="1K"),
            http_options=types.HttpOptions(
                timeout=max(1, int(remaining * 1000)),
                retry_options=types.HttpRetryOptions(attempts=1),
            ),
        )
        try:
            response = await asyncio.wait_for(
                client.aio.models.generate_content(
                    model=MODEL, contents=contents, config=config
                ),
                timeout=remaining,
            )
        except errors.APIError as error:
            if error.code != 503:
                raise
            if attempt + 1 == max_attempts:
                return result("deferred", "attempt_limit")
            # フルジッター。Retry-Afterより早く送らない。
            jitter = random.uniform(0.0, min(8.0, 2.0 ** attempt))
            delay = max(jitter, retry_after_seconds(error))
            if delay >= deadline - time.monotonic():
                return result("deferred", "retry_after_or_time_budget")
            await asyncio.sleep(delay)
            continue
        except Exception as error:
            # タイムアウトや切断は、サーバーでの完了状況が分からない。
            return result("uncertain", type(error).__name__)

        output = save_image(response, Path(output_dir), job_id)
        if output is None:
            return result(
                "review", "no_decodable_image",
                finish_reasons=[str(c.finish_reason) for c in response.candidates or []],
                prompt_feedback=str(response.prompt_feedback),
            )
        return result("succeeded", "image_saved", output=output)
    return result("deferred", "attempt_limit")

呼び出し元は、例えばawait generate_image(client, "白い背景の商品写真を生成してください")とします。編集ではedit_path="input.png"と編集指示を渡します。clientは上のmake_clientで作ったものを使い、既存アプリの終了処理でクライアントを閉じてください。Googleの画像生成・編集の説明に沿って、入力画像もテキストも同じ画像対応モデルへ送る構成です。

コードの重要な点は、503以外のAPIエラーを呼び出し元へ返すこと、待機時間も含めた残り時間を毎回確認すること、画像をデコードして保存するまで成功にしないことです。非同期の時間制限は、生成処理をサーバー側で取り消せた証明にはなりません。時間切れになったリクエストはuncertainとして扱います。画像の保存時間はこのネットワーク処理の時間枠とは別なので、巨大な応答を扱うアプリでは保存・デコードにも独立した制限を設けます。

この例は、構造化した疑似エラー、疑似応答、待機を進める仮想時計で分岐を確認した組み込み例です。実際のAPIへの送信、画像生成の所要時間や復旧率の測定は行っていません。利用中のSDKが上記フィールドを持つことも、組み込み前に確認してください。

画像が返ったかを確認し、保留と成功を分ける

HTTP 200やテキストの説明だけでは、画像の納品は完了していません。上の関数は、候補のinline_dataに画像MIMEとデータがあり、Pillowでデコードしてファイルに保存できた場合だけsucceededを返します。保存先を開いて利用できるか、アプリの表示まで続けて確認します。

画像がないときには、終了理由やprompt_feedbackを調べます。安全性による制限、画像を生成しなかった応答、取り出し方の誤りなどを分ける必要があります。「200なのに画像がない」からといって、必ず安全性フィルターが原因だとは限りません。ここで503用のループに入れると、同じ入力の生成を繰り返すことになります。

呼び出し元では、返された状態ごとに処理を分けてください。

job_idはアプリ側で処理を追跡する識別子です。Google側でPOSTの重複生成を防ぐキーではありません。通信が途切れる前にサーバーで生成が完了していれば、再送によって別の画像が生成される可能性があります。HTTPの非冪等リクエストの再試行に関する規定を踏まえ、結果不明の送信を、明確な503応答と分けています。

Retry-Afterが長いときは、待ち続けずキューへ移す

Retry-Afterは秒数、またはHTTPの日付で指定される任意のヘッダーです。上の関数は両形式を読み、ジッターの待機時間と比較して長い方を選びます。例えば残り時間が20秒なのにRetry-After: 120が返ったら、20秒へ切り詰めて早く再送するのではなく、その場の処理を保留します。Retry-Afterの定義に従った扱いです。

キューにはジョブID、入力への安全な参照、モデルID、累計試行回数、次に試せる時刻、ジョブの期限を保存します。ワーカーも無限に取り直すのではなく、累計の送信回数や期限を超えたら要確認へ移してください。キューに戻すたびに「初回から4回」を繰り返すだけでは、全体として無限の再試行になります。

同じジョブを複数ワーカーが同時に送らないよう、取得時に排他的な権利を確保します。生成を待つ利用者が画面を再読み込みした場合も、新しいジョブを作るより、既存ジョブの状態を返す方が重複を抑えられます。キューへの受付と画像の完成は、画面表示でも区別してください。

並列送信と半開状態のテストを制御する

一件ずつの再試行を直しても、全ワーカーが同時に再送すれば送信が集中します。アプリ全体の同時実行数を制限し、超えた分は待機キューへ回します。503が増えたときにワーカーを増やして取り戻そうとすると、かえってリクエストが集中するためです。許容する並列数は、契約上の制限と自分のログから決めます。Nano Banana 2なら一律に何並列が安全、という数値はここでは設定しません。

503が続く場面では、サーキットブレーカーで一定期間の新規送信を止める方法があります。通常時は送信し、障害が続いたら停止状態に移り、待機後には半開状態で一件だけ試します。複数の処理が同時に「テストしてよい」と判断しないよう、その一件の権利はロックや共有ストアで原子的に確保します。

半開状態の一件に、通常の4回再試行を丸ごと許す必要はありません。一回だけ送り、デコード可能な画像を取得できたら徐々に通常送信へ戻します。再び503なら停止状態に戻し、タイムアウトで結果が不明なら回復の成功に数えません。テストの権利は、成功・失敗・キャンセルのどの場合でも解放する設計にします。

一つのプロセス内のロックは、別のサーバーまでは制御できません。分散したワーカーでは共有した停止状態と期限付きの権利が必要です。また、入力エラーや個別ユーザーの権限エラーを、サービス全体の503障害の回数に混ぜないでください。これらは送信集中を抑える設計上の選択で、Googleが定めた固定の障害判定値ではありません。

別モデルとBatchを選ぶ条件

別モデルへ切り替えるなら、まず同じ仕事ができるかを確かめます。画像生成だけでなく、参照画像を使った編集、アスペクト比、解像度、入力画像の取り扱い、料金の上限が変わるかが判断点です。テキスト専用のGeminiモデルへ切り替えて文章が返っても、画像生成の復旧にはなりません。

画像を急いで用意する必要があり、事前に検証した別の画像モデルがあるなら、利用者が許容する品質・費用の範囲でフォールバックを検討します。別モデルを試す場合も、元の処理と合算した時間・送信・費用の予算を設けます。別のプロトコルやVertex AIへ自動で切り替えるには、認証も入力形式も別途整える必要があるため、モデル名を変えるだけの復旧例には含めていません。

即時の画像が不要なら、自前のキューに保留するか、GeminiのBatch APIを使う選択肢があります。Nano Banana 2はBatch対応で、Batchは非同期処理、完了目標は24時間です。ただし、その時間内の完了保証や503を完全に免れる保証ではありません。Batchの公式説明には、失敗・キャンセル・期限切れの状態もあります。

Batchではジョブ名を保存して状態を確認し、結果が遅いという理由だけで同じ一括処理を再投入しないでください。ジョブが成功していても、各リクエストのエラーと画像パーツを確認します。失敗した項目だけを再処理できるようにしておくと、成功済みの画像を重複生成せずに済みます。

よくある疑問

503は何分待てば直りますか？

復旧までの固定時間は、このエラーだけでは判断できません。短時間で回復する場合も、時間枠を超えて続く場合もあるため、再試行に上限を設けます。JSTの夜を避ければ成功する、といった一律の時間帯もここでは案内しません。実際のモデル・呼び出し先別のログがある場合に、その期間の傾向として使ってください。

503が出たら、Google全体がダウンしていますか？

それだけでは分かりません。利用中のエンドポイントとモデル、同じ時間の障害情報を照合します。Gemini APIならGoogle AI Studioの状態ページを確認し、中継サービスならそのサービスの案内も確認します。このページは、現在の世界的な障害の発生を判定するものではありません。

503の失敗は無料なので、何回でも再試行してよいですか？

無料とは一律に断定できません。Googleの課金FAQが明示しているのは、400または500で失敗したリクエストはトークン課金されないがクォータには数える、という範囲です。そこから503や中継サービス、結果不明のタイムアウトまで同じ扱いと広げることはできません。実際の利用量・請求・送信履歴で確認し、再試行や別モデルで新しい生成が発生する可能性も予算に含めます。

有料ティアへ変更すれば503がなくなりますか？

503がなくなる保証はありません。429なら、レート制限やクォータのどの項目に達したかを確認することが先です。確認済み503については、まず上限付きの再試行と送信集中の抑制を行い、ティア変更だけを復旧方法として扱わないでください。

公式ドキュメント

リクエストパラメータ、料金、課金ルールは LaoZhang API の開発者ドキュメントを参照してください：画像生成の課金と失敗処理。

注記：laozhang.ai は当ブログ運営チームの自社サービスです。その料金は中立的な比較ではなく、自社の提示価格です。