การตรวจสอบความผิดปกติและการตั้งค่าการแจ้งเตือนสำหรับ OpenAI API

บทนำ: เหตุการณ์จริงที่เกิดขึ้น

เมื่อเดือนที่แล้ว ระบบของผมเจอปัญหาใหญ่หลวง: ConnectionError: timeout ติดต่อกัน 3 ชั่วโมงโดยไม่มีใครรู้ ลูกค้าติดต่อเข้ามาบอกว่าแชทบอทไม่ตอบ พอเช็คดู Log ถึงเพิ่งรู้ว่า API เป็นสีแดงตั้งแต่ตี 3 ปัญหานี้สอนผมว่า "การมี monitoring ที่ดี สำคัญกว่าการมี API ที่ดี" วันนี้ผมจะแชร์วิธีการตั้งค่า monitoring และ alert ที่ครอบคลุม โดยใช้ HolySheep AI เป็นตัวอย่าง API endpoint ซึ่งมีความเสถียรสูงและราคาประหยัดกว่า 85% เมื่อเทียบกับผู้ให้บริการอื่น

โครงสร้างพื้นฐานสำหรับการตรวจสอบ

การตรวจสอบ API ที่ดีต้องครอบคลุม 3 ด้านหลัก: HTTP status code, response time และ error rate ในการนี้ผมใช้ Python กับ prometheus_client และ custom exception handler สำหรับจัดการ error ทุกรูปแบบ โดย base_url จะใช้ https://api.holysheep.ai/v1 ที่ให้ความเร็วต่ำกว่า 50 มิลลิวินาที

import requests
import time
import logging
from datetime import datetime
from typing import Optional, Dict, Any

class APIHealthMonitor:
    """ตัวตรวจสอบสถานะ API แบบครอบคลุม"""
    
    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.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.logger = logging.getLogger(__name__)
        
        # เก็บสถิติ
        self.stats = {
            "total_requests": 0,
            "failed_requests": 0,
            "timeout_count": 0,
            "auth_errors": 0,
            "avg_response_time": 0
        }
    
    def check_api_health(self) -> Dict[str, Any]:
        """ตรวจสอบสถานะ API พื้นฐาน"""
        start_time = time.time()
        result = {
            "timestamp": datetime.now().isoformat(),
            "status": "unknown",
            "response_time_ms": 0,
            "error_type": None,
            "error_message": None
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=self.headers,
                json={
                    "model": "gpt-4.1",
                    "messages": [{"role": "user", "content": "health check"}],
                    "max_tokens": 5
                },
                timeout=10
            )
            
            result["response_time_ms"] = round((time.time() - start_time) * 1000, 2)
            
            if response.status_code == 200:
                result["status"] = "healthy"
            elif response.status_code == 401:
                result["status"] = "auth_error"
                result["error_type"] = "401 Unauthorized"
                result["error_message"] = "API key ไม่ถูกต้องหรือหมดอายุ"
                self.stats["auth_errors"] += 1
            elif response.status_code == 429:
                result["status"] = "rate_limited"
                result["error_type"] = "429 Too Many Requests"
                result["error_message"] = "เกินขีดจำกัดการใช้งาน"
            else:
                result["status"] = "error"
                result["error_type"] = f"HTTP {response.status_code}"
                result["error_message"] = response.text[:200]
                
        except requests.exceptions.Timeout:
            result["status"] = "timeout"
            result["error_type"] = "ConnectionError: timeout"
            result["error_message"] = "เซิร์ฟเวอร์ไม่ตอบสนองภายในเวลาที่กำหนด"
            result["response_time_ms"] = 10000
            self.stats["timeout_count"] += 1
            
        except requests.exceptions.ConnectionError as e:
            result["status"] = "connection_error"
            result["error_type"] = "ConnectionError"
            result["error_message"] = str(e)
            self.stats["failed_requests"] += 1
            
        except Exception as e:
            result["status"] = "unexpected_error"
            result["error_type"] = type(e).__name__
            result["error_message"] = str(e)
            self.stats["failed_requests"] += 1
        
        self.stats["total_requests"] += 1
        self._update_stats(result)
        
        return result
    
    def _update_stats(self, result: Dict[str, Any]):
        """อัพเดทสถิติ"""
        if result["status"] == "healthy":
            current_avg = self.stats["avg_response_time"]
            total_healthy = self.stats["total_requests"] - self.stats["failed_requests"]
            if total_healthy > 0:
                self.stats["avg_response_time"] = (
                    (current_avg * (total_healthy - 1) + result["response_time_ms"]) / total_healthy
                )

