Pattern Circuit Breaker สำหรับ AI Services: ปกป้องระบบของคุณจากความล้มเหลว

บทนำ: เมื่อ AI API ล่มและระบบทั้งหมดพังทลาย

เช้าวันจันทร์ที่ยุ่งเหยิง ระบบ chatbot ของบริษัทคุณเริ่มส่งข้อผิดพลาดไปที่ลูกค้าทั้งหมด ทีม DevOps ตรวจสอบพบว่า:

ConnectionError: timeout after 30s
HTTPSConnectionPool(host='api.holysheep.ai', port=443): 
Max retries exceeded (Caused by ConnectTimeoutError)

หรือในกรณี API key หมดอายุ:
401 Unauthorized - Invalid API key or expired credentials
Retry-After: 3600

ปัญหาเกิดจาก cascade failure — เมื่อ AI service ตอบสนองช้าหรือล่ม คำขอจากผู้ใช้ทั้งหมดเริ่มค้าง ทำให้เกิด thread pool exhaustion และระบบทั้งหมดล่มในที่สุด วันนี้ผมจะสอนวิธี implement Circuit Breaker pattern ด้วย Python เพื่อป้องกันปัญหานี้ โดยใช้ HolySheep AI เป็นตัวอย่าง ซึ่งมีอัตรา ¥1=$1 (ประหยัด 85%+) และ latency <50ms

Circuit Breaker Pattern คืออะไร?

Circuit Breaker เปรียบเสมือนฟิวส์ไฟฟ้าในบ้าน — เมื่อกระแสไฟฟ้าลัดวงจร ฟิวส์จะตัดวงจรทันทีเพื่อป้องกันไม่ให้อุปกรณ์ไฟฟ้าเสียหาย ในระบบ software ก็เช่นเดียวกัน: สถานะ 3 ของ Circuit Breaker:

• CLOSED — ทำงานปกติ คำขอผ่านไปยัง AI service ได้
• OPEN — Circuit ถูกตัด คำขอถูก reject ทันทีโดยไม่ไปถึง API
• HALF-OPEN — ทดสอบว่า service กลับมาทำงานแล้วหรือยัง

การ Implement Circuit Breaker สำหรับ HolySheep AI

import time
import asyncio
from enum import Enum
from typing import Callable, Any
from dataclasses import dataclass, field
import aiohttp
import httpx

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

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5      # จำนวนครั้งที่ล้มเหลวก่อนเปิด circuit
    success_threshold: int = 3      # จำนวนครั้งที่ต้องสำเร็จใน half-open ก่อนปิด
    timeout_duration: float = 60.0  # วินาทีที่จะเปิด circuit ค้างไว้
    half_open_max_calls: int = 3    # จำนวนคำขอสูงสุดใน half-open

class CircuitBreaker:
    def __init__(self, config: CircuitBreakerConfig = None):
        self.config = config or CircuitBreakerConfig()
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
        self.last_state_change = time.time()
        self.half_open_calls = 0

    def _should_allow_request(self) -> bool:
        now = time.time()
        
        if self.state == CircuitState.CLOSED:
            return True
        
        if self.state == CircuitState.OPEN:
            if now - self.last_failure_time >= self.config.timeout_duration:
                self._transition_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

    def _transition_to_half_open(self):
        self.state = CircuitState.HALF_OPEN
        self.half_open_calls = 1
        self.last_state_change = time.time()

    def record_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.config.success_threshold:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                self.success_count = 0
                self.last_state_change = time.time()
        else:
            self.failure_count = 0

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.state == CircuitState.HALF_OPEN:
            self._transition_to_open()
        elif self.failure_count >= self.config.failure_threshold:
            self._transition_to_open()

    def _transition_to_open(self):
        self.state = CircuitState.OPEN
        self.last_state_change = time.time()
        self.success_count = 0

    async def call(self, func: Callable, *args, **kwargs) -> Any:
        if not self._should_allow_request():
            raise CircuitBreakerOpenError(
                f"Circuit is {self.state.value}. "
                f"Wait {self.config.timeout_duration}s before retry."
            )
        
        try:
            result = await func(*args, **kwargs)
            self.record_success()
            return result
        except Exception as e:
            self.record_failure()
            raise

Custom Exception
class CircuitBreakerOpenError(Exception):
    pass

การใช้งานกับ HolySheep AI API

import asyncio
from circuit_breaker import CircuitBreaker, CircuitBreakerConfig, CircuitBreakerOpenError

Configuration สำหรับ AI Service
ai_circuit_breaker = CircuitBreaker(
    config=CircuitBreakerConfig(
        failure_threshold=3,      # หลังจากล้มเหลว 3 ครั้ง
        success_threshold=2,      # ต้องสำเร็จ 2 ครั้งใน half-open
        timeout_duration=30.0,    # รอ 30 วินาที
        half_open_max_calls=1    # ทดสอบทีละคำขอ
    )
)

BASE_URL = "https://api.holysheep.ai/v1"
HEADERS = {
    "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
    "Content-Type": "application/json"
}

async def call_holysheep_chat(prompt: str) -> dict:
    """เรียก HolySheep AI Chat API พร้อม Circuit Breaker"""
    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(
            f"{BASE_URL}/chat/completions",
            headers=HEADERS,
            json={
                "model": "gpt-4.1",
                "messages": [{"role": "user", "content": prompt}],
                "max_tokens": 1000
            }
        )
        response.raise_for_status()
        return response.json()

async def handle_user_request(user_id: str, prompt: str) -> str:
    """Handler หลักที่ใช้ Circuit Breaker"""
    
    try:
        result = await ai_circuit_breaker.call(
            call_holysheep_chat, 
            prompt
        )
        return result["choices"][0]["message"]["content"]
    
    except CircuitBreakerOpenError:
        # Circuit เปิดอยู่ ให้ return fallback response
        return "ระบบ AI กำลังรีสตาร์ท กรุณาลองใหม่ในอีกสักครู่"
    
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 401:
            return "ระบบไม่สามารถเข้าถึง AI service ได้ (Unauthorized)"
        elif e.response.status_code == 429:
            return "จำนวนคำขอเกินขีดจำกัด กรุณารอสักครู่"
        raise

async def main():
    # ทดสอบเรียกหลายครั้ง
    for i in range(10):
        try:
            result = await handle_user_request(
                user_id=f"user_{i}",
                prompt=f"บอกอะไรสั้นๆ เกี่ยวกับหัวข้อที่ {i}"
            )
            print(f"Request {i}: {result[:50]}...")
        except Exception as e:
            print(f"Request {i} Error: {e}")
        await asyncio.sleep(1)

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

Implementation แบบ Production-Ready พร้อม Fallback

import logging
from functools import wraps
from typing import Optional, List, Dict, Any

logger = logging.getLogger(__name__)

