Gemini API Error Handling: Graceful Degradation สำหรับ Production

ในฐานะวิศวกรที่พัฒนา AI Application มาหลายปี ผมเจอปัญหา Gemini API ล่มกลางคันมาหลายครั้ง บางครั้งทำให้ระบบทั้งระบบล่มไปด้วย วันนี้ผมจะมาแชร์วิธีที่ผมใช้ Graceful Degradation เพื่อให้ระบบยังทำงานได้แม้ API มีปัญหา

ตารางเปรียบเทียบบริการ API

บริการ	ราคา/MTok	Latency	ความเสถียร	วิธีชำระเงิน
HolySheep AI	$0.42 - $8	<50ms	99.9%	WeChat/Alipay
API อย่างเป็นทางการ	$2.50 - $15	100-500ms	99.5%	บัตรเครดิต
บริการรีเลย์อื่น	$1 - $12	80-300ms	98%	หลากหลาย

HolySheep AI โดดเด่นเรื่องราคาประหยัดสูงสุด 85%+ เมื่อเทียบกับ API อย่างเป็นทางการ และ latency ต่ำกว่า 50ms พร้อมรองรับ WeChat/Alipay สมัครที่นี่ เพื่อรับเครดิตฟรีเมื่อลงทะเบียน

ทำไมต้อง Graceful Degradation?

จากประสบการณ์ของผม การเรียก API โดยตรงโดยไม่มี error handling ที่ดีนั้นอันตรายมาก ลองนึกภาพว่าผู้ใช้งาน 1,000 คนกำลังใช้งานระบบของคุณ แต่ Gemini API ล่มไป 10 นาที หากไม่มี fallback คุณจะเสียผู้ใช้ทั้งหมด แต่ถ้ามี Graceful Degradation คุณอาจเสียแค่ 10% ของผู้ใช้เท่านั้น

โครงสร้างพื้นฐานสำหรับ Error Handling

class GeminiClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.model = "gemini-2.0-flash"
        self.fallback_model = "deepseek-v3.2"
        
    async def generate_with_fallback(
        self, 
        prompt: str, 
        max_retries: int = 3,
        timeout: float = 30.0
    ) -> dict:
        """Generate response with automatic fallback on failure"""
        
        # Strategy 1: Primary model (Gemini)
        try:
            response = await self._call_gemini(prompt, timeout)
            return {"status": "success", "data": response, "model": self.model}
            
        except (TimeoutError, ConnectionError) as e:
            # Retry with exponential backoff
            for attempt in range(max_retries):
                try:
                    await asyncio.sleep(2 ** attempt)  # 1s, 2s, 4s
                    response = await self._call_gemini(prompt, timeout)
                    return {"status": "success", "data": response, "model": self.model}
                except:
                    continue
                    
            # Strategy 2: Fallback to DeepSeek
            try:
                response = await self._call_deepseek(prompt, timeout=60.0)
                return {"status": "fallback", "data": response, "model": self.fallback_model}
            except:
                return {"status": "failed", "error": "All providers unavailable"}
                
        except RateLimitError as e:
            # Handle rate limit specifically
            return await self._handle_rate_limit(prompt)

Circuit Breaker Pattern สำหรับ Gemini API

import time
from enum import Enum
from dataclasses import dataclass

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing recovery

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5      # Open after 5 failures
    recovery_timeout: float = 60.0  # Try recovery after 60s
    half_open_max_calls: int = 3   # Allow 3 test calls
    
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    last_failure_time: float = 0.0
    half_open_calls: int = 0
    
    def call(self, func, *args, **kwargs):
        current_time = time.time()
        
        # Check if circuit should transition
        if self.state == CircuitState.OPEN:
            if current_time - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
            else:
                raise CircuitOpenError("Circuit breaker is OPEN")
        
        # Execute the call
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
            
    def _on_success(self):
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.half_open_calls += 1
            if self.half_open_calls >= self.half_open_max_calls:
                self.state = CircuitState.CLOSED
                
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