การใช้งาน
monitor = APIHealthMonitor(api_key="YOUR_HOLYSHEEP_API_KEY")
health_status = monitor.check_api_health()
print(f"สถานะ: {health_status['status']}")
print(f"เวลาตอบสนอง: {health_status['response_time_ms']} ms")

การตั้งค่า Alert System อัตโนมัติ

การมี monitoring เฉยๆ ไม่พอ ต้องมีการแจ้งเตือนอัตโนมัติเมื่อเกิดปัญหา ผมจะสร้างระบบ alert ที่ส่ง notification ผ่านหลายช่องทาง ไม่ว่าจะเป็น Line Notify, Slack หรือ Email โดยกำหนดเงื่อนไขการแจ้งเตือนตามระดับความรุนแรง ระบบนี้ใช้กับ HolySheep AI ได้ทันทีเพราะ API structure เข้ากันได้กับ OpenAI SDK

import smtplib
import json
from email.mime.text import MIMEText
from typing import List, Callable
from dataclasses import dataclass
from enum import Enum
import threading
import queue

class AlertLevel(Enum):
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"

@dataclass
class Alert:
    level: AlertLevel
    title: str
    message: str
    timestamp: str
    metadata: dict

class AlertManager:
    """จัดการการแจ้งเตือนหลายช่องทาง"""
    
    def __init__(self):
        self.channels: List[Callable] = []
        self.alert_queue = queue.Queue()
        self.alert_history: List[Alert] = []
        self.alert_rules = []
        self._setup_default_rules()
    
    def _setup_default_rules(self):
        """กำหนดกฎการแจ้งเตือน"""
        self.alert_rules = [
            {
                "condition": lambda s: s["timeout_count"] >= 3,
                "level": AlertLevel.CRITICAL,
                "title": "API Timeout ติดต่อกัน",
                "message": "พบ timeout {} ครั้งในช่วงเวลาที่ผ่านมา"
            },
            {
                "condition": lambda s: s["auth_errors"] >= 1,
                "level": AlertLevel.CRITICAL,
                "title": "401 Unauthorized",
                "message": "พบข้อผิดพลาดการยืนยันตัวตน API key อาจหมดอายุ"
            },
            {
                "condition": lambda s: s["failed_requests"] / max(s["total_requests"], 1) > 0.1,
                "level": AlertLevel.ERROR,
                "title": "Error Rate สูงเกิน 10%",
                "message": "อัตราความผิดพลาด {}%"
            },
            {
                "condition": lambda s: s["avg_response_time"] > 5000,
                "level": AlertLevel.WARNING,
                "title": "Response Time สูง",
                "message": "เวลาตอบสนองเฉลี่ย {} ms"
            }
        ]
    
    def add_channel(self, channel: Callable):
        """เพิ่มช่องทางการแจ้งเตือน"""
        self.channels.append(channel)
    
    def send_line_notify(self, token: str, message: str):
        """ส่งการแจ้งเตือนผ่าน Line Notify"""
        import requests
        url = "https://notify-api.line.me/api/notify"
        headers = {"Authorization": f"Bearer {token}"}
        data = {"message": message}
        try:
            requests.post(url, headers=headers, data=data, timeout=5)
        except Exception as e:
            print(f"ส่ง Line Notify ไม่สำเร็จ: {e}")
    
    def send_slack_webhook(self, webhook_url: str, alert: Alert):
        """ส่งการแจ้งเตือนผ่าน Slack"""
        import requests
        payload = {
            "text": f"*{alert.title}*",
            "attachments": [{
                "color": self._get_color(alert.level),
                "fields": [
                    {"title": "ระดับ", "value": alert.level.value, "short": True},
                    {"title": "เวลา", "value": alert.timestamp, "short": True},
                    {"title": "รายละเอียด", "value": alert.message}
                ]
            }]
        }
        try:
            requests.post(webhook_url, json=payload, timeout=5)
        except Exception as e:
            print(f"ส่ง Slack ไม่สำเร็จ: {e}")
    
    def send_email(self, smtp_server: str, port: int, sender: str, 
                   password: str, recipients: List[str], alert: Alert):
        """ส่งการแจ้งเตือนทาง Email"""
        msg = MIMEText(f"{alert.title}\n\n{alert.message}\n\nเวลา: {alert.timestamp}")
        msg["Subject"] = f"[{alert.level.value.upper()}] {alert.title}"
        msg["From"] = sender
        msg["To"] = ", ".join(recipients)
        
        try:
            with smtplib.SMTP(smtp_server, port) as server:
                server.starttls()
                server.login(sender, password)
                server.send_message(msg)
        except Exception as e:
            print(f"ส่ง Email ไม่สำเร็จ: {e}")
    
    def _get_color(self, level: AlertLevel) -> str:
        colors = {
            AlertLevel.INFO: "#36a64f",
            AlertLevel.WARNING: "#ff9900",
            AlertLevel.ERROR: "#ff0000",
            AlertLevel.CRITICAL: "#8b0000"
        }
        return colors.get(level, "#808080")
    
    def evaluate_rules(self, stats: dict):
        """ประเมินกฎและส่งการแจ้งเตือน"""
        for rule in self.alert_rules:
            if rule["condition"](stats):
                # สร้างข้อความจาก template
                message = rule["message"]
                for key, value in stats.items():
                    message = message.replace(f"{{{key}}}", str(value))
                
                alert = Alert(
                    level=rule["level"],
                    title=rule["title"],
                    message=message,
                    timestamp=datetime.now().isoformat(),
                    metadata=stats
                )
                
                # ส่งการแจ้งเตือนไปยังทุกช่องทาง
                for channel in self.channels:
                    try:
                        channel(alert)
                    except Exception as e:
                        print(f"ส่งการแจ้งเตือนล้มเหลว: {e}")
                
                self.alert_history.append(alert)
                break  # ส่งแค่ alert แรกที่ match
    
    def get_alert_summary(self) -> dict:
        """สรุปสถานะการแจ้งเตือน"""
        return {
            "total_alerts": len(self.alert_history),
            "by_level": {
                level.value: sum(1 for a in self.alert_history if a.level == level)
                for level in AlertLevel
            },
            "recent_alerts": [
                {"title": a.title, "level": a.level.value, "time": a.timestamp}
                for a in self.alert_history[-5:]
            ]
        }

