ในฐานะ Senior Backend Engineer ที่ดูแลระบบ AI หลายตัวในองค์กร ผมเจอปัญหา API failure จนชินมือแล้ว วันนี้เลยจะมาแชร์วิธีตั้งระบบ Monitoring และ Alert ที่ใช้งานจริงกับ HolySheep AI โดยเฉพาะ ใครที่กำลังใช้งาน AI API แล้วเจอปัญหา Rate Limit, Server Error หรือ Service Unavailable บ่อยๆ บทความนี้จะช่วยได้แน่นอน
ทำไมต้อง Monitor AI API Errors?
จากประสบการณ์ที่ผมดูแลระบบ RAG ขององค์กรขนาดใหญ่ พบว่า API errors เหล่านี้ส่งผลกระทบต่อธุรกิจโดยตรง:
- Error 429 (Rate Limit Exceeded): เกิดเมื่อเรียกใช้ API บ่อยเกินกว่าที่กำหนด ทำให้ระบบ Chatbot หยุดตอบลูกค้า
- Error 502 (Bad Gateway): เซิร์ฟเวอร์ปลายทางมีปัญหา มักเกิดตอน Peak hours ของวัน
- Error 503 (Service Unavailable): AI service ปิดซ่อมบำรุงหรือ Overload
ระบบที่ผมพัฒนาให้กับลูกค้าอีคอมเมิร์ซรายใหญ่ เคยเสียหายหลายแสนบาทเพราะ AI chatbot ล่ม 2-3 ชั่วโมงโดยไม่มีใครรู้ จนกระทั่งลูกค้าติดต่อเข้ามาเยอะมากถึงได้รู้ ดังนั้นการมีระบบ Monitor และ Alert ที่ดีจึงไม่ใช่ทางเลือก แต่เป็นความจำเป็น
สถาปัตยกรรมระบบ Monitoring สำหรับ HolySheep AI
ก่อนจะเข้าสู่โค้ด ผมอยากให้เข้าใจสถาปัตยกรรมโดยรวมก่อน:
┌─────────────────────────────────────────────────────────────────┐
│ HolySheep AI Monitoring Stack │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────────┐ ┌─────────────────────┐ │
│ │ Application│──▶│ Retry Layer │──▶│ HolySheep API │ │
│ │ Service │ │ (Exponential │ │ https://api.holy- │ │
│ │ │ │ Backoff) │ │ sheep.ai/v1 │ │
│ └──────────┘ └──────────────┘ └─────────────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────┐ ┌──────────────┐ │
│ │ Metrics │◀─────────────────────────│ Error Tracker │ │
│ │ Collector│ │ │ │
│ └──────────┘ └──────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ Alert Manager (Discord/Slack/Line) │ │
│ └──────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘การตั้งค่า Client พื้นฐาน พร้อม Retry Logic
เริ่มจากการตั้งค่า HTTP Client ที่มีระบบ Retry อัตโนมัติกับ HolySheep AI กันก่อน:
import httpx
import asyncio
import logging
from typing import Optional, Dict, Any
from datetime import datetime, timedelta
ตั้งค่า Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
คอนฟิกสำหรับ HolySheep AI
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY", # แทนที่ด้วย API Key จริงของคุณ
"timeout": 30.0,
"max_retries": 5,
"retry_codes": [429, 502, 503, 504], # HTTP codes ที่ต้อง retry
}
class HolySheepAIClient:
"""
HolySheep AI Client พร้อมระบบ Retry แบบ Exponential Backoff
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.client = httpx.AsyncClient(
base_url=base_url,
timeout=30.0,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
)
# ตัวแปรสำหรับเก็บ Metrics
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"retried_requests": 0,
"errors_by_code": {},
"avg_response_time_ms": 0,
"last_error": None,
"last_error_time": None
}
async def chat_completion_with_retry(
self,
model: str,
messages: list,
temperature: float = 0.7,
max_tokens: int = 1000
) -> Dict[str, Any]:
"""
ส่ง request ไปยัง HolySheep AI พร้อมระบบ Retry แบบ Exponential Backoff
"""
self.metrics["total_requests"] += 1
last_exception = None
for attempt in range(HOLYSHEEP_CONFIG["max_retries"]):
try:
start_time = datetime.now()
response = await self.client.post(
"/chat/completions",
json={
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
)
# คำนวณ Response Time
response_time = (datetime.now() - start_time).total_seconds() * 1000
if response.status_code == 200:
self.metrics["successful_requests"] += 1
self.metrics["avg_response_time_ms"] = (
(self.metrics["avg_response_time_ms"] *
(self.metrics["successful_requests"] - 1) + response_time)
/ self.metrics["successful_requests"]
)
return response.json()
elif response.status_code in HOLYSHEEP_CONFIG["retry_codes"]:
# เก็บ Error Metrics
error_code = response.status_code
self.metrics["errors_by_code"][error_code] = \
self.metrics["errors_by_code"].get(error_code, 0) + 1
self.metrics["last_error"] = f"HTTP {error_code}"
self.metrics["last_error_time"] = datetime.now().isoformat()
# Log การ Retry
retry_delay = min(2 ** attempt * 0.5, 30) # Max 30 วินาที
logger.warning(
f"Attempt {attempt + 1}/{HOLYSHEEP_CONFIG['max_retries']}: "
f"HTTP {error_code} - Retrying in {retry_delay}s"
)
if attempt < HOLYSHEEP_CONFIG["max_retries"] - 1:
self.metrics["retried_requests"] += 1
await asyncio.sleep(retry_delay)
continue
# Status code อื่นๆ ไม่ต้อง Retry
response.raise_for_status()
except httpx.TimeoutException as e:
logger.error(f"Timeout on attempt {attempt + 1}: {e}")
last_exception = e
await asyncio.sleep(2 ** attempt)
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error on attempt {attempt + 1}: {e}")
last_exception = e
if e.response.status_code not in HOLYSHEEP_CONFIG["retry_codes"]:
break
await asyncio.sleep(2 ** attempt)
except Exception as e:
logger.error(f"Unexpected error: {e}")
last_exception = e
break
# ถ้าลองครบทุก attempt แล้วยังไม่สำเร็จ
self.metrics["failed_requests"] += 1
raise Exception(f"All retries exhausted. Last error: {last_exception}")
def get_metrics(self) -> Dict[str, Any]:
"""ดึง Metrics ปัจจุบัน"""
return {
**self.metrics,
"success_rate": round(
self.metrics["successful_requests"] /
max(self.metrics["total_requests"], 1) * 100, 2
),
"retry_rate": round(
self.metrics["retried_requests"] /
max(self.metrics["total_requests"], 1) * 100, 2
)
}
async def close(self):
await self.client.aclose()
ตัวอย่างการใช้งาน
async def main():
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
try:
result = await client.chat_completion_with_retry(
model="gpt-4.1",
messages=[
{"role": "system", "content": "คุณเป็นผู้ช่วยบริการลูกค้า"},
{"role": "user", "content": "สินค้าที่สั่งซื้อมาถึงเมื่อไหร่?"}
]
)
print(f"Response: {result['choices'][0]['message']['content']}")
# แสดง Metrics
print(f"Metrics: {client.get_metrics()}")
finally:
await client.close()
if __name__ == "__main__":
asyncio.run(main())ระบบ Alert แบบ Real-time พร้อม Line/Discord Notification
หลังจากมี Client ที่รองรับ Retry แล้ว ต่อไปเราต้องสร้างระบบ Alert ที่จะแจ้งเตือนเราทันทีเมื่อเกิดปัญหา ผมใช้ระบบ Alert ตัวนี้กับลูกค้าอีคอมเมิร์ซ 5 ราย และช่วยลดเวลาตอบสนองต่อปัญหาได้ถึง 90%:
import asyncio
import httpx
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import json
@dataclass
class AlertRule:
"""กำหนดเงื่อนไขการแจ้งเตือน"""
name: str
metric: str
threshold: float
comparison: str # "gt", "lt", "eq"
duration_seconds: int = 0 # 0 = ทันที, >0 = ต้องเกินเวลานา้
severity: str = "warning" # "info", "warning", "critical"
enabled: bool = True
@dataclass
class Alert:
"""ข้อมูลการแจ้งเตือน"""
rule: AlertRule
current_value: float
message: str
timestamp: datetime = field(default_factory=datetime.now)
acknowledged: bool = False
class HolySheepAlertManager:
"""
ระบบ Alert สำหรับ HolySheep AI
แจ้งเตือนผ่าน Line Notify, Discord Webhook หรือ Slack
"""
def __init__(self, client: 'HolySheepAIClient'):
self.client = client
self.alerts: List[Alert] = []
self.alert_history: List[Alert] = []
# ตั้งค่า Webhook URLs
self.webhook_configs = {
"line": {
"enabled": False,
"url": "YOUR_LINE_NOTIFY_TOKEN" # Line Notify Token
},
"discord": {
"enabled": True,
"url": "YOUR_DISCORD_WEBHOOK_URL" # Discord Webhook URL
}
}
# กำหนด Alert Rules
self.rules: List[AlertRule] = [
AlertRule(
name="High Error Rate",
metric="error_rate",
threshold=5.0, # 5%
comparison="gt",
severity="critical",
enabled=True
),
AlertRule(
name="Rate Limit Spike",
metric="429_count",
threshold=10,
comparison="gt",
duration_seconds=60,
severity="warning",
enabled=True
),
AlertRule(
name="Server Error Spike",
metric="5xx_count",
threshold=5,
comparison="gt",
severity="critical",
enabled=True
),
AlertRule(
name="High Latency",
metric="avg_response_time_ms",
threshold=2000.0, # 2 วินาที
comparison="gt",
duration_seconds=120,
severity="warning",
enabled=True
),
AlertRule(
name="Low Success Rate",
metric="success_rate",
threshold=95.0, # ต่ำกว่า 95%
comparison="lt",
severity="critical",
enabled=True
)
]
# ติดตามสถานะ Alert ที่กำลัง Active
self.active_alerts: Dict[str, datetime] = {}
def _check_threshold(self, rule: AlertRule, value: float) -> bool:
"""ตรวจสอบว่าค่าปัจจุบันเกินเกณฑ์หรือไม่"""
if rule.comparison == "gt":
return value > rule.threshold
elif rule.comparison == "lt":
return value < rule.threshold
elif rule.comparison == "eq":
return value == rule.threshold
return False
async def send_line_notification(self, message: str) -> bool:
"""ส่งการแจ้งเตือนผ่าน Line Notify"""
if not self.webhook_configs["line"]["enabled"]:
return False
try:
async with httpx.AsyncClient() as client:
response = await client.post(
"https://notify-api.line.me/api/notify",
headers={"Authorization": f"Bearer {self.webhook_configs['line']['url']}"},
data={"message": message}
)
return response.status_code == 200
except Exception as e:
print(f"Line notification failed: {e}")
return False
async def send_discord_notification(self, alert: Alert) -> bool:
"""ส่งการแจ้งเตือนผ่าน Discord Webhook"""
if not self.webhook_configs["discord"]["enabled"]:
return False
color_map = {
"info": 3447003,
"warning": 16776960,
"critical": 15158332
}
embed = {
"title": f"🚨 HolySheep AI Alert: {alert.rule.name}",
"color": color_map.get(alert.rule.severity, 3447003),
"fields": [
{"name": "Metric", "value": alert.rule.metric, "inline": True},
{"name": "Current Value", "value": str(alert.current_value), "inline": True},
{"name": "Threshold", "value": str(alert.rule.threshold), "inline": True},
{"name": "Severity", "value": alert.rule.severity.upper(), "inline": True},
{"name": "Message", "value": alert.message, "inline": False}
],
"timestamp": alert.timestamp.isoformat(),
"footer": {"text": "HolySheep AI Monitor"}
}
try:
async with httpx.AsyncClient() as client:
response = await client.post(
self.webhook_configs["discord"]["url"],
json={"embeds": [embed]}
)
return response.status_code == 204
except Exception as e:
print(f"Discord notification failed: {e}")
return False
async def check_alerts(self) -> List[Alert]:
"""
ตรวจสอบ Metrics ปัจจุบันและสร้าง Alert ถ้าจำเป็น
ควรเรียกใช้ทุก 10-30 วินาที
"""
metrics = self.client.get_metrics()
new_alerts: List[Alert] = []
# คำนวณค่าที่ต้องการตรวจสอบ
total = metrics["total_requests"]
errors = sum(metrics["errors_by_code"].values())
error_rate = (errors / total * 100) if total > 0 else 0
current_values = {
"error_rate": error_rate,
"429_count": metrics["errors_by_code"].get(429, 0),
"5xx_count": sum(metrics["errors_by_code"].get(c, 0)
for c in [502, 503, 504]),
"avg_response_time_ms": metrics["avg_response_time_ms"],
"success_rate": metrics["success_rate"]
}
for rule in self.rules:
if not rule.enabled:
continue
value = current_values.get(rule.metric)
if value is None:
continue
# ตรวจสอบว่าเกินเกณฑ์หรือไม่
if self._check_threshold(rule, value):
# ถ้า duration > 0 ต้องเกินเวลานา้ก่อนถึงจะ Alert
if rule.duration_seconds > 0:
alert_key = f"{rule.name}_start"
if alert_key not in self.active_alerts:
self.active_alerts[alert_key] = datetime.now()
else:
elapsed = (datetime.now() -
self.active_alerts[alert_key]).total_seconds()
if elapsed < rule.duration_seconds:
continue # ยังไม่ครบเวลา
# สร้าง Alert
alert = Alert(
rule=rule,
current_value=value,
message=self._generate_alert_message(rule, value, metrics)
)
new_alerts.append(alert)
self.alerts.append(alert)
self.alert_history.append(alert)
# ส่งการแจ้งเตือน
await self.send_discord_notification(alert)
# ถ้าเป็น Line ให้ส่งด้วย
await self.send_line_notification(
f"🚨 HolySheep AI Alert!\n{alert.message}"
)
else:
# ถ้าค่ากลับมาปกติ ลบ active alert
alert_key = f"{rule.name}_start"
if alert_key in self.active_alerts:
del self.active_alerts[alert_key]
return new_alerts
def _generate_alert_message(
self,
rule: AlertRule,
value: float,
metrics: Dict
) -> str:
"""สร้างข้อความสำหรับ Alert"""
messages = {
"High Error Rate": (
f"พบ Error Rate สูงถึง {value:.2f}% "
f"(Threshold: {rule.threshold}%)"
),
"Rate Limit Spike": (
f"พบ HTTP 429 (Rate Limit) {value:.0f} ครั้ง "
f"ในช่วงเวลาที่ผ่านมา"
),
"Server Error Spike": (
f"พบ Server Error (5xx) {value:.0f} ครั้ง "
f"ต้องตรวจสอบระบบ HolySheep AI"
),
"High Latency": (
f"Response Time เฉลี่ย {value:.0f}ms "
f"(Threshold: {rule.threshold}ms)"
),
"Low Success Rate": (
f"Success Rate ต่ำเพียง {value:.2f}% "
f"(Threshold: {rule.threshold}%)"
)
}
return messages.get(rule.name, f"Alert: {rule.name}")
async def start_monitoring(self, interval_seconds: int = 30):
"""
เริ่มระบบ Monitoring แบบ Loop
ควรรันใน Task แยก
"""
print(f"🔍 เริ่มระบบ Monitoring ทุก {interval_seconds} วินาที...")
while True:
try:
new_alerts = await self.check_alerts()
if new_alerts:
print(f"⚠️ พบ {len(new_alerts)} Alert ใหม่")
for alert in new_alerts:
print(f" - [{alert.rule.severity.upper()}] {alert.rule.name}")
await asyncio.sleep(interval_seconds)
except Exception as e:
print(f"Monitoring error: {e}")
await asyncio.sleep(5)
ตัวอย่างการใช้งานแบบ Complete
async def main_with_alert():
# สร้าง Client
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# สร้าง Alert Manager
alert_manager = HolySheepAlertManager(client)
# เริ่ม Monitoring Task
monitor_task = asyncio.create_task(alert_manager.start_monitoring(30))
# ทดสอบโดยส่ง Request
for i in range(10):
try:
result = await client.chat_completion_with_retry(
model="gpt-4.1",
messages=[
{"role": "user", "content": f"ทดสอบครั้งที่ {i+1}"}
]
)
print(f"✅ Request {i+1} สำเร็จ")
except Exception as e:
print(f"❌ Request {i+1} ล้มเหลว: {e}")
await asyncio.sleep(2)
# แสดงผล Metrics สุดท้าย
print("\n📊 Final Metrics:")
print(json.dumps(client.get_metrics(), indent=2, ensure_ascii=False))
# หยุด Monitoring
monitor_task.cancel()
await client.close()
if __name__ == "__main__":
asyncio.run(main_with_alert())Dashboard สำหรับติดตามสถานะแบบ Real-time
นอกจาก Alert แล้ว ผมยังแนะนำให้มี Dashboard สำหรับดูสถานะโดยรวม ซึ่งผมใช้ FastAPI + HTML Dashboard ง่ายๆ แต่ใช้งานได้จริง:
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
import uvicorn
from datetime import datetime
from typing import Optional
import json
app = FastAPI(title="HolySheep AI Monitoring Dashboard")
ตัวแปร Global สำหรับเก็บ Metrics
global_metrics = {
"status": "healthy",
"last_check": datetime.now().isoformat(),
"uptime_seconds": 0,
"models": {}
}
class MetricsCollector:
""" Singleton สำหรับเก็บ Metrics กลาง """
def __init__(self):
self.start_time = datetime.now()
self.data = {}
def update(self, model: str, metrics: dict):
if model not in self.data:
self.data[model] = {
"requests": 0,
"errors": 0,
"total_tokens": 0,
"cost_usd": 0.0
}
self.data[model]["requests"] += metrics.get("total_requests", 0)
self.data[model]["errors"] += metrics.get("failed_requests", 0)
self.data[model]["total_tokens"] += metrics.get("total_tokens", 0)
self.data[model]["cost_usd"] += self._calculate_cost(model, metrics)
def _calculate_cost(self, model: str, metrics: dict) -> float:
"""คำนวณค่าใช้จ่ายจาก Tokens (ราคาเป็น USD ต่อ MTok)"""
pricing = {
"gpt-4.1": 8.0, # $8/MTok
"claude-sonnet-4.5": 15.0, # $15/MTok
"gemini-2.5-flash": 2.50, # $2.50/MTok
"deepseek-v3.2": 0.42 # $0.42/MTok
}
# ประมาณการ tokens จาก requests
estimated_tokens = metrics.get("total_requests", 0) * 1000
rate_per_mtok = pricing.get(model, 8.0)
return (estimated_tokens / 1_000_000) * rate_per_mtok
def get_summary(self) -> dict:
total_requests = sum(d["requests"] for d in self.data.values())
total_errors = sum(d["errors"] for d in self.data.values())
total_cost = sum(d["cost_usd"] for d in self.data.values())
uptime = (datetime.now() - self.start_time).total_seconds()
return {
"status": "healthy" if total_errors == 0 else "degraded",
"uptime_seconds": int(uptime),
"uptime_human": f"{int(uptime/3600)}h {int((uptime%3600)/60)}m",
"total_requests": total_requests,
"total_errors": total_errors,
"error_rate": f"{(total_errors/total_requests*100):.2f}%" if total_requests > 0 else "0%",
"total_cost_usd": f"${total_cost:.4f}",
"total_cost_thb": f"฿{total_cost*35:.2f}", # อัตรา $1=฿35
"models": self.data,
"last_updated": datetime.now().isoformat()
}
metrics_collector = MetricsCollector()
@app.get("/")
async def root():
"""Dashboard HTML"""
return HTMLResponse(content=DASHBOARD_HTML, media_type="text/html")
@app.get("/api/metrics")
async def get_metrics():
"""API endpoint สำหรับดึง Metrics ทั้งหมด"""
return metrics_collector.get_summary()
@app.post("/api/metrics/{model}")
async def update_metrics(model: str, data: dict):
"""API endpoint สำหรับอัพเดท Metrics จาก Application"""
metrics_collector.update(model, data)
return {"status": "ok", "model": model}
@app.get("/api/health")
async def health_check():
"""Health check endpoint สำหรับ Load Balancer"""
summary = metrics_collector.get_summary()
return {
"status": summary["status"],
"timestamp": datetime.now().isoformat()
}
DASHBOARD_HTML = """
HolySheep AI Monitor Dashboard