Antigravity API の Python クライアントをローカルで叩いていたときは問題なく動いていたのに、本番にデプロイした途端にレート制限でアプリが落ちた、という経験はないでしょうか。私も最初に個人プロダクトで Antigravity を組み込んだとき、「SDK のサンプルコードそのままで十分だろう」と油断してしまい、深夜のアクセス集中でタイムアウトが連鎖し、結局ロールバックする羽目になりました。
Antigravity の Python SDK はとても優秀で、client.models.generate_content() を呼ぶだけで生成タスクが走ります。ただ、この「呼ぶだけ」で本番に出すのは、シートベルトをせずに高速道路に入るようなものです。私が複数のプロダクトで実際に使って効果を確認している、最小構成の「3つの保護装置」を順番にご紹介します。
なぜ素の SDK コールは本番で崩れるのか
Antigravity API の呼び出しが失敗する理由は、大きく3つに分けられます。1つ目は一時的なエラーで、レート制限（HTTP 429）、サービス側の瞬間的な 5xx、ネットワークの瞬断などが該当します。これらは数秒待って再送すれば成功することがほとんどです。2つ目は長時間ブロックで、プロンプトが長すぎる・モデルが重い入力に対して応答に時間がかかるケースです。タイムアウト設定がなければ、アプリ側のリクエストが延々と待たされます。3つ目は継続的な障害で、API 側の大規模障害や認証設定ミスのように、再送しても直らない失敗です。
SDK のデフォルト動作は「1回呼んでエラーならそのまま例外を上げる」ですから、上記のどの種類にも無防備です。これを解消するのが、以下の3つの保護装置です。
- リトライ（一時的なエラーから自動復旧する）
- タイムアウト（長時間ブロックを打ち切る）
- サーキットブレーカー（継続的な障害で API を叩き続けない）
保護装置①: 指数バックオフ + ジッターのリトライ
まずは tenacity を使ったリトライから入れます。単純な「3秒待って再送」ではなく、指数バックオフ + ジッター（ランダムずらし）を使うのがポイントです。ジッターがないと、同じ時刻に障害を踏んだ複数のプロセスが揃って再送し、API 側の復旧を遅らせる「シンクロナイズドリトライ」を起こします。
# retry.py — Antigravity API の呼び出しにリトライを被せる
from google import genai
from google.genai import errors as genai_errors
from tenacity import (
retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
)
client = genai.Client() # API キーは環境変数 GOOGLE_API_KEY から
# 一時的なエラーだけリトライ対象にする(認証エラーなど恒久的なものは即時失敗させる)
TRANSIENT = (
genai_errors.ServerError, # 5xx
genai_errors.ResourceExhausted, # 429 レート制限
TimeoutError,
)
@retry(
retry=retry_if_exception_type(TRANSIENT),
stop=stop_after_attempt(4), # 最大 4 回試行
wait=wait_exponential_jitter(initial=1, max=30), # 1s→2s→4s→8s(+jitter)
reraise=True,
)
def generate_with_retry(prompt: str) -> str:
resp = client.models.generate_content(
model="gemini-2.5-flash",
contents=prompt,
)
return resp.text
if __name__ == "__main__":
print(generate_with_retry("Antigravity のリトライ戦略を一文で説明して"))期待する動作は、一時的な 429 や 5xx に当たった場合に、最大 4 回まで指数バックオフで再試行し、それでも駄目なら元の例外を呼び出し側に投げるというものです。認証エラー（PermissionDenied 等）は TRANSIENT に含めないため、即座に例外が上がって、間違った API キーで延々と叩き続ける事故を防げます。
「全部リトライ対象に入れておけばいい」と思いがちですが、私の経験では恒久エラーはむしろ即時失敗させたほうが安全です。再送して直らないものを4回試すだけ、エンドユーザーを待たせることになります。
保護装置②: タイムアウトで長時間ブロックを打ち切る
次に、1回のリクエストに対するタイムアウトを入れます。SDK には request_options でタイムアウトを渡せるため、まずそこを設定します。それだけでは不十分なケース（SDK の内部再試行などで時間が伸びる場合）に備え、呼び出し側でも asyncio.wait_for で上限を被せる多層防御にします。
# timeout.py — タイムアウトを多層で被せる
import asyncio
from google import genai
from google.genai import types
client = genai.Client()
async def generate_with_timeout(prompt: str, deadline_sec: float = 20.0) -> str:
"""SDKタイムアウト + asyncioタイムアウトの二重防御"""
async def _call() -> str:
resp = await client.aio.models.generate_content(
model="gemini-2.5-flash",
contents=prompt,
# SDK レベルの HTTP タイムアウト
config=types.GenerateContentConfig(
http_options=types.HttpOptions(timeout=15_000), # ミリ秒
),
)
return resp.text
# アプリ側の最終タイムアウト(SDK 側を少し超えた値に)
return await asyncio.wait_for(_call(), timeout=deadline_sec)
async def main():
try:
text = await generate_with_timeout("2026年のAIトレンドを3行で")
print(text)
except asyncio.TimeoutError:
print("⏱ タイムアウト: フォールバック応答を返す")
asyncio.run(main())期待する動作は、15 秒で SDK 内部のタイムアウトが発火し、それでも抜けなければ 20 秒で asyncio 側が TimeoutError を投げるという二段構えです。呼び出し側でキャッチして、キャッシュ済みの古い応答や固定のメッセージを返せば、ユーザー体験を大きく崩さずに済みます。
本番ではタイムアウトを長めに取るほうが親切と思いがちですが、逆に短めにしてフォールバックを返すほうが、同時接続を枯渇させず、ほかのユーザーも救えます。私は体感時間を基準に「対話系は 10〜20 秒」「バッチは 60 秒」を目安にしています。
保護装置③: サーキットブレーカーで障害連鎖を止める
3つ目は pybreaker によるサーキットブレーカーです。API が長時間ダウンしているときに、リトライを繰り返すとこちらのサーバーも無駄にリソースを食います。一定回数連続で失敗したら**「今は API が死んでいる」と見なして即座に失敗させ**、少し待ってから少しずつ復帰確認をする、という制御が必要です。
# breaker.py — リトライ + サーキットブレーカー
import pybreaker
from google import genai
from google.genai import errors as genai_errors
from tenacity import retry, stop_after_attempt, wait_exponential_jitter
client = genai.Client()
# 5 回連続失敗で OPEN、60 秒後に HALF-OPEN へ
breaker = pybreaker.CircuitBreaker(
fail_max=5,
reset_timeout=60,
exclude=[genai_errors.PermissionDenied], # 認証エラーは回数に数えない
)
@breaker
@retry(stop=stop_after_attempt(3),
wait=wait_exponential_jitter(initial=1, max=10),
reraise=True)
def generate_safe(prompt: str) -> str:
resp = client.models.generate_content(
model="gemini-2.5-flash",
contents=prompt,
)
return resp.text
# 呼び出し側のフォールバック
def answer(prompt: str) -> str:
try:
return generate_safe(prompt)
except pybreaker.CircuitBreakerError:
return "ただいま AI サービスが混雑しています。少し時間をおいて再度お試しください。"
except Exception as e:
return f"一時的なエラーが発生しました ({type(e).__name__})"期待する動作は、5 回連続で失敗するとブレーカーが OPEN になり、以降 60 秒間は CircuitBreakerError が即時スローされる（API を叩きに行かない）、60 秒後は HALF-OPEN 状態で 1 回だけ試行し、成功すれば CLOSED に戻るというものです。これがないと、API 側の障害中でも毎リクエストで無駄に 30 秒待たされることになります。
なお、サーバーレス環境（Cloud Run / Vercel など）ではブレーカー状態がインスタンス間で共有されない点に注意してください。本当に堅牢にするなら、Redis などで状態を共有する拡張版（pybreaker.CircuitRedisStorage）を使います。
実際の障害タイムラインで見る、3つの連携動作
具体的にどう効くのかを、1つのシナリオで追いかけてみます。深夜 11:47 に Google 側の特定リージョンで不具合が入り、Antigravity 呼び出しが 5xx を返し始めたとします。
- 最初のリクエストは上流が遅いだけなので、クリーンなエラーではなくタイムアウトになります。
asyncio.wait_forが拾ってTimeoutErrorとして上に流します。 tenacityはこれを一時エラーと判定し、約 1 秒のディレイ（＋ジッター）で次の試行を予約します。隣のプロセスは偶然 1.3 秒を引いて、ふたつの試行がシンクロしません。- 2 回試行しても失敗したので、呼び出し側の
try/exceptがキャッチして、ユーザーにはすぐフォールバックの文字列が返ります。並行してpybreakerは失敗回数を数えています。 - 5 回連続の失敗に達するとブレーカーが OPEN になります。以降のリクエストはネットワークを叩かず、ミリ秒でフォールバックを返すだけになります。アプリの CPU もフラットなまま。
- 0:03 に Google 側が復旧すると、次の自然な試行タイミングで 1 回だけプローブが飛びます。成功するとブレーカーは CLOSED に戻り、人間の介入なしでトラフィックが正常化します。
この仕組みが無ければ、同じ障害は「16 分間の 30 秒スピナー」と「ポケベルの嵐」と「長い手動ウォームアップ」になっていたはずです。入れておけば、たいてい気づかずに寝過ごせます。
可観測性 — 見えない安全装置は安全装置ではない
「入れたつもり」をなくすには、観測ポイントを同時に用意します。私が必ずセットで入れるのは 2 つです。
- リトライのたびに構造化ログ:
tenacityにはbefore_sleep=before_sleep_log(logger, logging.WARNING)が用意されています。プロンプトハッシュと試行回数を一緒に出しておくと、「リトライ嵐」を後から grep で特定できます。 - ブレーカー状態のメトリクス化:
current_stateを Prometheus ゲージ（もしくは任意の時系列 DB）に書き出します。「OPEN が 5 分以上続いた」でアラートを組めば、通常のトラフィック揺れでは鳴らず、本物の障害だけが通知されます。
私のルールは「レジリエンスコードを書いたら、ログとメトリクスを同じ Pull Request で出す」です。そうでないと「効いている気がする」だけで本番に出してしまい、後で痛い目にあうからです。
3つを組み合わせた最小構成
保護装置は個別に使ってもよいのですが、本番では**外から順に「ブレーカー → リトライ → タイムアウト」**の順に重ねるのが定番です。ブレーカーが開いているときはリトライすらせず、開いていない時だけリトライし、その中で各試行にタイムアウトを被せる、という階層構造です。
- 一番外: サーキットブレーカー（全体障害の検出）
- 真ん中: リトライ + 指数バックオフ（一時エラーの吸収）
- 一番内: SDK のタイムアウト + asyncio タイムアウト（個別呼び出しの暴走防止）
Antigravity API を組み込んだサービスを運用すると、「負荷ピークで 429 が散発する」「深夜の Google 側メンテで 5xx が連発する」「たまに応答に 40 秒かかるリクエストが混ざる」といった現象に必ず出会います。上記3つを入れておけば、そのどれもがユーザー影響ゼロか、軽微な体感変化で吸収できるようになります。
より体系的なエラーハンドリング設計は Antigravity Python SDK 本番運用マスターガイド で、具体的なエラー原因の切り分けは Antigravity Python SDK エラー診断と修正 で扱っています。API の基礎から入りたい方は Antigravity Python × Gemini API クイックスタート から始めてみてください。
Python での本番運用設計
次の一歩
今日この記事を読み終えたら、手元の Antigravity 呼び出しコードを1つだけ選び、まず tenacity のリトライデコレータを付けてみてください。5行足すだけで、夜間のレート制限エラーがログから劇的に減ります。慣れてきたら、タイムアウト → サーキットブレーカーと段階的に積み重ねていくのが、挫折しない導入順序です。