Gemini 3.1 Pro × Cloud Run：サーバーレスAI APIの本番構築

# requirements.txt
fastapi==0.115.0
uvicorn[standard]==0.32.0
google-genai==1.14.0
sse-starlette==2.2.1
google-cloud-logging==3.11.0
google-cloud-secret-manager==2.21.0

# main.py
import os
import json
import logging
from contextlib import asynccontextmanager
 
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from sse_starlette.sse import EventSourceResponse
from google import genai
 
# --- ロギング設定 ---
if os.getenv("K_SERVICE"):  # Cloud Run 環境判定
    import google.cloud.logging
    client = google.cloud.logging.Client()
    client.setup_logging()
 
logger = logging.getLogger(__name__)
 
# --- Gemini クライアント初期化（起動時に1回だけ） ---
gemini_client = None
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    """アプリケーション起動時にGeminiクライアントを初期化"""
    global gemini_client
    api_key = os.environ.get("GEMINI_API_KEY")
    if not api_key:
        raise RuntimeError("GEMINI_API_KEY is not set")
    gemini_client = genai.Client(api_key=api_key)
    logger.info("Gemini client initialized successfully")
    yield
    logger.info("Application shutting down")
 
app = FastAPI(title="Gemini AI API", lifespan=lifespan)
 
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 本番では適切に制限する
    allow_methods=["POST"],
    allow_headers=["*"],
)

# main.py（続き）
from pydantic import BaseModel, Field
from typing import Optional
 
class GenerateRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=30000)
    model: str = Field(default="gemini-3.1-pro-preview")
    max_tokens: Optional[int] = Field(default=8192, ge=1, le=65000)
    temperature: Optional[float] = Field(default=0.7, ge=0.0, le=2.0)
    stream: bool = Field(default=True)
 
@app.post("/v1/generate")
async def generate(req: GenerateRequest):
    """Gemini APIを呼び出し、SSEストリーミングでレスポンスを返す"""
    if req.stream:
        return EventSourceResponse(
            stream_generate(req),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "X-Accel-Buffering": "no",  # Nginx バッファリング無効化
            },
        )
    else:
        return await batch_generate(req)
 
async def stream_generate(req: GenerateRequest):
    """SSEストリーミングジェネレーター"""
    try:
        response = gemini_client.models.generate_content_stream(
            model=req.model,
            contents=req.prompt,
            config=genai.types.GenerateContentConfig(
                max_output_tokens=req.max_tokens,
                temperature=req.temperature,
            ),
        )
 
        total_tokens = 0
        for chunk in response:
            if chunk.text:
                total_tokens += len(chunk.text.split())
                yield {
                    "event": "chunk",
                    "data": json.dumps({
                        "text": chunk.text,
                        "done": False,
                    }),
                }
 
        # 完了シグナル
        yield {
            "event": "done",
            "data": json.dumps({
                "text": "",
                "done": True,
                "usage": {"approximate_tokens": total_tokens},
            }),
        }
 
    except Exception as e:
        logger.error(f"Stream generation failed: {e}", exc_info=True)
        yield {
            "event": "error",
            "data": json.dumps({"error": str(e)}),
        }
 
async def batch_generate(req: GenerateRequest):
    """非ストリーミング（バッチ）レスポンス"""
    try:
        response = gemini_client.models.generate_content(
            model=req.model,
            contents=req.prompt,
            config=genai.types.GenerateContentConfig(
                max_output_tokens=req.max_tokens,
                temperature=req.temperature,
            ),
        )
        return {
            "text": response.text,
            "usage": {
                "prompt_tokens": response.usage_metadata.prompt_token_count,
                "output_tokens": response.usage_metadata.candidates_token_count,
            },
        }
    except Exception as e:
        logger.error(f"Batch generation failed: {e}", exc_info=True)
        raise HTTPException(status_code=500, detail=str(e))

# ---- ビルドステージ ----
FROM python:3.12-slim AS builder
 
WORKDIR /build
COPY requirements.txt .
RUN pip install --no-cache-dir --target=/build/deps -r requirements.txt
 
# ---- 実行ステージ ----
FROM python:3.12-slim
 
# セキュリティ: 非rootユーザーで実行
RUN groupadd -r appuser && useradd -r -g appuser appuser
 
WORKDIR /app
COPY --from=builder /build/deps /usr/local/lib/python3.12/site-packages/
COPY main.py .
 
# Cloud Run は PORT 環境変数を自動設定
ENV PORT=8080
ENV PYTHONUNBUFFERED=1
 
# ヘルスチェックエンドポイント
HEALTHCHECK --interval=30s --timeout=5s \
  CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')"
 
USER appuser
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080", "--workers", "1"]

