คืนนั้นผมนั่งดู Dashboard อยู่คนเดียว จู่ๆ ระบบก็ส่ง Alert มา 47 ฉบับพร้อมกัน: "ConnectionError: timeout after 30000ms" ลูกค้าติดต่อเข้ามาแจ้งว่า AI Chatbot หยุดทำงาน ผมลองเช็คดูพบว่า API Key หมดอายุการใช้งาน แต่ระบบไม่ได้แจ้งเตือนล่วงหน้าเลย นั่นคือจุดเริ่มต้นที่ทำให้ผมเริ่มสร้างระบบ Monitoring สำหรับ AI API อย่างจริงจัง

บทความนี้จะสอนการตั้งค่า Prometheus + Grafana เพื่อเฝ้าระวัง API ของ AI อย่างมีประสิทธิภาพ โดยใช้ HolySheep AI เป็นตัวอย่าง ราคาถูกกว่า 85%+ เมื่อเทียบกับผู้ให้บริการอื่น รองรับ WeChat และ Alipay พร้อมเครดิตฟรีเมื่อลงทะเบียน

ทำไมต้องมีระบบ Monitoring สำหรับ AI API

ปัญหาที่พบบ่อยที่สุดเมื่อใช้ AI API คือ:

ระบบ Monitoring ที่ดีจะช่วยให้เรา:

สถาปัตยกรรมระบบ Monitoring

ระบบที่เราจะสร้างประกอบด้วย 4 ส่วนหลัก:

ขั้นตอนที่ 1: สร้าง Prometheus Exporter

Exporter ทำหน้าที่เก็บ Metrics จาก API แล้วส่งให้ Prometheus โดยใช้ Format มาตรฐาน ติดตั้ง Library ที่จำเป็นก่อน:

pip install prometheus-client requests python-dotenv schedule

สร้างไฟล์ ai_api_exporter.py สำหรับเก็บ Metrics จาก HolySheep AI:

import requests
import time
from prometheus_client import Counter, Histogram, Gauge, start_http_server
from prometheus_client import REGISTRY
import schedule
import os
from dotenv import load_dotenv

load_dotenv()


Configuration สำหรับ HolySheep AI API

BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")


กำหนด Metrics ที่ต้องการเก็บ

REQUEST_COUNT = Counter(
    'ai_api_requests_total',
    'Total AI API requests',
    ['model', 'status_code']
)

REQUEST_LATENCY = Histogram(
    'ai_api_request_duration_seconds',
    'AI API request latency',
    ['model']
)

ERROR_COUNT = Counter(
    'ai_api_errors_total',
    'Total AI API errors',
    ['error_type', 'model']
)

API_CREDITS = Gauge(
    'ai_api_remaining_credits',
    'Remaining API credits'
)

RATE_LIMIT_REMAINING = Gauge(
    'ai_api_rate_limit_remaining',
    'Remaining rate limit'
)

