2026 開発者完全攻略ハンドブック：Google Gemini 公式無料APIを活用したゼロコスト商用グレードアプリのアーキテクチャ設計とエラー完全トラブルシューティング指南

import os
import sys
import time
import json
import sqlite3
import random
import requests
from typing import List, Optional
from pydantic import BaseModel, Field

# ---------------------------------------------------------
# 1. 契約定義：Pydantic による構造化 JSON Schema
# ---------------------------------------------------------
class ReviewAnalysis(BaseModel):
    sentiment: str = Field(description="感情の極性。POSITIVE, NEGATIVE, NEUTRAL のいずれか")
    confidence: float = Field(description="分析の信頼度スコア（0.0〜1.0）")
    summary: str = Field(description="レビュー内容の簡潔な1文要約")
    key_tags: List[str] = Field(description="レビュー内で言及された製品属性のキーワードタグ")

# ---------------------------------------------------------
# 2. クライアント側レートリミッター：トークンバケットアルゴリズム
# ---------------------------------------------------------
class TokenBucketRateLimiter:
    def __init__(self, capacity: float, refill_rate: float):
        self.capacity = float(capacity)
        self.refill_rate = float(refill_rate)
        self.tokens = float(capacity)
        self.last_update = time.monotonic()

    def acquire(self):
        while True:
            now = time.monotonic()
            elapsed = now - self.last_update
            self.last_update = now
            self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
            
            if self.tokens >= 1.0:
                self.tokens -= 1.0
                return
            
            sleep_duration = (1.0 - self.tokens) / self.refill_rate
            time.sleep(max(0.01, sleep_duration))

# ---------------------------------------------------------
# 3. 永続化 SQLite デッドレターキュー（DLQ）マネージャー
# ---------------------------------------------------------
class TaskDatabase:
    def __init__(self, db_path: str = "review_tasks.db"):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self._init_schema()

    def _init_schema(self):
        with self.conn:
            self.conn.execute("""
                CREATE TABLE IF NOT EXISTS dead_letter_tasks (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    review_id TEXT UNIQUE,
                    content TEXT,
                    error_message TEXT,
                    failed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)

    def record_failure(self, review_id: str, content: str, error_msg: str):
        with self.conn:
            self.conn.execute(
                "INSERT OR REPLACE INTO dead_letter_tasks (review_id, content, error_message) VALUES (?, ?, ?)",
                (review_id, content, error_msg)
            )

# ---------------------------------------------------------
# 4. カスケードフォールバックと全ジッターを備えた本番クライアント
# ---------------------------------------------------------
class ResilientGeminiBatchClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://generativelanguage.googleapis.com/v1beta/models"
        self.model_cascade = [
            "gemini-3.7-flash",
            "gemini-2.5-flash",
            "gemini-2.5-flash-lite",
            "gemma-4-26b-a4b-it"
        ]
        self.rate_limiter = TokenBucketRateLimiter(capacity=15.0, refill_rate=15.0 / 60.0)
        self.db = TaskDatabase()

    def _clean_pii(self, text: str) -> str:
        """無料枠利用規約に準拠するためのローカル個人情報（PII）マスキング"""
        import re
        text = re.sub(r'[\w\.-]+@[\w\.-]+\.\w+', '[REDACTED_EMAIL]', text)
        text = re.sub(r'\b(?:\+?\d{1,3}[- ]?)?\(?\d{2,4}\)?[- ]?\d{3,4}[- ]?\d{3,4}\b', '[REDACTED_PHONE]', text)
        return text

    def analyze_review_with_failover(self, review_id: str, raw_review: str) -> Optional[ReviewAnalysis]:
        sanitized_content = self._clean_pii(raw_review)
        current_model_idx = 0
        max_attempts_per_model = 3

        while current_model_idx < len(self.model_cascade):
            model_name = self.model_cascade[current_model_idx]
            
            for attempt in range(max_attempts_per_model):
                self.rate_limiter.acquire()
                
                endpoint = f"{self.base_url}/{model_name}:generateContent"
                headers = {
                    "Content-Type": "application/json",
                    "x-goog-api-key": self.api_key
                }
                
                payload = {
                    "contents": [{
                        "parts": [{"text": f"以下の商品レビューを分析し、構造化指標を出力してください：\n{sanitized_content}"}]
                    }],
                    "generationConfig": {
                        "responseMimeType": "application/json",
                        "responseSchema": ReviewAnalysis.model_json_schema()
                    }
                }
                
                try:
                    response = requests.post(endpoint, headers=headers, json=payload, timeout=30)
                    status_code = response.status_code
                    
                    if status_code == 200:
                        data = response.json()
                        text_output = data["candidates"][0]["content"]["parts"][0]["text"]
                        return ReviewAnalysis.model_validate_json(text_output)
                    
                    # 再試行不可能なクライアントエラーは即時記録して終了
                    if status_code in [400, 401, 403, 404]:
                        error_detail = response.text
                        self.db.record_failure(review_id, raw_review, f"HTTP {status_code}: {error_detail}")
                        return None
                    
                    # レート制限またはサーバー一時エラーの処理
                    if status_code in [429, 500, 503]:
                        if "ResourceExhausted" in response.text and "Requests per day" in response.text:
                            # 1日あたりの上限（RPD）枯渇を検知した場合は次のモデルへ即時降級
                            break
                        
                        # Full Jitter 指数バックオフ計算
                        backoff_cap = min(32.0, 1.5 * (2 ** attempt))
                        sleep_time = random.uniform(0.0, backoff_cap)
                        time.sleep(sleep_time)
                        continue

                except (requests.exceptions.RequestException, KeyError, json.JSONDecodeError):
                    backoff_cap = min(32.0, 1.5 * (2 ** attempt))
                    time.sleep(random.uniform(0.0, backoff_cap))
                    continue
            
            # 当該モデルでの試行上限到達、またはRPD枯渇によりフォールバックツリーを次へ進める
            current_model_idx += 1

        self.db.record_failure(review_id, raw_review, "All cascade models exhausted.")
        return None

if __name__ == "__main__":
    API_KEY = os.getenv("GEMINI_API_KEY", "YOUR_GEMINI_API_KEY")
    client = ResilientGeminiBatchClient(api_key=API_KEY)
    
    sample_reviews = [
        {"id": f"rev_{i}", "text": f"音質は素晴らしいですが梱包に少し傷がありました。サポートの alice@vendor.com は迅速に対応してくれました。ID: {i}"}
        for i in range(5)
    ]
    
    for item in sample_reviews:
        result = client.analyze_review_with_failover(item["id"], item["text"])
        if result:
            print(f"[{item['id']}] 解析成功: {result.sentiment} | 信頼度: {result.confidence} | タグ: {result.key_tags}")
        else:
            print(f"[{item['id']}] 失敗。ローカル SQLite キューに記録されました。")

import http from 'http';
import https from 'https';
import { URL } from 'url';

const PORT = process.env.PORT || 8080;
const GEMINI_API_KEY = process.env.GEMINI_API_KEY || 'YOUR_GEMINI_API_KEY';

// カスケードフォールバックの優先順位リスト
const MODEL_CASCADE: string[] = [
  'gemini-3.7-flash',
  'gemini-2.5-flash',
  'gemini-2.5-flash-lite',
  'gemma-4-26b-a4b-it'
];

function attemptStreamRequest(
  clientRes: http.ServerResponse,
  requestBody: string,
  modelIndex: number
): void {
  // すべてのモデルが枯渇した場合は 502 Bad Gateway を返却
  if (modelIndex >= MODEL_CASCADE.length) {
    clientRes.writeHead(502, { 'Content-Type': 'application/json' });
    clientRes.end(JSON.stringify({
      error: {
        code: 502,
        status: 'BAD_GATEWAY',
        message: 'All Gemini fallback models were exhausted or throttled.'
      }
    }));
    return;
  }

  const currentModel = MODEL_CASCADE[modelIndex];
  const targetUrl = new URL(
    `https://generativelanguage.googleapis.com/v1beta/models/${currentModel}:streamGenerateContent?alt=sse`
  );

  const options: https.RequestOptions = {
    method: 'POST',
    hostname: targetUrl.hostname,
    path: `${targetUrl.pathname}${targetUrl.search}`,
    headers: {
      'Content-Type': 'application/json',
      'x-goog-api-key': GEMINI_API_KEY,
      'Content-Length': Buffer.byteLength(requestBody)
    },
    timeout: 30000
  };

  const proxyReq = https.request(options, (upstreamRes) => {
    const statusCode = upstreamRes.statusCode || 500;

    // 429（過負荷）または 503（一時不能）を検知した場合、透過的に次のモデルへフォールバック
    if (statusCode === 429 || statusCode === 503) {
      upstreamRes.resume(); // コネクションプールを安全に解放
      const nextIndex = modelIndex + 1;
      attemptStreamRequest(clientRes, requestBody, nextIndex);
      return;
    }

    // 400番台のクライアントエラーはそのまま返却
    if (statusCode >= 400 && statusCode < 500) {
      clientRes.writeHead(statusCode, { 'Content-Type': 'application/json' });
      upstreamRes.pipe(clientRes);
      return;
    }

    // 正常接続：Server-Sent Events（SSE）ヘッダーを設定してストリーム中継
    clientRes.writeHead(200, {
      'Content-Type': 'text/event-stream; charset=utf-8',
      'Cache-Control': 'no-cache, no-transform',
      'Connection': 'keep-alive',
      'X-Served-By-Model': currentModel
    });

    upstreamRes.pipe(clientRes);
  });

  proxyReq.on('error', () => {
    attemptStreamRequest(clientRes, requestBody, modelIndex + 1);
  });

  proxyReq.on('timeout', () => {
    proxyReq.destroy();
    attemptStreamRequest(clientRes, requestBody, modelIndex + 1);
  });

  proxyReq.write(requestBody);
  proxyReq.end();
}

const server = http.createServer((req, res) => {
  // CORS ヘッダーの設定
  res.setHeader('Access-Control-Allow-Origin', '*');
  res.setHeader('Access-Control-Allow-Methods', 'POST, OPTIONS');
  res.setHeader('Access-Control-Allow-Headers', 'Content-Type');

  if (req.method === 'OPTIONS') {
    res.writeHead(204);
    res.end();
    return;
  }

  if (req.method === 'POST' && req.url === '/v1/chat/stream') {
    let rawBody = '';
    req.on('data', (chunk) => {
      rawBody += chunk;
      // 巨大ペイロードによる DoS を防ぐための制限（1MB）
      if (rawBody.length > 1024 * 1024) {
        res.writeHead(413, { 'Content-Type': 'application/json' });
        res.end(JSON.stringify({ error: 'Payload Too Large' }));
        req.destroy();
      }
    });

    req.on('end', () => {
      try {
        JSON.parse(rawBody); // JSONバリデーション
        attemptStreamRequest(res, rawBody, 0);
      } catch {
        res.writeHead(400, { 'Content-Type': 'application/json' });
        res.end(JSON.stringify({ error: 'Invalid JSON body' }));
      }
    });
  } else {
    res.writeHead(404, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({ error: 'Route Not Found' }));
  }
});

server.listen(PORT, () => {
  process.stdout.write(`High-Availability Gemini Proxy Server listening on port ${PORT}\n`);
});