Usage
gemini_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30.0)

โค้ดสำหรับ Multiple Provider Fallback

import asyncio
from typing import List, Optional, Dict, Any

class MultiProviderClient:
    """Client with automatic provider fallback chain"""
    
    def __init__(self):
        self.providers = [
            {"name": "gemini-2.5-flash", "url": "https://api.holysheep.ai/v1/chat/completions", "priority": 1},
            {"name": "deepseek-v3.2", "url": "https://api.holysheep.ai/v1/chat/completions", "priority": 2},
            {"name": "gpt-4.1", "url": "https://api.holysheep.ai/v1/chat/completions", "priority": 3},
        ]
        
    async def generate(
        self, 
        prompt: str, 
        api_key: str,
        context: Optional[Dict] = None
    ) -> Dict[str, Any]:
        
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        
        errors = []
        
        for provider in self.providers:
            try:
                response = await self._make_request(
                    url=provider["url"],
                    headers=headers,
                    model=provider["name"],
                    prompt=prompt,
                    timeout=30.0
                )
                
                return {
                    "success": True,
                    "data": response,
                    "provider": provider["name"],
                    "latency_ms": response.get("latency", 0)
                }
                
            except ProviderError as e:
                errors.append({"provider": provider["name"], "error": str(e)})
                continue
                
        # All providers failed
        return {
            "success": False,
            "errors": errors,
            "fallback_content": self._generate_fallback_response(prompt, context)
        }
        
    async def _make_request(
        self, 
        url: str, 
        headers: Dict, 
        model: str,
        prompt: str,
        timeout: float
    ) -> Dict:
        """Make actual API request with timeout"""
        async with asyncio.timeout(timeout):
            payload = {
                "model": model,
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.7
            }
            
            async with aiohttp.ClientSession() as session:
                start = time.time()
                async with session.post(url, json=payload, headers=headers) as resp:
                    latency = (time.time() - start) * 1000
                    
                    if resp.status == 200:
                        data = await resp.json()
                        data["latency"] = latency
                        return data
                    else:
                        raise ProviderError(f"HTTP {resp.status}")

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

กรณีที่ 1: 429 Rate Limit Error

อาการ: ได้รับ error 429 บ่อยครั้ง โดยเฉพาะเมื่อมี traffic สูง

# ❌ ไม่ดี: Retry ทันทีทำให้ล่มหนักขึ้น
for _ in range(10):
    response = await client.generate(prompt)
    if response.status == 200:
        break

✅ ดี: Exponential backoff พร้อม queue
async def generate_with_queue(prompt: str, client: GeminiClient) -> dict:
    retry_count = 0
    max_retries = 5
    base_delay = 1.0  # 1 วินาที
    
    while retry_count < max_retries:
        try:
            response = await client.generate(prompt)
            if response.status == 200:
                return response
        except RateLimitError as e:
            retry_count += 1
            # Exponential backoff: 1s, 2s, 4s, 8s, 16s
            delay = base_delay * (2 ** (retry_count - 1))
            # บวก jitter แบบสุ่ม ±500ms เพื่อไม่ให้ request ชนกัน
            delay += random.uniform(-0.5, 0.5)
            await asyncio.sleep(delay)
            
    return {"status": "queued", "message": "Request queued for later processing"}

กรณีที่ 2: Connection Timeout ตอน Production

อาการ: Request ค้างนานเกินไปจน client timeout แต่ server ยังประมวลผลอยู่

# ❌ ไม่ดี: ไม่มี timeout ที่เหมาะสม
response = requests.post(url, json=payload, headers=headers)

✅ ดี: กำหนด timeout หลายระดับพร้อม retry
import httpx