การใช้งาน
alert_manager = AlertManager()

เพิ่มช่องทางการแจ้งเตือน
alert_manager.add_channel(lambda a: alert_manager.send_line_notify("LINE_TOKEN", f"{a.title}\n{a.message}"))
alert_manager.add_channel(lambda a: alert_manager.send_slack_webhook("SLACK_WEBHOOK_URL", a))

ประเมิน stats จาก monitor
stats = monitor.stats
alert_manager.evaluate_rules(stats)

print("สรุปการแจ้งเตือน:", alert_manager.get_alert_summary())

การสร้าง Health Dashboard แบบ Real-time

นอกจากการแจ้งเตือนแล้ว การมี dashboard ที่มองเห็นสถานะแบบ real-time ก็สำคัญมาก ผมจะสร้าง web dashboard เล็กๆ ด้วย Flask ที่แสดงสถานะ API และประวัติการแจ้งเตือน พร้อมทั้ง chart แสดง response time ย้อนหลัง ซึ่งทำให้เห็นภาพรวมได้ชัดเจน

from flask import Flask, jsonify, render_template_string
import threading
import time
from datetime import datetime

app = Flask(__name__)

เก็บประวัติ
health_history = []
ALERT_THRESHOLDS = {
    "response_time_ms": 5000,
    "error_rate_percent": 10,
    "timeout_count": 3
}