def check_api_health():
    """ตรวจสอบสถานะ API และเก็บ Metrics"""
    models = ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"]
    
    for model in models:
        start_time = time.time()
        
        try:
            headers = {
                "Authorization": f"Bearer {API_KEY}",
                "Content-Type": "application/json"
            }
            
            # ทดสอบ API ด้วย Chat Completions endpoint
            payload = {
                "model": model,
                "messages": [{"role": "user", "content": "ping"}],
                "max_tokens": 5
            }
            
            response = requests.post(
                f"{BASE_URL}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            
            latency = time.time() - start_time
            
            # บันทึก Metrics
            REQUEST_COUNT.labels(model=model, status_code=response.status_code).inc()
            REQUEST_LATENCY.labels(model=model).observe(latency)
            
            # เก็บ Rate Limit info จาก Headers
            if 'X-RateLimit-Remaining' in response.headers:
                remaining = int(response.headers['X-RateLimit-Remaining'])
                RATE_LIMIT_REMAINING.labels(model=model).set(remaining)
            
            # ตรวจสอบ Error types
            if response.status_code == 401:
                ERROR_COUNT.labels(error_type='unauthorized', model=model).inc()
            elif response.status_code == 429:
                ERROR_COUNT.labels(error_type='rate_limit', model=model).inc()
            elif response.status_code >= 500:
                ERROR_COUNT.labels(error_type='server_error', model=model).inc()
                
        except requests.exceptions.Timeout:
            latency = time.time() - start_time
            REQUEST_COUNT.labels(model=model, status_code='timeout').inc()
            REQUEST_LATENCY.labels(model=model).observe(30.0)
            ERROR_COUNT.labels(error_type='timeout', model=model).inc()
            
        except requests.exceptions.ConnectionError as e:
            latency = time.time() - start_time
            REQUEST_COUNT.labels(model=model, status_code='connection_error').inc()
            REQUEST_LATENCY.labels(model=model).observe(latency)
            ERROR_COUNT.labels(error_type='connection_error', model=model).inc()

def check_credits():
    """ตรวจสอบ Credits ที่เหลืออยู่"""
    try:
        headers = {"Authorization": f"Bearer {API_KEY}"}
        response = requests.get(f"{BASE_URL}/usage", headers=headers, timeout=10)
        
        if response.status_code == 200:
            data = response.json()
            credits = data.get('remaining', 0)
            API_CREDITS.set(credits)
            
            # Alert ถ้า Credits ต่ำกว่า 10%
            if credits < 100:  # ปรับตาม Usage pattern
                print(f"⚠️ Warning: Low credits! Remaining: {credits}")
    except Exception as e:
        print(f"Error checking credits: {e}")

if __name__ == '__main__':
    # Start HTTP server สำหรับ Prometheus scrape
    start_http_server(8000)
    print("🚀 AI API Exporter started on port 8000")
    
    # Schedule tasks
    schedule.every(30).seconds.do(check_api_health)
    schedule.every(5).minutes.do(check_credits)
    
    # Run ครั้งแรกทันที
    check_api_health()
    check_credits()
    
    # Loop หลัก
    while True:
        schedule.run_pending()
        time.sleep(1)

ขั้นตอนที่ 2: ตั้งค่า Prometheus

สร้างไฟล์ prometheus.yml เพื่อกำหนด Target ที่ต้องการ Scrape:

global:
  scrape_interval: 15s
  evaluation_interval: 15s

alerting:
  alertmanagers:
    - static_configs:
        - targets:
          - alertmanager:9093

rule_files:
  - "alert_rules.yml"

scrape_configs:
  - job_name: 'ai_api_exporter'
    static_configs:
      - targets: ['ai-api-exporter:8000']
    metrics_path: '/metrics'
    scrape_interval: 30s

  - job_name: 'prometheus'
    static_configs:
      - targets: ['localhost:9090']

สร้างไฟล์ alert_rules.yml สำหรับกำหนด Alert Conditions:

groups:
  - name: ai_api_alerts
    rules:
      # Alert เมื่อ Error Rate สูงกว่า 5%
      - alert: HighErrorRate
        expr: |
          rate(ai_api_errors_total[5m]) / 
          rate(ai_api_requests_total[5m]) > 0.05
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "AI API Error Rate สูง ({{ $value | humanizePercentage }})"
          description: "Model {{ $labels.model }} มี Error Rate สูงผิดปกติ"

      # Alert เมื่อ Latency เฉลี่ยเกิน 3 วินาที
      - alert: HighLatency
        expr: |
          histogram_quantile(0.95, 
            rate(ai_api_request_duration_seconds_bucket[5m])
          ) > 3
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "AI API Latency สูง ({{ $value | humanizeDuration }})"
          description: "Model {{ $labels.model }} มี Response Time สูง"

      # Alert เมื่อ Connection Error เกิดขึ้น
      - alert: ConnectionErrors
        expr: |
          increase(ai_api_errors_total{error_type="connection_error"}[5m]) > 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: "Connection Error กับ AI API"
          description: "ไม่สามารถเชื่อมต่อกับ {{ $labels.model }} API"

      # Alert เมื่อ Credits ใกล้หมด
      - alert: LowCredits
        expr: ai_api_remaining_credits < 100
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "API Credits ใกล้หมด"
          description: "เหลือ Credits เพียง {{ $value }} หน่วย"

      # Alert เมื่อ Rate Limit ถูก Block
      - alert: RateLimitExceeded
        expr: |
          increase(ai_api_errors_total{error_type="rate_limit"}[10m]) > 5
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "Rate Limit ถูก Block บ่อยครั้ง"
          description: "Model {{ $labels.model }} ถูก Rate Limit {{ $value }} ครั้งใน 10 นาที"

      # Alert เมื่อ 401 Unauthorized
      - alert: UnauthorizedAPI
        expr: |
          increase(ai_api_errors_total{error_type="unauthorized"}[1m]) > 0
        for: 0m
        labels:
          severity: critical
        annotations:
          summary: "401 Unauthorized - API Key อาจหมดอายุ"
          description: "การเข้าถึง API ถูก Reject ด้วย 401 Unauthorized"

ขั้นตอนที่ 3: สร้าง Grafana Dashboard

Import Dashboard นี้ใน Grafana เพื่อดูภาพรวมของ API Performance:

{
  "dashboard": {
    "title": "AI API Monitoring Dashboard",
    "tags": ["ai", "api", "monitoring"],
    "timezone": "Asia/Bangkok",
    "panels": [
      {
        "title": "Request Rate (RPM)",
        "type": "graph",
        "gridPos": {"x": 0, "y": 0, "w": 12, "h": 8},
        "targets": [
          {
            "expr": "sum(rate(ai_api_requests_total[1m])) by (model)",
            "legendFormat": "{{model}}"
          }
        ]
      },
      {
        "title": "Error Rate by Type",
        "type": "piechart",
        "gridPos": {"x": 12, "y": 0, "w": 12, "h": 8},
        "targets": [
          {
            "expr": "sum(increase(ai_api_errors_total[1h])) by (error_type)",
            "legendFormat": "{{error_type}}"
          }
        ]
      },
      {
        "title": "P95 Latency (Seconds)",
        "type": "gauge",
        "gridPos": {"x": 0, "y": 8, "w": 8, "h": 6},
        "targets": [
          {
            "expr": "histogram_quantile(0.95, rate(ai_api_request_duration_seconds_bucket[5m]))",
            "legendFormat": "P95 Latency"
          }
        ],
        "fieldConfig": {
          "defaults": {
            "thresholds": {
              "mode": "absolute",
              "steps": [
                {"color": "green", "value": null},
                {"color": "yellow", "value": 2},
                {"color": "red", "value": 5}
              ]
            },
            "unit": "s"
          }
        }
      },
      {
        "title": "Remaining Credits",
        "type": "stat",
        "gridPos": {"x": 8, "y": 8, "w": 8, "h": 6},
        "targets": [
          {
            "expr": "ai_api_remaining_credits",
            "legendFormat": "Credits"
          }
        ]
      },
      {
        "title": "Rate Limit Status",
        "type": "gauge",
        "gridPos": {"x": 16, "y": 8, "w": 8, "h": 6},
        "targets": [
          {
            "expr": "ai_api_rate_limit_remaining",
            "legendFormat": "{{model}}"
          }
        ]
      }
    ]
  }
}

ขั้นตอนที่ 4: ตั้งค่า Alert Notifications

แก้ไขไฟล์ alertmanager.yml เพื่อส่ง Alert ไปยังช่องทางที่ต้องการ:

global:
  resolve_timeout: 5m

route:
  group_by: ['alertname', 'severity']
  group_wait: 10s
  group_interval: 10s
  repeat_interval: 12h
  receiver: 'multi-notifications'
  
receivers:
  - name: 'multi-notifications'
    
    # Email notification
    email_configs:
      - to: '[email protected]'
        send_resolved: true
        smarthost: 'smtp.gmail.com:587'
        auth_username: '[email protected]'
        auth_password: 'your-app-password'
        
    # Slack notification  
    slack_configs:
      - api_url: 'https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK'
        channel: '#ai-api-alerts'
        send_resolved: true
        title: '{{ if eq .Status "firing" }}🔥{{ else }}✅{{ end }} {{ .GroupLabels.alertname }}'
        text: |
          {{ range .Alerts }}
          *Alert:* {{ .Annotations.summary }}
          *Description:* {{ .Annotations.description }}
          *Severity:* {{ .Labels.severity }}
          *Model:* {{ .Labels.model }}
          *Time:* {{ .StartsAt.Format "2006-01-02 15:04:05" }}
          {{ end }}

    # Line Notify (ใช้ Line Notify Token)
    webhook_configs:
      - url: 'https://notify-api.line.me/api/notify'
        headers:
          Authorization: 'Bearer YOUR_LINE_NOTIFY_TOKEN'
        max_alerts: 10

ราคาและค่าใช้จ่าย

สำหรับผู้ที่กำลังมองหา AI API ที่คุ้มค่า สมัครที่นี่ HolySheep AI มีราคาที่ประหยัดมาก:

รองรับการชำระเงินผ่าน WeChat และ Alipay พร้อมเครดิตฟรีเมื่อลงทะเบียน ความหน่วงต่ำกว่า 50ms สำหรับการตอบสนองที่รวดเร็ว

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

กรณีที่ 1: "ConnectionError: timeout after 30000ms"

สาเหตุ: เกิดจาก API Server ตอบสนองช้าเกินไป หรือ Network มีปัญหา หรือ API Key ถูก Block

วิธีแก้ไข: เพิ่ม Retry Logic และ Timeout ที่เหมาะสมในโค้ด:

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

def create_session_with_retry():
    """สร้าง Session ที่มี Retry Logic ในตัว"""
    session = requests.Session()
    
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,  # Delay: 1s, 2s, 4s
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["HEAD", "GET", "OPTIONS", "POST"]
    )
    
    adapter = HTTPAdapter(
        max_retries=retry_strategy,
        pool_connections=10,
        pool_maxsize=20
    )
    
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    return session