class TimeoutAwareClient:
    def __init__(self, api_key: str):
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(
                connect=5.0,      # เชื่อมต่อ max 5 วินาที
                read=30.0,        # รอ response max 30 วินาที
                write=10.0,       # ส่ง request max 10 วินาที
                pool=5.0          # รอใน queue max 5 วินาที
            ),
            limits=httpx.Limits(max_keepalive_connections=20)
        )
        self.api_key = api_key
        
    async def safe_generate(self, prompt: str) -> dict:
        try:
            response = await self.client.post(
                "https://api.holysheep.ai/v1/chat/completions",
                json={"model": "gemini-2.5-flash", "messages": [{"role": "user", "content": prompt}]},
                headers={"Authorization": f"Bearer {self.api_key}"}
            )
            return {"status": "success", "data": response.json()}
            
        except httpx.TimeoutException as e:
            # เก็บ log แล้ว return fallback
            logger.error(f"Timeout: {str(e)}")
            return await self._generate_fallback(prompt)
            
        except httpx.ConnectError:
            # Network error - ใช้ cache หรือ fallback
            return await self._get_from_cache(prompt)

กรณีที่ 3: Invalid Response Format

อาการ: API คืนค่ากลับมาแต่ format ไม่ตรงตามที่คาดหวัง ทำให้ parse error

# ❌ ไม่ดี: สมมติว่า response จะมี format ที่ถูกต้องเสมอ
data = response.json()
content = data["choices"][0]["message"]["content"]

✅ ดี: Validate response format ก่อนใช้งาน
from pydantic import BaseModel, ValidationError
from typing import Optional

class APIResponse(BaseModel):
    id: str
    model: str
    choices: list
    usage: Optional[dict] = None
    
    class Config:
        extra = "allow"  # อนุญาตให้มี field เกินมาได้

def safe_parse_response(response: httpx.Response) -> dict:
    try:
        raw_data = response.json()
        validated = APIResponse(**raw_data)
        
        return {
            "success": True,
            "content": validated.choices[0]["message"]["content"],
            "model": validated.model,
            "tokens_used": validated.usage.get("total_tokens", 0)
        }
        
    except ValidationError as e:
        # Log error แต่พยายาม extract ข้อมูลที่มี
        logger.warning(f"Response validation failed: {e}")
        
        # Fallback: ดึงข้อมูลจาก raw response
        raw_data = response.json()
        return {
            "success": True,
            "content": raw_data.get("content", raw_data.get("text", "")),
            "model": raw_data.get("model", "unknown"),
            "tokens_used": 0,
            "warning": "Response format non-standard"
        }
        
    except Exception as e:
        logger.error(f"Failed to parse response: {e}")
        return {
            "success": False,
            "error": str(e),
            "raw_response": response.text[:500]  # เก็บ debug info
        }

Best Practices จากประสบการณ์จริง

• กำหนด timeout เหมาะสม: Production ควรตั้ง timeout ไม่เกิน 30 วินาที เพื่อไม่ให้ request ค้างนานเกินไป
• ใช้ Circuit Breaker: ป้องกันไม่ให้ระบบพยายามเรียก API ที่ล่มอยู่ซ้ำๆ จนทำให้ระบบล่มตามไปด้วย
• มี fallback หลายระดับ: เช่น Gemini → DeepSeek → Cache → Static Response
• เก็บ log ทุก error: ช่วยให้วิเคราะห์ปัญหาและ optimize ได้
• ทดสอบ failover สม่ำเสมอ: ทำ chaos testing เพื่อให้แน่ใจว่า fallback ทำงานได้จริง

สรุป

Graceful Degradation ไม่ใช่ทางเลือก แต่เป็นสิ่งจำเป็นสำหรับ production system ที่ต้องการความเสถียรสูง จากประสบการณ์ของผม ระบบที่มี error handling ที่ดีสามารถลด downtime ได้ถึง 90% และ HolySheep AI ก็เป็นตัวเลือกที่ดีด้วยราคาประหยัด 85%+ และ latency ต่ำกว่า 50ms ทำให้เป็น fallback ที่เหมาะสม

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน