เมื่อวันที่ 15 มีนาคม 2024 เวลาประมาณ 14:32 น. ระบบ API ที่ผมใช้งานอยู่เกิดปัญหา ConnectionError: timeout ส่งผลให้การประมวลผล AI ทั้งระบบหยุดชะงัก สถานการณ์นี้ทำให้ผมตระหนักว่าการมี แผนสำรองข้อมูล สำหรับ AI API เป็นสิ่งจำเป็นอย่างยิ่ง ในบทความนี้ ผมจะแบ่งปันประสบการณ์ตรงและโค้ดที่ใช้งานได้จริงในการสร้าง ระบบ Failover อัตโนมัติ สำหรับ AI API
ทำไมต้องมีแผนสำรองสำหรับ AI API?
ในปัจจุบัน ระบบ AI API เช่น HolySheep AI มีความเสถียรสูง แต่ทุกระบบย่อมมีโอกาสเกิดข้อผิดพลาด ปัจจัยที่ทำให้ระบบหยุดทำงานมีหลายประการ ได้แก่ ปัญหาเครือข่าย (NetworkError: Connection refused) การหมดเวลาการเชื่อมต่อ (TimeoutError: timed out) การหมดอายุโทเค็น (401 Unauthorized) และปัญหา Rate Limit (429 Too Many Requests) การมี ระบบสำรองอัตโนมัติ จะช่วยให้แอปพลิเคชันของคุณทำงานต่อเนื่องได้แม้ API หลักเกิดปัญหา
โครงสร้างพื้นฐานของระบบ Failover
ระบบ Failover ที่ดีต้องมีองค์ประกอบหลัก 3 ส่วน คือ Health Check ที่ตรวจสอบสถานะของ API อย่างต่อเนื่อง Automatic Switch ที่สลับไปใช้ API สำรองเมื่อ API หลักเกิดปัญหา และ Logging & Alerting ที่บันทึกข้อผิดพลาดและแจ้งเตือนผู้ดูแลระบบ โค้ดต่อไปนี้เป็นตัวอย่างระบบ Failover ที่ใช้งานได้จริง
"""
AI API Failover System - ระบบสำรอง API อัตโนมัติ
สร้างเมื่อ: 2024-03-15
เวอร์ชัน: 2.0.0
"""
import time
import logging
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from enum import Enum
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class APIStatus(Enum):
"""สถานะของ API"""
HEALTHY = "healthy"
DEGRADED = "degraded"
DOWN = "down"
UNKNOWN = "unknown"
@dataclass
class APIConfig:
"""การตั้งค่าสำหรับ API endpoint"""
name: str
base_url: str
api_key: str
timeout: float = 30.0
max_retries: int = 3
weight: int = 1 # น้ำหนักสำหรับการจัดสรรงาน
@dataclass
class APIHealth:
"""ข้อมูลสุขภาพของ API"""
status: APIStatus = APIStatus.UNKNOWN
latency_ms: float = 0.0
last_check: float = 0.0
consecutive_failures: int = 0
total_requests: int = 0
failed_requests: int = 0
class AIFailoverManager:
"""ตัวจัดการระบบ Failover สำหรับ AI API"""
def __init__(self, health_check_interval: int = 30):
self.apis: Dict[str, APIConfig] = {}
self.health: Dict[str, APIHealth] = {}
self.health_check_interval = health_check_interval
self.current_primary: Optional[str] = None
self.session = self._create_session()
def _create_session(self) -> requests.Session:
"""สร้าง requests session พร้อม retry strategy"""
session = requests.Session()
retry_strategy = Retry(
total=0, # เราจะจัดการ retry เอง
backoff_factor=0,
status_forcelist=[],
allowed_methods=[]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
def add_api(self, config: APIConfig) -> None:
"""เพิ่ม API ใหม่เข้าสู่ระบบ"""
self.apis[config.name] = config
self.health[config.name] = APIHealth()
logger.info(f"เพิ่ม API: {config.name} ({config.base_url})")
def check_api_health(self, api_name: str) -> APIHealth:
"""ตรวจสอบสุขภาพของ API"""
if api_name not in self.apis:
logger.warning(f"ไม่พบ API: {api_name}")
return APIHealth(status=APIStatus.UNKNOWN)
api = self.apis[api_name]
health = self.health[api_name]
try:
start_time = time.time()
response = self.session.get(
f"{api.base_url}/models",
headers={"Authorization": f"Bearer {api.api_key}"},
timeout=api.timeout
)
health.latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
health.status = APIStatus.HEALTHY
health.consecutive_failures = 0
logger.info(f"{api_name}: สถานะปกติ (latency: {health.latency_ms:.2f}ms)")
elif response.status_code == 401:
health.status = APIStatus.DOWN
health.consecutive_failures += 1
logger.error(f"{api_name}: 401 Unauthorized - API key ไม่ถูกต้อง")
elif response.status_code == 429:
health.status = APIStatus.DEGRADED
logger.warning(f"{api_name}: Rate limit exceeded")
else:
health.status = APIStatus.DEGRADED
health.consecutive_failures += 1
logger.warning(f"{api_name}: สถานะ degraded (HTTP {response.status_code})")
except requests.exceptions.Timeout:
health.status = APIStatus.DOWN
health.consecutive_failures += 1
health.latency_ms = api.timeout * 1000
logger.error(f"{api_name}: Connection timeout ({api.timeout}s)")
except requests.exceptions.ConnectionError as e:
health.status = APIStatus.DOWN
health.consecutive_failures += 1
logger.error(f"{api_name}: Connection refused - {str(e)}")
except Exception as e:
health.status = APIStatus.UNKNOWN
health.consecutive_failures += 1
logger.error(f"{api_name}: ข้อผิดพลาดไม่ทราบสาเหตุ - {str(e)}")
health.last_check = time.time()
return health
def get_available_api(self) -> Optional[str]:
"""ดึง API ที่พร้อมใช้งาน"""
for name, health in self.health.items():
if health.status == APIStatus.HEALTHY:
return name
# ถ้าไม่มี API ที่ healthy ให้ลองใช้ degraded
for name, health in self.health.items():
if health.status == APIStatus.DEGRADED:
return name
return None
def call_api(self, prompt: str, model: str = "gpt-4") -> Dict[str, Any]:
"""เรียกใช้ AI API พร้อมระบบ Failover"""
available_api = self.get_available_api()
if not available_api:
return {
"success": False,
"error": "ไม่มี API ที่พร้อมใช้งาน",
"code": "NO_API_AVAILABLE"
}
api = self.apis[available_api]
health = self.health[available_api]
health.total_requests += 1
try:
response = self.session.post(
f"{api.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {api.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}]
},
timeout=api.timeout
)
if response.status_code == 200:
return {
"success": True,
"data": response.json(),
"api_used": available_api,
"latency_ms": health.latency_ms
}
elif response.status_code == 401:
health.failed_requests += 1
health.status = APIStatus.DOWN
logger.error(f"API {available_api} ได้รับ 401 - ตัดออกจากระบบ")
# ลองเรียก API อื่น
return self.call_api(prompt, model)
else:
return {
"success": False,
"error": response.text,
"code": response.status_code,
"api_used": available_api
}
except Exception as e:
health.failed_requests += 1
logger.error(f"เรียก API {available_api} ล้มเหลว: {str(e)}")
return self.call_api(prompt, model)
การใช้งาน
if __name__ == "__main__":
manager = AIFailoverManager()
# เพิ่ม HolySheep API เป็น API หลัก
manager.add_api(APIConfig(
name="holysheep",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
timeout=30.0,
weight=3
))
# เพิ่ม API สำรองอื่นๆ
manager.add_api(APIConfig(
name="backup-api",
base_url="https://api.backup.example.com/v1",
api_key="BACKUP_API_KEY",
timeout=30.0,
weight=1
))
# ตรวจสอบสุขภาพทั้งหมด
for api_name in manager.apis.keys():
manager.check_api_health(api_name)
# เรียกใช้งาน
result = manager.call_api("ทดสอบระบบ Failover")
print(result)ระบบ Health Check และ Auto-Switch
การตรวจสอบสุขภาพของ API อย่างสม่ำเสมอเป็นหัวใจสำคัญของระบบ Failover ในโค้ดต่อไปนี้ ผมจะสร้าง Health Monitor ที่ทำงานเป็นเธรดแยกต่างหาก พร้อมระบบแจ้งเตือนเมื่อ API หลักเกิดปัญหา ระบบนี้ใช้งานได้กับ HolySheep AI ซึ่งมีความเสถียรสูงและมี latency ต่ำกว่า 50ms ทำให้เหมาะสำหรับใช้เป็น API หลัก
"""
Health Monitor และ Auto-Switch System
ระบบตรวจสอบสุขภาพและสลับ API อัตโนมัติ
"""
import threading
import time
import queue
from datetime import datetime
from typing import Callable, Optional
from dataclasses import dataclass, asdict
import json
@dataclass
class AlertMessage:
"""ข้อความแจ้งเตือน"""
timestamp: str
severity: str # info, warning, error, critical
api_name: str
message: str
details: dict
class HealthMonitor:
"""ระบบตรวจสอบสุขภาพ API แบบ Real-time"""
def __init__(
self,
check_interval: int = 30,
failure_threshold: int = 3,
recovery_threshold: int = 2
):
self.check_interval = check_interval
self.failure_threshold = failure_threshold
self.recovery_threshold = recovery_threshold
self.alert_queue = queue.Queue()
self.is_running = False
self.monitor_thread: Optional[threading.Thread] = None
self.api_manager = None
self.callbacks: list = []
def set_api_manager(self, manager) -> None:
"""ตั้งค่า API Manager"""
self.api_manager = manager
def register_alert_callback(self, callback: Callable[[AlertMessage], None]) -> None:
"""ลงทะเบียน callback สำหรับแจ้งเตือน"""
self.callbacks.append(callback)
def _send_alert(self, api_name: str, severity: str, message: str, details: dict = None) -> None:
"""ส่งข้อความแจ้งเตือน"""
alert = AlertMessage(
timestamp=datetime.now().isoformat(),
severity=severity,
api_name=api_name,
message=message,
details=details or {}
)
self.alert_queue.put(alert)
for callback in self.callbacks:
try:
callback(alert)
except Exception as e:
print(f"Alert callback error: {e}")
def _monitor_loop(self) -> None:
"""ลูปหลักสำหรับตรวจสอบสุขภาพ"""
while self.is_running:
if self.api_manager is None:
time.sleep(self.check_interval)
continue
for api_name in self.api_manager.apis.keys():
health = self.api_manager.check_api_health(api_name)
# ตรวจสอบว่า API ล่ม
if health.consecutive_failures >= self.failure_threshold:
if health.status.value != "down":
self._send_alert(
api_name=api_name,
severity="critical",
message=f"API {api_name} หยุดทำงานหลังจาก {health.consecutive_failures} ครั้ง",
details={
"latency_ms": health.latency_ms,
"failure_count": health.consecutive_failures,
"total_requests": health.total_requests,
"failed_requests": health.failed_requests
}
)
# ตรวจสอบว่า API กลับมาทำงาน
elif health.consecutive_failures <= self.recovery_threshold and health.status.value == "healthy":
self._send_alert(
api_name=api_name,
severity="info",
message=f"API {api_name} กลับมาทำงานปกติ",
details={
"latency_ms": health.latency_ms,
"recovery_time": health.last_check
}
)
# ตรวจสอบว่า API ช้าลง
if health.latency_ms > 5000: # เกิน 5 วินาที
self._send_alert(
api_name=api_name,
severity="warning",
message=f"API {api_name} ตอบสนองช้า ({health.latency_ms:.0f}ms)",
details={"latency_ms": health.latency_ms}
)
time.sleep(self.check_interval)
def start(self) -> None:
"""เริ่มระบบตรวจสอบ"""
if self.is_running:
return
self.is_running = True
self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
self.monitor_thread.start()
print(f"Health Monitor เริ่มทำงาน (ตรวจสอบทุก {self.check_interval} วินาที)")
def stop(self) -> None:
"""หยุดระบบตรวจสอบ"""
self.is_running = False
if self.monitor_thread:
self.monitor_thread.join(timeout=5)
print("Health Monitor หยุดทำงาน")
class TelegramAlertHandler:
"""ตัวส่งแจ้งเตือนผ่าน Telegram"""
def __init__(self, bot_token: str, chat_id: str):
self.bot_token = bot_token
self.chat_id = chat_id
self.api_url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
def send_alert(self, alert: AlertMessage) -> bool:
"""ส่งข้อความแจ้งเตือน"""
emoji = {
"info": "ℹ️",
"warning": "⚠️",
"error": "❌",
"critical": "🚨"
}
message = f"""
{emoji.get(alert.severity, "📢")} *AI API Alert*
*API:* {alert.api_name}
*ความรุนแรง:* {alert.severity.upper()}
*เวลา:* {alert.timestamp}
*ข้อความ:* {alert.message}
"""
if alert.details:
message += f"\n*รายละเอียด:*\n``json\n{json.dumps(alert.details, indent=2, default=str)}\n``"
try:
import requests
response = requests.post(
self.api_url,
json={
"chat_id": self.chat_id,
"text": message,
"parse_mode": "Markdown"
},
timeout=10
)
return response.status_code == 200
except Exception as e:
print(f"Telegram send failed: {e}")
return False
class SlackAlertHandler:
"""ตัวส่งแจ้งเตือนผ่าน Slack"""
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
def send_alert(self, alert: AlertMessage) -> bool:
"""ส่งข้อความแจ้งเตือนไป Slack"""
color = {
"info": "#36a64f",
"warning": "#ff9900",
"error": "#ff0000",
"critical": "#aa0000"
}
payload = {
"attachments": [{
"color": color.get(alert.severity, "#cccccc"),
"fields": [
{"title": "API", "value": alert.api_name, "short": True},
{"title": "Severity", "value": alert.severity.upper(), "short": True},
{"title": "Message", "value": alert.message, "short": False}
],
"footer": "AI Failover System",
"ts": datetime.now().timestamp()
}]
}
try:
import requests
response = requests.post(
self.webhook_url,
json=payload,
timeout=10
)
return response.status_code == 200
except Exception as e:
print(f"Slack send failed: {e}")
return False
การใช้งาน
if __name__ == "__main__":
from failover import AIFailoverManager, APIConfig
# สร้างระบบ Failover
manager = AIFailoverManager()
manager.add_api(APIConfig(
name="holysheep-primary",
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
timeout=30.0
))
# สร้าง Health Monitor
monitor = HealthMonitor(
check_interval=30,
failure_threshold=3,
recovery_threshold=2
)
monitor.set_api_manager(manager)
# เพิ่ม Telegram Alert
# telegram_handler = TelegramAlertHandler("BOT_TOKEN", "CHAT_ID")
# monitor.register_alert_callback(telegram_handler.send_alert)
# เพิ่ม Slack Alert
# slack_handler = SlackAlertHandler("https://hooks.slack.com/...")
# monitor.register_alert_callback(slack_handler.send_alert)
# เริ่มตรวจสอบ
monitor.start()
# ทดสอบการแจ้งเตือน
monitor._send_alert(
api_name="holysheep-primary",
severity="info",
message="ระบบเริ่มทำงาน",
details={"version": "2.0.0"}
)ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
จากประสบการณ์ในการใช้งาน AI API มาหลายปี ผมพบข้อผิดพลาดที่เกิดขึ้นบ่อยครั้งและมีวิธีแก้ไขที่ชัดเจน ต่อไปนี้คือ 3 กรณีหลัก ที่ควรทราบ
กรณีที่ 1: ConnectionError: Connection refused
ข้อผิดพลาดนี้เกิดขึ้นเมื่อไม่สามารถเชื่อมต่อกับ API endpoint ได้ สาเหตุหลักมักมาจาก URL ผิดพลาด หรือเซิร์ฟเวอร์ปิดให้บริการชั่วคราว วิธีแก้ไขคือตรวจสอบ base_url ให้ถูกต้อง และใช้ระบบ Failover เพื่อสลับไปใช้ API สำรอง ดังตัวอย่างโค้ดต่อไปนี้
"""
กรณีที่ 1: ConnectionError: Connection refused
สาเหตุ: ไม่สามารถเชื่อมต่อ API endpoint ได้
วิธีแก้ไข: ตรวจสอบ URL และใช้ Fallback
"""
import requests
from requests.exceptions import ConnectionError, Timeout, HTTPError
class RobustAPIClient:
"""Client ที่มีระบบ Fallback หลายชั้น"""
def __init__(self):
self.primary_url = "https://api.holysheep.ai/v1"
self.fallback_urls = [
"https://backup.holysheep.ai/v1",
"https://api.v2.holysheep.ai/v1"
]
self.api_key = "YOUR_HOLYSHEEP_API_KEY"
self.timeout = 30
def call_with_fallback(self, endpoint: str, payload: dict) -> dict:
"""เรียก API พร้อม Fallback หลายระดับ"""
urls_to_try = [self.primary_url] + self.fallback_urls
last_error = None
for url in urls_to_try:
try:
full_url = f"{url}/{endpoint}"
response = requests.post(
full_url,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json=payload,
timeout=self.timeout
)
# ตรวจสอบ HTTP Status
if response.status_code == 200:
return {
"success": True,
"data": response.json(),
"url_used": url
}
elif response.status_code == 401:
# API Key ผิดพลาด - ไม่ต้องลอง URL อื่น
return {
"success": False,
"error": "API Key ไม่ถูกต้อง",
"code": "INVALID_API_KEY"
}
elif response.status_code == 429:
# Rate Limit - ลอง URL ถัดไป
last_error = f"Rate limit on {url}"
continue
else:
last_error = f"HTTP {response.status_code}: {response.text}"
continue
except ConnectionError as e:
last_error = f"Connection refused: {str(e)}"
print(f"❌ เชื่อมต่อ {url} ไม่ได้: {last_error}")
continue
except Timeout as e:
last_error = f"Timeout: {str(e)}"
print(f"⏰ {url} ตอบสนองช้าเกินไป: {last_error}")
continue
except Exception as e:
last_error = f"Unexpected error: {str(e)}"
print(f"⚠️ {url} ข้อผิดพลาดไม่คาดคิด: {last_error}")
continue
# ถ้าลองหมดทุก URL แล้ว
return {
"success": False,
"error": f"ทุก API endpoint ล้มเหลว. ข้อผิดพลาดล่าสุด: {last_error}",
"code": "ALL_ENDPOINTS_FAILED",
"urls_tried": urls_to_try
}
def chat_completion(self, messages: list, model: str = "gpt-4") -> dict:
"""ส่งข้อความแชทและรับการตอบกลับ"""
payload = {
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2000
}
result = self.call_with_fallback("chat/completions", payload)
if result["success"]:
return result["data"]
else:
# ส่งข้อความแจ้งเตือน
print(f"🚨 API Call Failed: {result['error']}")
# ส่งคืนข้อความแจ้งเตือนแทน
return {
"error": True,
"message": "ขออภัย ไม่สามารถประมวลผลคำขอได้ในขณะนี้",
"technical_detail": result["error"]
}
ทดสอบการใช้งาน
if __name__ == "__main__":
client = RobustAPIClient()
# ทดสอบกรณีปกติ
response = client.chat_completion([
{"role": "user", "content": "ทดสอบระบบ Fallback"}
])
print("ผลลัพธ์:", response)กรณีที่ 2: 401 Unauthorized - API Key หมดอายุ
ข้อผิดพลาดนี้เกิดขึ้นเมื่อ API Key ไม่ถูกต้องหรือหมดอายุ สาเหตุอาจมาจาก Key ถูก Revoke หรือมีการเปลี่ยนแปลงสิทธิ์การเข้าถึง วิธีแก้ไขคือตรวจสอบ API Key และอัปเดต รวมถึงตั้งค่า Key Rotation อัตโนมัติ ด้านล่างนี้คือโค้ดสำหรับ API Key Rotation ที่ใช้งานได้จริง
"""
กรณีที่ 2: 401 Unauthorized - API Key Rotation
สาเหตุ: API Key หมดอายุ ถูก Revoke หรือไม่ถูกต้อง
วิธีแก้ไข: ระบบ Key Rotation อัตโนมัติ
"""
import time
import threading
from typing import List, Optional, Dict
from dataclasses import dataclass
import requests
@dataclass
class APIKey:
"""ข้อมูล API Key"""
key: str
name: str
created_at