HTML_TEMPLATE = '''



    
    
    
    


    
📊 API Health Dashboard
    

        

            
สถานะปัจจุบัน
            

                {{ '✅' if current_status == 'healthy' else '❌' }}
            
            
{{ current_status.upper() }}
        
        

            
สถิติรวม
            

                

                    
{{ stats.total_requests }}
                    
คำขอทั้งหมด
                
                

                    
{{ stats.failed_requests }}
                    
คำขอที่ล้มเหลว
                
                

                    
{{ "%.1f"|format(stats.avg_response_time) }}ms
                    
เวลาตอบสนอง
                
            
        
        

            
การแจ้งเตือนล่าสุด
            

                {% for alert in recent_alerts %}
                

                    {{ alert.title }}

                    {{ alert.message }}

                    {{ alert.timestamp }}
                
                {% endfor %}
            
        
    
    

        
ประวัติ Response Time
        
    


'''

def run_monitoring_loop():
    """ลูปการตรวจสอบอัตโนมัติ"""
    while True:
        try:
            result = monitor.check_api_health()
            health_history.append(result)
            
            # เก็บแค่ 100 รายการล่าสุด
            if len(health_history) > 100:
                health_history.pop(0)
            
            # ประเมิน alert
            alert_manager.evaluate_rules(monitor.stats)
            
        except Exception as e:
            print(f"เกิดข้อผิดพลาดในลูปตรวจสอบ: {e}")
        
        time.sleep(60)  # ตรวจสอบทุก 60 วินาที

@app.route('/')
def dashboard():
    current = health_history[-1] if health_history else {"status": "unknown", "response_time_ms": 0}
    
    return render_template_string(HTML_TEMPLATE,
        current_status=current.get("status", "unknown"),
        stats=monitor.stats,
        recent_alerts=alert_manager.get_alert_summary()["recent_alerts"]
    )

@app.route('/api/health')
def api_health():
    """API endpoint สำหรับดึงข้อมูล JSON"""
    return jsonify({
        "current": health_history[-1] if health_history else None,
        "stats": monitor.stats,
        "history": health_history[-20:],
        "alerts": alert_manager.get_alert_summary()
    })

@app.route('/api/check')
def force_check():
    """บังคับตรวจสอบทันที"""
    result = monitor.check_api_health()
    health_history.append(result)
    alert_manager.evaluate_rules(monitor.stats)
    return jsonify(result)

if __name__ == '__main__':
    # เริ่มลูปการตรวจสอบใน thread แยก
    monitor_thread = threading.Thread(target=run_monitoring_loop, daemon=True)
    monitor_thread.start()
    
    # รัน Flask server
    app.run(host='0.0.0.0', port=5000, debug=False)

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

1. ConnectionError: timeout บ่อยครั้ง

ปัญหานี้เกิดจากหลายสาเหตุ ส่วนใหญ่คือ network timeout หรือ API server ตอบสนองช้า วิธีแก้ไขคือตั้งค่า timeout ที่เหมาะสมและใช้ retry logic พร้อม exponential backoff

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_resilient_session() -> requests.Session:
    """สร้าง session ที่ทนทานต่อข้อผิดพลาด"""
    session = requests.Session()
    
    # ตั้งค่า retry strategy
    retry_strategy = Retry(
        total=3,  # ลองใหม่สูงสุด 3 ครั้ง
        backoff_factor=1,  # รอ 1, 2, 4 วินาทีระหว่าง retry
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["POST", "GET"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    return session

def safe_api_call(messages: list, model: str = "gpt-4.1") -> dict:
    """เรียก API แบบปลอดภัยพร้อม fallback"""
    
    session = create_resilient_session()
    
    try:
        response = session.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
                "Content-Type": "application/json"
            },
            json={
                "model": model,
                "messages": messages
            },
            timeout=(10, 30)  # (connect_timeout, read_timeout)
        )
        response.raise_for_status()
        return {"success": True, "data": response.json()}
        
    except requests.exceptions.Timeout:
        # Fallback ไปใช้ model ที่เร็วกว่า
        print("Timeout - ลองใช้ Gemini 2.5 Flash แทน")
        try:
            response = session.post(
                "https://api.holysheep.ai/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "gemini-2.5-flash",  # model ที่เร็วกว่า
                    "messages": messages
                },
                timeout=(5, 15)
            )
            return {"success": True, "data": response.json(), "fallback": True}
        except Exception as e:
            return {"success": False, "error": str(e), "fallback_failed": True}
            
    except requests.exceptions.ConnectionError:
        return {"success": False, "error": "ConnectionError", "suggestion": "ตรวจสอบ internet connection"}
    
    except Exception as e:
        return {"success": False, "error": str(e)}

ทดสอบ
result = safe_api_call([{"role": "user", "content": "ทดสอบ"}])
print(result)

2. 401 Unauthorized: Invalid API Key

ข้อผิดพลาดนี้เกิดจาก API key ไม่ถูกต้อง หมดอายุ หรือถูก revoke โดยเฉพาะเมื่อใช้ HolySheep AI ต้องตรวจสอบว่าใช้ key ที่ถูกต้องและมีเครดิตเพียงพอ วิธีแก้คือตรวจสอบ key ก่อนใช้งานและเก็บใน environment variable

import os
from dotenv import load_dotenv

class APIKeyValidator:
    """ตรวจสอบความถูกต้องของ API key"""
    
    @staticmethod
    def validate_key(api_key: str) -> dict:
        """ตรวจสอบ key และคืนค่าสถานะ"""
        import requests
        
        if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
            return {
                "valid": False,
                "error": "API key ยังไม่ได้ตั้งค่า",
                "solution": "ตั้งค่า HOLYSHEEP_API_KEY ใน .env file"
            }
        
        try:
            # ทดสอบเรียก API ด้วย model ราคาถูกที่สุด
            response = requests.post(
                "https://api.holysheep.ai/v1/chat/completions",
                headers={"Authorization": f"Bearer {api_key}"},
                json={
                    "model": "deepseek-v3.2",  # model ราคาถูก $0.42/MTok
                    "messages": [{"role": "user", "content": "test"}],
                    "max_tokens": 5
                },
                timeout=5
            )
            
            if response.status_code == 401:
                return {
                    "valid": False,
                    "error": "401 Unauthorized - API key ไม่ถูกต้อง",
                    "solution": "ตรวจสอบ API key ที่ https://www.holysheep.ai/dashboard"
                }
            elif response.status_code == 200:
                return {"valid": True, "message": "API key ถูกต้อง"}
            else:
                return {
                    "valid": False,
                    "error": f"HTTP {response.status_code}",
                    "response": response.text[:200]
                }
                
        except Exception as e:
            return {
                "valid": False,
                "error": str(e),
                "solution": "ตรวจสอบ internet connection และ API endpoint"
            }

def get_api_key() -> str:
    """ดึง API key จาก environment variable"""
    load_dotenv()
    api_key = os.getenv("HOLYSHEEP_API_KEY")
    
    if not api_key:
        print("⚠️ ไม่พบ HOLYSHEEP_API_KEY")
        print("กรุณาสร้างไฟล์ .env และเพิ่มบรรทัด:")
        print("HOLYSHEEP_API_KEY=your_api_key_here")
        print("\n📝 สมัครได้ที่: https://www.holysheep.ai/register")
        return None
    
    # ตรวจสอบความถูกต้อง
    validator = APIKeyValidator()
    result = validator.validate_key(api_key)
    
    if result["valid"]:
        print("✅ API key ถูกต้องพร้อมใช้งาน")
    else:
        print(f"❌ {result['error']}")
        print(f"💡 {result.get('solution', '')}")
    
    return api_key if result["valid"] else None

การใช้งาน
if __name__ == "__main__":
    api_key = get_api_key()
    if api_key:
        # ใช้ API key ต่อได้เลย
        print(f"API Key: {api_key[:10]}...")

3. Rate Limit Exceeded (429 Too Many Requests)

ปัญหานี้เกิดเมื่อส่ง request เร็วเกินไปเกินขีดจำ