ใช้งาน

session = create_session_with_retry()

try:
    response = session.post(
        f"{BASE_URL}/chat/completions",
        headers=headers,
        json=payload,
        timeout=(5, 30)  # (connect_timeout, read_timeout)
    )
except requests.exceptions.Timeout:
    # Log และ Alert
    print("Request timeout - switching to backup API")
    # fallback to alternative provider
except requests.exceptions.ConnectionError:
    print("Connection error - check network and API status")

กรณีที่ 2: "401 Unauthorized" หลังจากใช้งานไปได้สักระยะ

สาเหตุ: API Key หมดอายุ, Key ถูก Revoke, หรือ Permission เปลี่ยนแปลง

วิธีแก้ไข: เพิ่มการตรวจสอบ Token Validity และ Fallback:

import hashlib
import time


Cache API Key validity

_key_validity_cache = {
    'is_valid': None,
    'checked_at': 0,
    'cache_ttl': 300  # 5 นาที
}

def validate_api_key(api_key: str) -> bool:
    """ตรวจสอบความถูกต้องของ API Key พร้อม Caching"""
    current_time = time.time()
    
    # ถ้าเคยตรวจสอบแล้วในช่วง TTL ใช้ค่าเดิม
    if (_key_validity_cache['is_valid'] is not None and 
        current_time - _key_validity_cache['checked_at'] < _key_validity_cache['cache_ttl']):
        return _key_validity_cache['is_valid']
    
    try:
        headers = {"Authorization": f"Bearer {api_key}"}
        response = requests.get(
            f"{BASE_URL}/models",
            headers=headers,
            timeout=5
        )
        
        is_valid = response.status_code == 200
        _key_validity_cache['is_valid'] = is_valid
        _key_validity_cache['checked_at'] = current_time
        
        return is_valid
        
    except Exception as e:
        print(f"Error validating API key: {e}")
        return _key_validity_cache['is_valid'] if _key_validity_cache['is_valid'] is not None else False

def make_api_request_with_fallback(payload: dict, primary_key: str, backup_key: str = None):
    """ส่ง Request พร้อม Fallback ไปยัง Key สำรอง"""
    
    for attempt_key in [primary_key, backup_key]:
        if not attempt_key:
            continue
            
        if not validate_api_key(attempt_key):
            print(f"API Key validation failed for {attempt_key[:10]}...")
            continue
            
        headers = {
            "Authorization": f"Bearer {attempt_key}",
            "Content-Type": "application/json"
        }
        
        try:
            response = requests.post(
                f"{BASE_URL}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            
            if response.status_code == 401:
                # Mark key as invalid
                _key_validity_cache['is_valid'] = False
                _key_validity_cache['checked_at'] = 0
                print(f"Key {attempt_key[:10]}... returned 401")
                continue
                
            return response
            
        except Exception as e:
            print(f"Request failed with {attempt_key[:10]}...: {e}")
            continue
    
    raise Exception("All API keys failed")

กรณีที่ 3: "429 Rate Limit Exceeded" ทำให้ระบบหยุดทำงาน

สาเหตุ: ส่ง Request เกินจำนวนที่กำหนดต่อนาที หรือไม่ได้ใช้ Exponential Backoff

วิธีแก้ไข: ใช้ Rate Limiter และ Queue สำหรับ Request:

import threading
import time
from collections import deque
from datetime import datetime, timedelta

class RateLimiter:
    """Token Bucket Rate Limiter สำหรับ API Requests"""
    
    def __init__(self, max_requests: int, time_window: int):
        self.max_requests = max_requests
        self.time_window = time_window  # วินาที
        self.requests = deque()
        self.lock = threading.Lock()
    
    def acquire(self) -> bool:
        """รอจนกว่าจะสามารถส่ง Request ได้"""
        with self.lock:
            now = datetime.now()
            cutoff = now - timedelta(seconds=self.time_window)
            
            # ลบ Request ที่เก่ากว่า time_window
            while self.requests and self.requests[0] < cutoff:
                self.requests.popleft()
            
            if len(self.requests) < self.max_requests:
                self.requests.append(now)
                return True
            
            # คำนวณเวลารอ
            wait_time = (self.requests[0] - cutoff).total_seconds()
            return False
    
    def wait_and_acquire(self):
        """รอจนกว่าจะสามารถ Acquire ได้"""
        while not self.acquire():
            time.sleep(0.1)

class APIRequestQueue:
    """Queue สำหรับจัดการ API Requests อย่างมีประสิทธิภาพ"""
    
    def __init__(self, requests_per_minute: int = 60):
        self.rate_limiter = RateLimiter(requests_per_minute, 60)
        self.queue = deque()
        self.results = {}
        self.lock = threading.Lock()
        self.processing = True
        
    def add_request(self, request_id: str, payload: dict, callback=None):
        """เพิ่ม Request เข้า Queue"""
        with self.lock:
            self.queue.append({
                'id': request_id,
                'payload': payload,
                'callback': callback,
                'added_at': time.time()
            })
    
    def process_queue(self):
        """Process Request จาก Queue"""
        while self.processing:
            if not self.queue:
                time.sleep(0.1)
                continue
            
            self.rate_limiter.wait_and_acquire()
            
            with self.lock:
                if not self.queue:
                    continue
                    
                request = self.queue.popleft()
            
            try:
                headers = {
                    "Authorization": f"Bearer {API_KEY}",
                    "Content-Type": "application/json"
                }
                
                response = requests.post(
                    f"{BASE_URL}/chat/completions",
                    headers=headers,
                    json=request['payload'],
                    timeout=30
                )
                
                if response.status_code == 429:
                    # Rate limited - ใส่กลับเข้า Queue
                    with self.lock:
                        self.queue.appendleft(request)
                    time.sleep(2)  # รอ 2 วินาทีก่อนลองใหม่
                    continue
                
                self.results[request['id']] = response.json()
                
                if request['callback']:
                    request['callback'](response.json())
                    
            except Exception as e:
                print(f"Request {request['id']} failed: {e}")
                self.results[request['id']] = {'error': str(e)}


ใช้งาน

api_queue = APIRequestQueue(requests_per_minute=50)

def my_callback(result):
    print(f"Request completed: {result}")

api_queue.add_request(
    'req_001',
    {'model': 'gpt-4.1', 'messages': [{'role': 'user', 'content': 'Hello'}]},
    callback=my_callback
)

สคริปต์ Docker Compose สำหรับ Production

รวมทุกอย่างเข้าด้วยกันด้วย Docker Compose:

version: '3.8'

services:
  prometheus:
    image: prom/prometheus:latest
    container_name: prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - ./alert_rules.yml:/etc/prometheus/alert_rules.yml
      - prometheus_data:/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.path=/prometheus'
      - '--web.enable-lifecycle'
    restart: unless-stopped

  alertmanager:
    image: prom/alertmanager:latest
    container_name: alertmanager
    ports:
      - "9093:9093"
    volumes:
      - ./alertmanager.yml:/etc/alertmanager/alertmanager.yml
    restart: unless-stopped

  grafana:
    image: grafana/grafana:latest
    container_name: grafana
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_USER=admin
      - GF_SECURITY_ADMIN_PASSWORD=your_secure_password
      - GF_USERS_ALLOW_SIGN_UP=false
    volumes:
      - grafana_data:/var/lib/grafana
      - ./grafana/provisioning:/etc/grafana/provisioning
    restart: unless-stopped

  ai-api-exporter:
    build:
      context: .
      dockerfile: Dockerfile.exporter
    container_name: ai-api

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง