Gemini API × Google Cloud 本番エラーを層で切り分ける診断手順

# 必要なAPI一覧を確認
gcloud services list --enabled --filter="name:(aiplatform OR ml OR storage)"
 
# 不足しているAPIを有効化
gcloud services enable \
  aiplatform.googleapis.com \
  ml.googleapis.com \
  storage.googleapis.com \
  --project=YOUR_PROJECT_ID

from googleapiclient.discovery import build
from google.auth import default
 
credentials, project = default()
service = build('serviceusage', 'v1', credentials=credentials)
 
response = service.services().get(
    name=f"projects/{project}/services/aiplatform.googleapis.com"
).execute()
 
print(f"API状態: {response.get('state')}")  # "ENABLED" であることを確認

# プロジェクトのクォータを確認
gcloud compute project-info describe --project=YOUR_PROJECT_ID
 
# または Cloud Console で確認
# IAM & Admin → Quotas → フィルタ: "aiplatform"

import time
import random
import google.api_core.exceptions
import vertexai
from vertexai.generative_models import GenerativeModel
 
vertexai.init(project="YOUR_PROJECT_ID", location="us-central1")
model = GenerativeModel("gemini-2.0-flash-001")
 
def generate_with_retry(prompt: str, max_retries: int = 5) -> str:
    """クォータエラーへの対応込みリトライ実装"""
    for attempt in range(max_retries):
        try:
            response = model.generate_content(prompt)
            return response.text
 
        except google.api_core.exceptions.ResourceExhausted as e:
            if attempt >= max_retries - 1:
                raise
 
            # Retry-Info ヘッダーを確認（ある場合は優先）
            retry_delay = None
            if hasattr(e, 'metadata'):
                for key, value in e.metadata:
                    if key == 'retry-info':
                        # メタデータからリトライ遅延を取得
                        retry_delay = float(value)
                        break
 
            if retry_delay is None:
                # エクスポネンシャルバックオフ + ジッター
                retry_delay = min(2 ** attempt + random.uniform(0, 1), 60)
 
            print(f"クォータ超過: {retry_delay:.1f}秒後にリトライ (試行 {attempt + 1}/{max_retries})")
            time.sleep(retry_delay)
 
        except google.api_core.exceptions.DeadlineExceeded:
            if attempt >= max_retries - 1:
                raise
            time.sleep(2 ** attempt)
 
    return ""

from vertexai.generative_models import GenerativeModel
 
model = GenerativeModel("gemini-2.0-flash-001")
 
def estimate_tokens(text: str) -> int:
    """大まかなトークン数を事前計算（1トークン≒4文字の目安）"""
    return len(text) // 4
 
def batch_process_with_tpm_control(
    texts: list[str],
    tpm_limit: int = 100000
) -> list[str]:
    """TPM制限を考慮したバッチ処理"""
    results = []
    current_minute_tokens = 0
    minute_start = time.time()
 
    for text in texts:
        estimated_tokens = estimate_tokens(text)
 
        # 1分あたりのトークン数をチェック
        if current_minute_tokens + estimated_tokens > tpm_limit * 0.9:  # 90%で調整
            elapsed = time.time() - minute_start
            if elapsed < 60:
                wait_time = 60 - elapsed + 1  # 次の分まで待機
                print(f"TPM制限に近い: {wait_time:.1f}秒待機")
                time.sleep(wait_time)
            current_minute_tokens = 0
            minute_start = time.time()
 
        response = generate_with_retry(text)
        results.append(response)
        current_minute_tokens += estimated_tokens
 
    return results

# VPC-SC アクセスレポートでブロックされたリクエストを確認
gcloud access-context-manager perimeters list \
  --policy=YOUR_POLICY_NUMBER
 
# Cloud Audit Logs で詳細を確認
gcloud logging read \
  'protoPayload.@type="type.googleapis.com/google.cloud.audit.AuditLog"
   AND protoPayload.status.code=7
   AND resource.type="audited_resource"' \
  --freshness=1h \
  --format="json" | head -100

# 管理者が実行: VPC-SC ペリメータにサービスを追加
gcloud access-context-manager perimeters update PERIMETER_NAME \
  --add-restricted-services=aiplatform.googleapis.com \
  --policy=POLICY_NUMBER

import vertexai
 
# PSCエンドポイントを使用した初期化
vertexai.init(
    project="YOUR_PROJECT_ID",
    location="us-central1",
    api_endpoint="YOUR_PSC_ENDPOINT.p.googleapis.com"
)

import vertexai
from vertexai.generative_models import GenerativeModel
import google.api_core.exceptions
import logging
 
logger = logging.getLogger(__name__)
 
REGION_PRIORITY = ["us-central1", "us-east4", "europe-west4", "asia-northeast1"]
 