class AIServiceOrchestrator:
    """
    Orchestrator ที่จัดการ AI services หลายตัวพร้อม Circuit Breaker
    รองรับ fallback และ graceful degradation
    """
    
    def __init__(self):
        self.circuit_breakers: Dict[str, CircuitBreaker] = {}
        self.fallback_responses: Dict[str, str] = {}
        self._init_services()
    
    def _init_services(self):
        """กำหนด AI services ที่รองรับ"""
        
        # HolySheep AI - ราคาประหยัด รองรับ GPT-4.1, Claude, Gemini
        self.circuit_breakers["holysheep"] = CircuitBreaker(
            config=CircuitBreakerConfig(
                failure_threshold=5,
                success_threshold=2,
                timeout_duration=45.0
            )
        )
        self.fallback_responses["holysheep"] = "ขออภัย AI service หลักไม่พร้อมใช้งาน"
        
        # สามารถเพิ่ม providers อื่นๆ ได้
        # self.circuit_breakers["openai"] = CircuitBreaker(...)
    
    async def chat_completion(
        self,
        prompt: str,
        preferred_model: str = "gpt-4.1",
        use_fallback: bool = True
    ) -> Dict[str, Any]:
        """
        ส่งคำขอไปยัง AI service พร้อม fallback logic
        
        Args:
            prompt: ข้อความที่ต้องการส่งให้ AI
            preferred_model: โมเดลที่ต้องการใช้
            use_fallback: ใช้ fallback response หรือไม่
        """
        
        # ลำดับความสำคัญของ services
        service_order = ["holysheep"]
        
        last_error = None
        
        for service in service_order:
            breaker = self.circuit_breakers[service]
            
            if not breaker._should_allow_request():
                logger.warning(f"Circuit open for {service}, skipping...")
                continue
            
            try:
                result = await self._call_service(service, prompt, preferred_model)
                breaker.record_success()
                return {
                    "success": True,
                    "data": result,
                    "service": service,
                    "model": preferred_model
                }
            
            except Exception as e:
                breaker.record_failure()
                last_error = e
                logger.error(f"{service} failed: {e}")
                continue
        
        # ทุก service ล้มเหลว
        if use_fallback:
            return {
                "success": False,
                "data": self._get_fallback_response(prompt),
                "error": str(last_error),
                "fallback": True
            }
        
        raise last_error
    
    async def _call_service(
        self, 
        service: str, 
        prompt: str, 
        model: str
    ) -> str:
        """เรียก AI service ตามชื่อที่กำหนด"""
        
        if service == "holysheep":
            async with httpx.AsyncClient(timeout=30.0) as client:
                response = await client.post(
                    f"{BASE_URL}/chat/completions",
                    headers=HEADERS,
                    json={
                        "model": model,
                        "messages": [{"role": "user", "content": prompt}],
                        "temperature": 0.7,
                        "max_tokens": 2000
                    }
                )
                
                if response.status_code == 401:
                    raise PermissionError("Invalid HolySheep API key")
                
                response.raise_for_status()
                data = response.json()
                return data["choices"][0]["message"]["content"]
        
        raise ValueError(f"Unknown service: {service}")
    
    def _get_fallback_response(self, prompt: str) -> str:
        """Fallback response เมื่อ AI services ทั้งหมดล่ม"""
        return (
            "ขออภัยครับ ระบบ AI กำลังมีปัญหา "
            "กรุณาติดต่อฝ่ายสนับสนุนหรือลองใหม่ภายหลัง"
        )
    
    def get_health_status(self) -> Dict[str, Any]:
        """ตรวจสอบสถานะของทุก circuit breaker"""
        return {
            service: {
                "state": breaker.state.value,
                "failure_count": breaker.failure_count,
                "last_failure": breaker.last_failure_time
            }
            for service, breaker in self.circuit_breakers.items()
        }

การใช้งาน
async def example_usage():
    orchestrator = AIServiceOrchestrator()
    
    # คำขอปกติ
    result = await orchestrator.chat_completion(
        prompt="อธิบายเรื่อง Circuit Breaker pattern สั้นๆ"
    )
    
    if result["success"]:
        print(f"AI Response: {result['data']}")
        print(f"Via: {result['service']}/{result['model']}")
    else:
        print(f"Fallback: {result['data']}")
    
    # ตรวจสอบสถานะ
    print(orchestrator.get_health_status())

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

1. 401 Unauthorized - API Key ไม่ถูกต้อง

# ❌ ผิด: Header ผิด format
HEADERS = {
    "Authorization": "YOUR_HOLYSHEEP_API_KEY",  # ขาด Bearer
    "Content-Type": "application/json"
}

✅ ถูกต้อง: Bearer token format
HEADERS = {
    "Authorization": f"Bearer {os.environ.get('HOLYSHEEP_API_KEY')}",
    "Content-Type": "application/json"
}

หรือใช้ .env file
pip install python-dotenv
from dotenv import load_dotenv
load_dotenv()

HEADERS = {
    "Authorization": f"Bearer {os.getenv('HOLYSHEEP_API_KEY')}",
    "Content-Type": "application/json"
}

2. Circuit Breaker ตั้งค่า timeout นานเกินไป

# ❌ ผิด: Timeout นานเกินไป ทำให้รอนานเมื่อ service ล่ม
breaker = CircuitBreaker(
    config=CircuitBreakerConfig(
        failure_threshold=1,      # ล้มเหลวแค่ครั้งเดียวก็เปิด circuit
        timeout_duration=300.0,   # รอ 5 นาที!
        success_threshold=1
    )
)

✅ ถูกต้อง: สมดุลระหว่างป้องกันและการกู้คืน
breaker = CircuitBreaker(
    config=CircuitBreakerConfig(
        failure_threshold=5,      # ล้มเหลว 5 ครั้งติดกัน
        timeout_duration=30.0,    # รอ 30 วินาทีก่อนลองใหม่
        success_threshold=3,      # ต้องสำเร็จ 3 ครั้งถึงปิด circuit
        half_open_max_calls=2     # ทดสอบได้ 2 ครั้งใน half-open
    )
)

3. ไม่จัดการ HTTP Status Codes อย่างถูกต้อง

# ❌ ผิด: raise_for_status() จะ raise exception เสมอ
async def bad_call():
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=payload)
        response.raise_for_status()  # 500 error ก็ raise!
        return response.json()

✅ ถูกต้อง: ตรวจสอบ status code ก่อน
async def good_call():
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=payload)
        
        # 500/502/503 = server error = record failure
        # 429 = rate limit = record failure + wait
        # 401/403 = auth error = ไม่ควร retry
        # 400 = bad request = ไม่ควร retry
        
        if response.status_code >= 500:
            # Server error - ควร retry
            raise AI ServiceError(f"Server error: {response.status_code}")
        elif response.status_code == 429:
            # Rate limit - รอตาม Retry-After header
            retry_after = int(response.headers.get("Retry-After", 60))
            await asyncio.sleep(retry_after)
            raise RateLimitError("Rate limit exceeded")
        elif response.status_code >= 400:
            # Client error - ไม่ควร retry
            raise ValueError(f"Request error: {response.status_code}")
        
        return response.json()

✅ ดีที่สุด: ตรวจสอบแบบละเอียด
async def robust_call():
    try:
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                f"{BASE_URL}/chat/completions",
                headers=HEADERS,
                json={"model": "gpt-4.1", "messages": [...]}
            )
            
            # Timeout เกิดที่ client level
            if response.status_code == 0:
                raise ConnectionError("Connection timeout")
            
            data = response.json()
            
            # AI API error (non-2xx HTTP but valid JSON response)
            if "error" in data:
                error_type = data["error"].get("type", "")
                if error_type in ["rate_limit", "server_error"]:
                    raise AI ServiceError(data["error"]["message"])
            
            return data
            
    except httpx.ConnectTimeout:
        raise ConnectionError("Connection timeout to AI service")
    except httpx.ReadTimeout:
        raise ConnectionError("AI service response timeout")

4. Memory Leak จากไม่ reset state

# ❌ ผิด: failure_count ไม่ถูก reset เมื่อสำเร็จ
class BadCircuitBreaker:
    def record_success(self):
        # ไม่ได้ reset failure_count
        self.last_success = time.time()

✅ ถูกต้อง: reset counters ทั้งหมดเมื่อปิด circuit
class GoodCircuitBreaker:
    def record_success(self):
        self.success_count += 1
        
        if self.state == CircuitState.HALF_OPEN:
            if self.success_count >= self.config.success_threshold:
                # Transition to CLOSED - reset ทุกอย่าง
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                self.success_count = 0
                self.half_open_calls = 0
                self.last_state_change = time.time()
                logger.info("Circuit closed - service recovered")

สรุป

Circuit Breaker pattern เป็น essential pattern สำหรับระบบที่พึ่งพา AI services ภายนอก ช่วยป้องกัน cascade failure และทำให้ระบบมีความ resilient มากขึ้น จากประสบการณ์ตรงในการ implement ระบบ production ที่ใช้ HolySheep AI ซึ่งมีอัตรา ¥1=$1 (ประหยัด 85%+) ร่วมกับ Circuit Breaker ทำให้ระบบสามารถ:

• รักษา uptime ได้ >99.9% แม้ AI service จะมีปัญหาชั่วคราว
• ลด cost จากการ retry ที่ไม่จำเป็น
• ให้ fallback response ที่เหมาะสมกับผู้ใช้
• Monitor และ alert ได้เมื่อ service health ลดลง

ราคา 2026/MTok ของ HolySheheep: GPT-4.1 $8, Claude Sonnet 4.5 $15, Gemini 2.5 Flash $2.50, DeepSeek V3.2 $0.42 ซึ่งถูกกว่าผู้ให้บริการอื่นๆ มาก และรองรับการชำระเงินผ่าน WeChat/Alipay พร้อม <50ms latency 👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน