ในยุคที่ AI API กลายเป็นหัวใจสำคัญของแอปพลิเคชัน การจัดการความล้มเหลวอย่างชาญฉลาดเป็นสิ่งที่ทีมพัฒนาทุกคนต้องเผชิญ วันนี้เราจะมาเรียนรู้วิธีติดตั้ง Circuit Breaker Pattern ที่ช่วยให้ระบบของคุณทำงานได้อย่างเสถียรแม้ในสถานการณ์ที่ API ตอบสนองช้าหรือล่ม
กรณีศึกษา: ทีมสตาร์ทอัพ AI ในกรุงเทพฯ
ทีมพัฒนาแพลตฟอร์ม Chatbot อัจฉริยะแห่งหนึ่งในกรุงเทพฯ มีจุดเจ็บปวดหลักคือ ระบบต้องเรียกใช้ AI API จากหลายผู้ให้บริการ (OpenAI, Anthropic, Google) เพื่อให้บริการแชทบอทที่รองรับหลายภาษา
ปัญหาที่พบคือ เมื่อ API ตัวใดตัวหนึ่งตอบสนองช้าเกิน 5 วินาที ระบบทั้งหมดจะค้าง ทำให้ผู้ใช้งานรอนานและบางครั้งเกิด Timeout Error สูงถึง 30% ของคำขอทั้งหมด ค่าใช้จ่ายรายเดือนพุ่งสูงถึง $4,200 เพราะระบบพยายามเรียก API ที่ล่มซ้ำแล้วซ้ำเล่า
หลังจากที่ทีมได้ทดลองใช้ HolySheep AI ซึ่งรวม API ของหลายผู้ให้บริการเข้าด้วยกัน พร้อมทั้งมีฟีเจอร์ Circuit Breaker ในตัว ผลลัพธ์ที่ได้คือ ค่าเฉลี่ยความหน่วง (Latency) ลดลงจาก 420ms เหลือ 180ms และค่าใช้จ่ายรายเดือนลดลงเหลือ $680 ประหยัดได้ถึง 83%
Circuit Breaker Pattern คืออะไร
Circuit Breaker เป็น Design Pattern ที่ทำหน้าที่เหมือนฟิวส์ไฟฟ้า เมื่อระบบตรวจพบว่า API ตัวใดตอบสนองผิดพลาดติดต่อกันหลายครั้ง ระบบจะ "ตัดวงจร" ชั่วคราว ไม่เรียก API ตัวนั้นอีก แต่จะส่งต่อคำขอไปยัง API สำรองแทน เป็นการป้องกันไม่ให้ระบบทั้งหมดล่มจากปัญหาเพียงจุดเดียว
การติดตั้ง Circuit Breaker กับ HolySheep AI
HolySheep AI มี endpoint เดียวที่รวมทุก Model ไว้แล้ว ทำให้การติดตั้ง Circuit Breaker ง่ายมาก มาดูตัวอย่างการใช้งานกัน
// การติดตั้ง Circuit Breaker ด้วย Python
import requests
import time
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Dict, Any
class CircuitState(Enum):
CLOSED = "closed" # ปกติ ทำงานได้
OPEN = "open" # ตัดวงจร รอการกู้คืน
HALF_OPEN = "half_open" # ทดสอบการกู้คืน
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # จำนวนครั้งที่ล้มเหลวก่อนตัดวงจร
recovery_timeout: int = 30 # วินาทีที่รอก่อนลองใหม่
half_open_max_calls: int = 3 # จำนวนครั้งที่ทดสอบในโหมด half-open
success_threshold: int = 2 # ครั้งที่ต้องสำเร็จก่อนเปิดวงจร
class CircuitBreaker:
def __init__(self, config: CircuitBreakerConfig):
self.config = config
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: Optional[float] = None
self.half_open_calls = 0
def call(self, func, *args, **kwargs):
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
else:
raise Exception("Circuit Breaker is OPEN - API temporarily unavailable")
try:
result = func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise e
def _should_attempt_reset(self) -> bool:
if self.last_failure_time is None:
return True
return (time.time() - self.last_failure_time) >= self.config.recovery_timeout
def _on_success(self):
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self.state = CircuitState.CLOSED
self.success_count = 0
print("Circuit Breaker RESET to CLOSED state")
def _on_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
print("Circuit Breaker OPENED from HALF_OPEN state")
elif self.failure_count >= self.config.failure_threshold:
self.state = CircuitState.OPEN
print(f"Circuit Breaker OPENED after {self.failure_count} failures")
การใช้งานกับ HolySheep API
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
circuit_breaker = CircuitBreaker(CircuitBreakerConfig())
def call_ai_model(model: str, prompt: str) -> Dict[str, Any]:
def _make_request():
response = requests.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}]
},
timeout=10
)
response.raise_for_status()
return response.json()
return circuit_breaker.call(_make_request)
ตัวอย่างการใช้งาน
try:
result = call_ai_model("gpt-4.1", "อธิบายเรื่อง Circuit Breaker Pattern")
print(result)
except Exception as e:
print(f"Request failed: {e}")
# Fallback ไปยัง logic อื่นMulti-Model Fallback Strategy
นอกจาก Circuit Breaker แล้ว การมี Fallback Strategy ที่ดีจะช่วยให้ระบบทำงานได้อย่างต่อเนื่อง ตัวอย่างด้านล่างแสดงการตั้งค่า Priority-based Fallback ที่จะเรียก Model สำรองหาก Model หลักไม่สามารถใช้งานได้
// TypeScript Implementation พร้อม Multi-Model Fallback
interface AIModelConfig {
name: string;
priority: number;
maxLatency: number; // milliseconds
}
interface CircuitBreakerState {
failures: number;
lastFailure: number;
isOpen: boolean;
}
class MultiModelRouter {
private models: AIModelConfig[] = [
{ name: "gpt-4.1", priority: 1, maxLatency: 3000 },
{ name: "claude-sonnet-4.5", priority: 2, maxLatency: 4000 },
{ name: "gemini-2.5-flash", priority: 3, maxLatency: 2000 },
{ name: "deepseek-v3.2", priority: 4, maxLatency: 2500 }
];
private circuitBreakers: Map<string, CircuitBreakerState> = new Map();
private readonly FAILURE_THRESHOLD = 5;
private readonly RECOVERY_TIMEOUT = 30000; // 30 seconds
constructor() {
// Initialize circuit breakers for each model
this.models.forEach(model => {
this.circuitBreakers.set(model.name, {
failures: 0,
lastFailure: 0,
isOpen: false
});
});
}
async chatCompletion(prompt: string): Promise<any> {
const availableModels = this.getAvailableModels();
for (const model of availableModels) {
try {
const result = await this.callWithTimeout(
model.name,
prompt,
model.maxLatency
);
this.resetCircuitBreaker(model.name);
return result;
} catch (error) {
console.error(Model ${model.name} failed:, error.message);
this.recordFailure(model.name);
if (this.isCircuitOpen(model.name)) {
console.log(Circuit breaker OPEN for ${model.name});
continue;
}
}
}
throw new Error("All models unavailable");
}
private async callWithTimeout(
modelName: string,
prompt: string,
timeout: number
): Promise<any> {
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), timeout);
try {
const response = await fetch("https://api.holysheep.ai/v1/chat/completions", {
method: "POST",
headers: {
"Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
body: JSON.stringify({
model: modelName,
messages: [{ role: "user", content: prompt }]
}),
signal: controller.signal
});
clearTimeout(timeoutId);
if (!response.ok) {
throw new Error(HTTP ${response.status});
}
return await response.json();
} catch (error) {
clearTimeout(timeoutId);
throw error;
}
}
private getAvailableModels(): AIModelConfig[] {
return this.models
.filter(m => !this.isCircuitOpen(m.name))
.sort((a, b) => a.priority - b.priority);
}
private isCircuitOpen(modelName: string): boolean {
const cb = this.circuitBreakers.get(modelName);
if (!cb || !cb.isOpen) return false;
const now = Date.now();
if (now - cb.lastFailure >= this.RECOVERY_TIMEOUT) {
cb.isOpen = false;
cb.failures = 0;
return false;
}
return true;
}
private recordFailure(modelName: string): void {
const cb = this.circuitBreakers.get(modelName)!;
cb.failures++;
cb.lastFailure = Date.now();
if (cb.failures >= this.FAILURE_THRESHOLD) {
cb.isOpen = true;
console.log(Circuit breaker opened for ${modelName});
}
}
private resetCircuitBreaker(modelName: string): void {
const cb = this.circuitBreakers.get(modelName)!;
cb.failures = 0;
cb.isOpen = false;
}
getHealthStatus(): object {
const status: any = {};
this.models.forEach(model => {
const cb = this.circuitBreakers.get(model.name)!;
status[model.name] = {
failures: cb.failures,
isOpen: cb.isOpen,
available: !cb.isOpen
};
});
return status;
}
}
// การใช้งาน
const router = new MultiModelRouter();
async function main() {
try {
const result = await router.chatCompletion("สวัสดีครับ");
console.log("Response:", result);
} catch (error) {
console.error("All models failed:", error.message);
}
// ตรวจสอบสถานะสุขภาพของทุก Model
console.log("Health Status:", router.getHealthStatus());
}
main();ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. ปัญหา Timeout ไม่ทำงานตามที่ตั้งค่า
สาเหตุ: การตั้งค่า timeout ใน Python ด้วย requests.post(timeout=10) ไม่สามารถยกเลิก request ที่กำลังดำเนินอยู่ได้ทันที เมื่อ timeout ครบ แต่การเชื่อมต่อยังคงเปิดอยู่
# ❌ วิธีที่ผิด - timeout ไม่ยกเลิกการเชื่อมต่อจริง
response = requests.post(url, json=data, timeout=10)
✅ วิธีที่ถูก - ใช้ signal-based timeout หรือ session
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_timeout():
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
return session
หรือใช้วิธี Manual timeout control
import signal
def timeout_handler(signum, frame):
raise TimeoutError("Request took too long")
def call_with_real_timeout(url, data, timeout_seconds=10):
# ตั้งค่า signal handler
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(timeout_seconds)
try:
response = requests.post(url, json=data)
signal.alarm(0) # ยกเลิก alarm
return response.json()
finally:
signal.alarm(0) # ทำความสะอาดในกรณี exception
หรือใช้ aiohttp สำหรับ async operations
import aiohttp
async def async_call_with_real_timeout(url, data, timeout_ms=10000):
timeout = aiohttp.ClientTimeout(total=timeout_ms / 1000)
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.post(url, json=data) as response:
return await response.json()2. ปัญหา Race Condition ใน Circuit Breaker
สาเหตุ: เมื่อมี requests หลายตัวพร้อมกันและ Circuit Breaker เปิดขึ้น state อาจไม่สอดคล้องกันระหว่าง threads
# ❌ วิธีที่ผิด - ไม่มีการล็อกเมื่อแก้ไข state
class BrokenCircuitBreaker:
def __init__(self):
self.state = "closed"
self.failure_count = 0
def record_failure(self):
self.failure_count += 1 # Race condition!
if self.failure_count >= 5:
self.state = "open"
✅ วิธีที่ถูก - ใช้ Lock สำหรับ thread safety
import threading
from enum import Enum
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class ThreadSafeCircuitBreaker:
def __init__(self, failure_threshold=5):
self._lock = threading.RLock()
self._state = CircuitState.CLOSED
self._failure_count = 0
self._failure_threshold = failure_threshold
self._last_failure_time = None
@property
def state(self):
with self._lock:
return self._state
def record_failure(self):
with self._lock:
self._failure_count += 1
self._last_failure_time = time.time()
if self._failure_count >= self._failure_threshold:
self._state = CircuitState.OPEN
print(f"Circuit opened after {self._failure_count} failures")
def record_success(self):
with self._lock:
if self._state == CircuitState.HALF_OPEN:
self._state = CircuitState.CLOSED
self._failure_count = 0
print("Circuit closed - service recovered")
def can_execute(self):
with self._lock:
if self._state == CircuitState.CLOSED:
return True