การใช้งาน AI API ในระดับ Production มีความเสี่ยงหลายประการ ไม่ว่าจะเป็น API ล่ม การตอบสนองช้า หรือค่าใช้จ่ายที่พุ่งสูงจากการ retry ซ้ำซ้อน บทความนี้จะสอนการใช้งาน Circuit Breaker Pattern เพื่อป้องกันปัญหาเหล่านี้ พร้อมโค้ด Production-Ready ที่พิสูจน์แล้ว

ทำไมต้องใช้ Circuit Breaker

เมื่อ AI API มีปัญหา (เช่น latency สูงผิดปกติ หรือ error rate พุ่งสูง) ระบบที่ไม่มีการป้องกันจะยังคงส่ง request ไปเรื่อยๆ ทำให้เกิด:

State Machine ของ Circuit Breaker

Circuit Breaker มี 3 สถานะหลัก:

Implementation แบบ Production-Ready

import time
import asyncio
import logging
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, TypeVar, Optional
from collections import deque
import aiohttp

logger = logging.getLogger(__name__)

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

T = TypeVar('T')

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5        # จำนวน failure ก่อนเปิด circuit
    success_threshold: int = 3        # จำนวน success หลัง half-open ก่อนปิด
    timeout: float = 60.0             # วินาที ก่อนเปลี่ยน OPEN -> HALF-OPEN
    half_open_max_calls: int = 3      # จำนวน request สูงสุดใน half-open state

@dataclass
class CircuitBreakerMetrics:
    total_calls: int = 0
    successful_calls: int = 0
    failed_calls: int = 0
    rejected_calls: int = 0
    consecutive_failures: int = 0
    consecutive_successes: int = 0
    last_failure_time: Optional[float] = None
    state_history: deque = field(default_factory=lambda: deque(maxlen=100))

class CircuitBreaker:
    def __init__(self, name: str, config: CircuitBreakerConfig):
        self.name = name
        self.config = config
        self.state = CircuitState.CLOSED
        self.metrics = CircuitBreakerMetrics()
        self._half_open_calls = 0
        self._lock = asyncio.Lock()
    
    async def call(
        self, 
        func: Callable[..., T], 
        *args, 
        fallback: Optional[Callable] = None,
        **kwargs
    ) -> T:
        """Execute function with circuit breaker protection"""
        self.metrics.total_calls += 1
        
        # Check if request should be rejected
        if not await self._allow_request():
            self.metrics.rejected_calls += 1
            logger.warning(f"Circuit [{self.name}] OPEN - rejecting request")
            
            if fallback:
                return await fallback(*args, **kwargs) if asyncio.iscoroutinefunction(fallback) else fallback(*args, **kwargs)
            raise CircuitBreakerOpenError(f"Circuit {self.name} is OPEN")
        
        try:
            result = await func(*args, **kwargs) if asyncio.iscoroutinefunction(func) else func(*args, **kwargs)
            await self._on_success()
            return result
        except Exception as e:
            await self._on_failure()
            raise
    
    async def _allow_request(self) -> bool:
        async with self._lock:
            if self.state == CircuitState.CLOSED:
                return True
            
            if self.state == CircuitState.OPEN:
                if time.time() - self.metrics.last_failure_time >= self.config.timeout:
                    self.state = CircuitState.HALF_OPEN
                    self._half_open_calls = 0
                    logger.info(f"Circuit [{self.name}] transitioning to HALF-OPEN")
                    return True
                return False
            
            if self.state == CircuitState.HALF_OPEN:
                if self._half_open_calls < self.config.half_open_max_calls:
                    self._half_open_calls += 1
                    return True
                return False
            
            return False
    
    async def _on_success(self):
        async with self._lock:
            self.metrics.consecutive_successes += 1
            self.metrics.consecutive_failures = 0
            self.metrics.successful_calls += 1
            
            if self.state == CircuitState.HALF_OPEN:
                if self.metrics.consecutive_successes >= self.config.success_threshold:
                    self.state = CircuitState.CLOSED
                    self.metrics.consecutive_successes = 0
                    logger.info(f"Circuit [{self.name}] CLOSED after recovery")
    
    async def _on_failure(self):
        async with self._lock:
            self.metrics.consecutive_failures += 1
            self.metrics.consecutive_failures += 1  # Extra for half-open call
            self.metrics.failed_calls += 1
            self.metrics.last_failure_time = time.time()
            self.metrics.consecutive_successes = 0
            
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.OPEN
                logger.warning(f"Circuit [{self.name}] back to OPEN after half-open failure")
            elif self.metrics.consecutive_failures >= self.config.failure_threshold:
                self.state = CircuitState.OPEN
                logger.warning(f"Circuit [{self.name}] OPEN after {self.metrics.consecutive_failures} failures")
    
    def get_status(self) -> dict:
        return {
            "name": self.name,
            "state": self.state.value,
            "metrics": {
                "total": self.metrics.total_calls,
                "success": self.metrics.successful_calls,
                "failed": self.metrics.failed_calls,
                "rejected": self.metrics.rejected_calls,
                "consecutive_failures": self.metrics.consecutive_failures
            }
        }

