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 結構化 Schema
# ---------------------------------------------------------
class ReviewAnalysis(BaseModel):
    sentiment: str = Field(description="情感極性，必須為 POSITIVE, NEGATIVE 或 NEUTRAL")
    confidence: float = Field(description="分析信心指標，介於 0.0 至 1.0 之間")
    summary: str = Field(description="評論的精簡一句話核心摘要")
    key_tags: List[str] = Field(description="評論中提及的產品屬性關鍵標籤")

# ---------------------------------------------------------
# 2. 客戶端本地速率限制器：權杖桶演算法 (Token Bucket)
# ---------------------------------------------------------
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 任務隊列管理器
# ---------------------------------------------------------
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:
        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:
                            break  # RPD 耗盡，切換下一模型
                        
                        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
            
            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 回覆很迅速。代號：{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 {
  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;
    }

    if (statusCode >= 400 && statusCode < 500) {
      clientRes.writeHead(statusCode, { 'Content-Type': 'application/json' });
      upstreamRes.pipe(clientRes);
      return;
    }

    // 成功建立串流連線，轉發 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) => {
  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;
      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);
        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`);
});