gcloud run deploy gemini-api \
  --min-instances=1 \
  --max-instances=10 \
  --concurrency=80 \
  --cpu=2 \
  --memory=1Gi \
  --timeout=600

gcloud run deploy gemini-api \
  --cpu-boost

# シークレットの作成
echo -n "YOUR_GEMINI_API_KEY" | \
  gcloud secrets create gemini-api-key --data-file=-
 
# Cloud Run サービスにマウント
gcloud run deploy gemini-api \
  --set-secrets="GEMINI_API_KEY=gemini-api-key:latest"

# main.py に追加するミドルウェア
import time
 
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
 
    logger.info(
        "API request completed",
        extra={
            "json_fields": {
                "method": request.method,
                "path": request.url.path,
                "status": response.status_code,
                "duration_ms": round(duration * 1000, 2),
                "model": request.state.model if hasattr(request.state, "model") else "unknown",
            }
        },
    )
    return response

from google.api_core.exceptions import ResourceExhausted
 
async def stream_generate_with_retry(req: GenerateRequest, max_retries: int = 3):
    """指数バックオフ付きリトライ"""
    for attempt in range(max_retries):
        try:
            async for chunk in stream_generate(req):
                yield chunk
            return
        except ResourceExhausted:
            if attempt < max_retries - 1:
                wait = 2 ** attempt
                logger.warning(f"Rate limited, retrying in {wait}s (attempt {attempt + 1})")
                await asyncio.sleep(wait)
            else:
                yield {
                    "event": "error",
                    "data": json.dumps({"error": "Rate limit exceeded. Please try again later."}),
                }

# 予算アラートの作成（月額$50を超えたら通知）
gcloud billing budgets create \
  --billing-account=YOUR_BILLING_ACCOUNT \
  --display-name="Gemini API Cloud Run Budget" \
  --budget-amount=50USD \
  --threshold-rule=percent=80 \
  --threshold-rule=percent=100

# 推奨設定
gcloud run deploy gemini-api \
  --concurrency=80 \
  --cpu=2 \
  --memory=1Gi

from functools import lru_cache
import hashlib
 
# インメモリキャッシュ（非ストリーミングのみ）
_cache = {}
 
def get_cache_key(prompt: str, model: str, temperature: float) -> str:
    """プロンプトのハッシュをキーにする"""
    content = f"{model}:{temperature}:{prompt}"
    return hashlib.sha256(content.encode()).hexdigest()
 
async def cached_generate(req: GenerateRequest):
    """キャッシュ付き生成（temperature=0の場合のみ）"""
    if req.temperature == 0:
        key = get_cache_key(req.prompt, req.model, req.temperature)
        if key in _cache:
            logger.info("Cache hit")
            return _cache[key]
 
    result = await batch_generate(req)
 
    if req.temperature == 0:
        _cache[key] = result
 
    return result

# cloudbuild.yaml
steps:
  # テスト実行
  - name: 'python:3.12-slim'
    entrypoint: 'bash'
    args:
      - '-c'
      - |
        pip install -r requirements.txt
        pip install pytest httpx
        pytest tests/ -v
 
  # Docker イメージのビルド & プッシュ
  - name: 'gcr.io/cloud-builders/docker'
    args: ['build', '-t', 'gcr.io/$PROJECT_ID/gemini-api:$SHORT_SHA', '.']
 
  - name: 'gcr.io/cloud-builders/docker'
    args: ['push', 'gcr.io/$PROJECT_ID/gemini-api:$SHORT_SHA']
 
  # Cloud Run へデプロイ
  - name: 'gcr.io/cloud-builders/gcloud'
    args:
      - 'run'
      - 'deploy'
      - 'gemini-api'
      - '--image=gcr.io/$PROJECT_ID/gemini-api:$SHORT_SHA'
      - '--region=asia-northeast1'
      - '--platform=managed'
      - '--min-instances=1'
      - '--max-instances=10'
      - '--concurrency=80'
      - '--cpu=2'
      - '--memory=1Gi'
      - '--timeout=600'
      - '--cpu-boost'
      - '--set-secrets=GEMINI_API_KEY=gemini-api-key:latest'
 
options:
  logging: CLOUD_LOGGING_ONLY

# 新リビジョンをデプロイ（トラフィックは既存リビジョンのまま）
gcloud run deploy gemini-api \
  --image=gcr.io/PROJECT_ID/gemini-api:v2 \
  --no-traffic
 
# 10% のトラフィックを新リビジョンに振り分け
gcloud run services update-traffic gemini-api \
  --to-revisions=gemini-api-v2=10
 
# 問題なければ100%に移行
gcloud run services update-traffic gemini-api \
  --to-latest