def generate_with_region_failover(
    prompt: str,
    project_id: str
) -> tuple[str, str]:
    """マルチリージョンフェイルオーバー付き生成"""
 
    for region in REGION_PRIORITY:
        try:
            vertexai.init(project=project_id, location=region)
            model = GenerativeModel("gemini-2.0-flash-001")
            response = model.generate_content(prompt)
            return response.text, region
 
        except google.api_core.exceptions.ServiceUnavailable as e:
            logger.warning(f"リージョン {region} が利用不可: {e}")
            continue
 
        except google.api_core.exceptions.DeadlineExceeded as e:
            logger.warning(f"リージョン {region} タイムアウト: {e}")
            continue
 
    raise RuntimeError("すべてのリージョンで生成に失敗しました")
 
# 利用例
result, used_region = generate_with_region_failover(
    "本番環境のフェイルオーバーテスト",
    "YOUR_PROJECT_ID"
)
print(f"リージョン {used_region} で生成成功")

from google.cloud import monitoring_v3
import time
 
def create_gemini_api_alerts(project_id: str):
    """Gemini API の重要アラートを設定"""
    client = monitoring_v3.AlertPolicyServiceClient()
    project_name = f"projects/{project_id}"
 
    # クォータ使用率 80% 超過アラート
    quota_alert = monitoring_v3.AlertPolicy(
        display_name="Gemini API Quota Usage > 80%",
        conditions=[
            monitoring_v3.AlertPolicy.Condition(
                display_name="High quota usage",
                condition_threshold=monitoring_v3.AlertPolicy.Condition.MetricThreshold(
                    filter='resource.type="aiplatform.googleapis.com/Location"'
                           ' AND metric.type="aiplatform.googleapis.com/quota/online_prediction_requests_per_base_model/usage"',
                    comparison=monitoring_v3.ComparisonType.COMPARISON_GT,
                    threshold_value=0.8,
                    duration={"seconds": 300},
                )
            )
        ],
        notification_channels=["projects/YOUR_PROJECT_ID/notificationChannels/YOUR_CHANNEL_ID"],
        combiner=monitoring_v3.AlertPolicy.ConditionCombinerType.OR
    )
 
    policy = client.create_alert_policy(
        name=project_name,
        alert_policy=quota_alert
    )
    print(f"アラートポリシー作成: {policy.name}")

# 組織ポリシーのリソースロケーション制約を確認
gcloud resource-manager org-policies describe \
  constraints/gcp.resourceLocations \
  --project=YOUR_PROJECT_ID

import google.generativeai as genai
 
genai.configure(api_key="YOUR_API_KEY")
 
# モデル名をハードコード。停止・改名に弱い
model = genai.GenerativeModel("gemini-3.1-flash-image-preview")
 
def generate(prompt: str):
    try:
        return model.generate_content(prompt).text
    except Exception:
        # 原因の層が分からないまま握りつぶしてしまう
        return None

import logging
import google.generativeai as genai
from google.api_core.exceptions import NotFound
 
genai.configure(api_key="YOUR_API_KEY")
log = logging.getLogger("gemini.preflight")
 
# 非推奨モデル -> 後継モデルのフォールバック表（停止日付き）
DEPRECATIONS = {
    "gemini-3.1-flash-image-preview": ("gemini-3.5-flash", "2026-06-25"),
    "gemini-3-pro-image-preview": ("gemini-3.5-flash", "2026-06-25"),
}
 
def resolve_model(requested: str) -> str:
    """起動時に1回だけ実行。停止・改名を検知して後継へ寄せる。"""
    target = requested
    if requested in DEPRECATIONS:
        successor, shutdown = DEPRECATIONS[requested]
        log.warning("model %s は %s に停止予定。%s へフォールバックします",
                    requested, shutdown, successor)
        target = successor
    # 実在確認：存在しなければ NotFound が飛ぶ
    try:
        genai.get_model(f"models/{target}")
    except NotFound:
        log.error("後継モデル %s も見つかりません。設定を確認してください", target)
        raise
    return target
 
MODEL_NAME = resolve_model("gemini-3.1-flash-image-preview")
model = genai.GenerativeModel(MODEL_NAME)
 
def generate(prompt: str) -> str | None:
    try:
        resp = model.generate_content(prompt)
    except NotFound:
        log.error("実行時にモデルが消失しました。preflight をやり直してください")
        raise
    # 要求と応答のモデルがずれていないか毎回突き合わせる
    used = getattr(resp, "model_version", MODEL_NAME)
    if used != MODEL_NAME:
        log.info("応答モデルが %s に切り替わりました（要求: %s）", used, MODEL_NAME)
    return resp.text