class CircuitBreakerOpenError(Exception):
    pass

AI API Client พร้อม Circuit Breaker

ตัวอย่างการใช้งานกับ HolySheep AI API ที่มีอัตรา ¥1=$1 (ประหยัด 85%+) และ response time <50ms

import asyncio
import aiohttp
import json
from circuit_breaker import CircuitBreaker, CircuitBreakerConfig, CircuitBreakerOpenError

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"

class AIAPIClient:
    def __init__(self):
        self.circuit_breaker = CircuitBreaker(
            name="holysheep-ai",
            config=CircuitBreakerConfig(
                failure_threshold=3,
                success_threshold=2,
                timeout=30.0,
                half_open_max_calls=2
            )
        )
        self.session: Optional[aiohttp.ClientSession] = None
    
    async def _get_session(self) -> aiohttp.ClientSession:
        if self.session is None or self.session.closed:
            self.session = aiohttp.ClientSession(
                headers={
                    "Authorization": f"Bearer {API_KEY}",
                    "Content-Type": "application/json"
                },
                timeout=aiohttp.ClientTimeout(total=30)
            )
        return self.session
    
    async def chat_completion(
        self,
        messages: list,
        model: str = "gpt-4.1",
        temperature: float = 0.7,
        max_tokens: int = 1000
    ):
        """Send chat completion request with circuit breaker protection"""
        
        async def _make_request():
            session = await self._get_session()
            payload = {
                "model": model,
                "messages": messages,
                "temperature": temperature,
                "max_tokens": max_tokens
            }
            
            async with session.post(f"{BASE_URL}/chat/completions", json=payload) as response:
                if response.status != 200:
                    text = await response.text()
                    raise AIAPIError(f"API Error {response.status}: {text}")
                
                result = await response.json()
                return result
        
        async def fallback_handler(messages, **kwargs):
            logger.warning("Returning cached/fallback response")
            return {
                "fallback": True,
                "content": "Service temporarily unavailable. Please try again later.",
                "model": model
            }
        
        try:
            result = await self.circuit_breaker.call(
                _make_request,
                fallback=fallback_handler
            )
            return result
        except CircuitBreakerOpenError:
            return await fallback_handler(messages)

    async def batch_completion(self, requests: list) -> list:
        """Process multiple requests with concurrency control"""
        semaphore = asyncio.Semaphore(5)  # Max 5 concurrent requests
        
        async def limited_request(req):
            async with semaphore:
                return await self.chat_completion(**req)
        
        tasks = [limited_request(req) for req in requests]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        return [
            r if not isinstance(r, Exception) else {"error": str(r)}
            for r in results
        ]
    
    async def close(self):
        if self.session and not self.session.closed:
            await self.session.close()

class AIAPIError(Exception):
    pass


Usage Example

async def main():
    client = AIAPIClient()
    
    messages = [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain circuit breaker pattern in Thai."}
    ]
    
    try:
        response = await client.chat_completion(messages)
        print(f"Response: {response['choices'][0]['message']['content']}")
    except Exception as e:
        print(f"Error: {e}")
    finally:
        await client.close()

if __name__ == "__main__":
    asyncio.run(main())

Benchmark Results

ทดสอบประสิทธิภาพของ Circuit Breaker ในสถานการณ์ต่างๆ:

Advanced Configuration สำหรับ Production

# production_config.py
from circuit_breaker import CircuitBreakerConfig


Configuration ที่ปรับแต่งแล้วสำหรับ AI API

PRODUCTION_CONFIG = {
    # HolySheep AI - ราคาถูก + response เร็ว <50ms
    "holysheep": CircuitBreakerConfig(
        failure_threshold=5,
        success_threshold=3,
        timeout=30.0,
        half_open_max_calls=3
    ),
    
    # Fallback API - ให้เวลามากกว่าเพราะอาจช้ากว่า
    "fallback": CircuitBreakerConfig(
        failure_threshold=3,
        success_threshold=2,
        timeout=60.0,
        half_open_max_calls=2
    ),
}


Rate Limiter Integration

class RateLimitedCircuitBreaker:
    def __init__(self, circuit: CircuitBreaker, max_calls_per_minute: int):
        self.circuit = circuit
        self.max_calls = max_calls_per_minute
        self.calls = deque(maxlen=max_calls_per_minute)
        self._lock = asyncio.Lock()
    
    async def call(self, func, *args, **kwargs):
        async with

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง