บทนำ: วันที่ทำให้ฉันตื่นกลางดึกเพราะบิล AI $2,847

เช้าวันจันทร์เดือนพฤศจิกายน ผมตื่นขึ้นมาเห็นอีเมลจากผู้ให้บริการ AI API รายหนึ่งแจ้งว่าบัญชีของบริษัทถูกเรียกเก็บค่าบริการไปแล้ว $2,847.07 ในสัปดาห์เดียว สาเหตุคือโค้ดที่ทีม Dev ทำ feature ใหม่มี bug ทำให้เกิด infinite loop ที่เรียก API ซ้ำๆ โดยไม่รู้ตัว

นี่คือประสบการณ์จริงที่ทำให้ผมเข้าใจว่า การตรวจสอบต้นทุน AI API ไม่ใช่ทางเลือก แต่เป็นสิ่งจำเป็นเร่งด่วน บทความนี้จะสอนวิธีสร้างระบบ Budget Alert และ Dashboard สำหรับ HolySheep AI โดยเฉพาะ พร้อมโค้ดตัวอย่างที่รันได้จริง

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

AI API มีความแตกต่างจาก cloud service ทั่วไปอย่างมาก:

การมีระบบ แจ้งเตือนงบประมาณ ช่วยให้คุณ:

การตั้งค่าระบบ Cost Monitoring พื้นฐาน

1. การติดตั้ง Python Client Library

# ติดตั้ง library ที่จำเป็น
pip install requests python-dotenv prometheus-client grafana-api


สร้างไฟล์ .env สำหรับเก็บ API key

cat > .env << 'EOF'
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
BUDGET_WARNING_THRESHOLD=50.00  # แจ้งเตือนเมื่อค่าใช้จ่ายเกิน $50
BUDGET_CRITICAL_THRESHOLD=100.00  # Critical alert เมื่อเกิน $100
CHECK_INTERVAL_MINUTES=5
EOF

2. สร้าง Cost Tracker Class

import requests
import time
import json
from datetime import datetime, timedelta
from collections import defaultdict

class HolySheepCostTracker:
    """
    ระบบติดตามค่าใช้จ่าย AI API แบบ Real-time
    รองรับ: HolySheep AI (https://api.holysheep.ai/v1)
    """
    
    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.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
        
        # เก็บประวัติการใช้งาน
        self.usage_history = []
        self.daily_costs = defaultdict(float)
        self.model_costs = defaultdict(lambda: {"requests": 0, "cost": 0.0})
        
        # ราคา per million tokens (อัปเดต มกราคม 2026)
        self.pricing = {
            "gpt-4.1": 8.00,           # $8/MTok
            "claude-sonnet-4.5": 15.00, # $15/MTok
            "gemini-2.5-flash": 2.50,   # $2.50/MTok
            "deepseek-v3.2": 0.42       # $0.42/MTok
        }
    
    def calculate_token_cost(self, model: str, prompt_tokens: int, 
                            completion_tokens: int) -> float:
        """คำนวณค่าใช้จ่ายจากจำนวน tokens"""
        total_tokens = prompt_tokens + completion_tokens
        price_per_million = self.pricing.get(model, 8.00)  # default to GPT-4.1
        return (total_tokens / 1_000_000) * price_per_million
    
    def log_usage(self, model: str, prompt_tokens: int, 
                  completion_tokens: int, request_id: str = None):
        """บันทึกการใช้งาน API"""
        cost = self.calculate_token_cost(model, prompt_tokens, completion_tokens)
        
        record = {
            "timestamp": datetime.now().isoformat(),
            "model": model,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "cost": round(cost, 6),
            "request_id": request_id or f"req_{int(time.time() * 1000)}"
        }
        
        self.usage_history.append(record)
        self.daily_costs[datetime.now().date()] += cost
        self.model_costs[model]["requests"] += 1
        self.model_costs[model]["cost"] += cost
        
        return record
    
    def get_daily_spend(self, target_date: datetime = None) -> float:
        """ดึงค่าใช้จ่ายรายวัน"""
        if target_date is None:
            target_date = datetime.now().date()
        return self.daily_costs.get(target_date, 0.0)
    
    def get_monthly_spend(self, year: int = None, month: int = None) -> float:
        """ดึงค่าใช้จ่ายรายเดือน"""
        if year is None:
            year = datetime.now().year
        if month is None:
            month = datetime.now().month
        
        total = 0.0
        for date, cost in self.daily_costs.items():
            if date.year == year and date.month == month:
                total += cost
        return total
    
    def check_budget_alert(self, daily_limit: float, 
                          monthly_limit: float) -> dict:
        """ตรวจสอบสถานะงบประมาณและส่ง alert"""
        today_spend = self.get_daily_spend()
        monthly_spend = self.get_monthly_spend()
        
        alerts = []
        
        # Daily budget check
        if today_spend >= monthly_limit * 0.5:  # 50% ของ monthly limit
            alerts.append({
                "level": "CRITICAL",
                "message": f"ค่าใช้จ่ายวันนี้ ${today_spend:.2f} เกินครึ่งหนึ่งของงบเดือน",
                "today_spend": today_spend,
                "limit": daily_limit
            })
        elif today_spend >= daily_limit:
            alerts.append({
                "level": "WARNING",
                "message": f"ค่าใช้จ่ายวันนี้ ${today_spend:.2f} เกินงบประมาณรายวัน",
                "today_spend": today_spend,
                "limit": daily_limit
            })
        
        # Monthly budget check
        if monthly_spend >= monthly_limit:
            alerts.append({
                "level": "CRITICAL",
                "message": f"ค่าใช้จ่ายเดือนนี้ ${monthly_spend:.2f} เกินงบประมาณ",
                "monthly_spend": monthly_spend,
                "limit": monthly_limit
            })
        
        return {
            "has_alerts": len(alerts) > 0,
            "alerts": alerts,
            "today_spend": today_spend,
            "monthly_spend": monthly_spend,
            "daily_limit": daily_limit,
            "monthly_limit": monthly_limit
        }
    
    def generate_usage_report(self) -> str:
        """สร้างรายงานการใช้งาน"""
        report = f"""
╔══════════════════════════════════════════════════════════════╗
║          AI API USAGE REPORT - {datetime.now().strftime('%Y-%m-%d')}                  ║
╠══════════════════════════════════════════════════════════════╣
║  ค่าใช้จ่ายวันนี้:     ${self.get_daily_spend():>10.4f}                          ║
║  ค่าใช้จ่ายเดือนนี้:   ${self.get_monthly_spend():>10.4f}                          ║
╠══════════════════════════════════════════════════════════════╣
║  การใช้งานแยกตาม Model:                                    ║
"""
        for model, data in self.model_costs.items():
            report += f"║    {model:<25} {data['requests']:>5} requests  ${data['cost']:>10.4f}  ║\n"
        
        report += "╚══════════════════════════════════════════════════════════════╝"
        return report


ตัวอย่างการใช้งาน

tracker = HolySheepCostTracker(
    api_key="YOUR_HOLYSHEEP_API_KEY",
    base_url="https://api.holysheep.ai/v1"
)


จำลองการใช้งาน

tracker.log_usage("gpt-4.1", 1500, 500)
tracker.log_usage("deepseek-v3.2", 2000, 300)
print(tracker.generate_usage_report())


ตรวจสอบงบประมาณ

budget_status = tracker.check_budget_alert(
    daily_limit=10.00,
    monthly_limit=100.00
)
print(f"Alert Status: {budget_status['has_alerts']}")
if budget_status['has_alerts']:
    for alert in budget_status['alerts']:
        print(f"  [{alert['level']}] {alert['message']}")

3. ระบบ Budget Alert แบบ Real-time

import smtplib
import asyncio
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from dataclasses import dataclass
from typing import Callable, Optional
import logging

@dataclass
class BudgetAlert:
    level: str  # WARNING, CRITICAL, EMERGENCY
    message: str
    current_spend: float
    threshold: float
    percentage: float
    timestamp: str

class BudgetAlertSystem:
    """
    ระบบแจ้งเตือนงบประมาณ AI API แบบ Real-time
    รองรับหลายช่องทาง: Email, LINE, Slack, Webhook
    """
    
    def __init__(self):
        self.alert_history = []
        self.handlers = []
        self.cooldown_period = 300  # วินาที ระหว่าง alert ซ้ำ
        
    def add_handler(self, handler: Callable[[BudgetAlert], None]):
        """เพิ่ม handler สำหรับส่ง alert"""
        self.handlers.append(handler)
    
    async def send_alert(self, alert: BudgetAlert):
        """ส่ง alert ไปยังทุกช่องทางที่ตั้งค่าไว้"""
        # ตรวจสอบ cooldown
        if self._should_skip_alert(alert):
            logging.info(f"Skip duplicate alert: {alert.message}")
            return
        
        self.alert_history.append(alert)
        
        for handler in self.handlers:
            try:
                if asyncio.iscoroutinefunction(handler):
                    await handler(alert)
                else:
                    handler(alert)
            except Exception as e:
                logging.error(f"Alert handler error: {e}")
    
    def _should_skip_alert(self, alert: BudgetAlert) -> bool:
        """ตรวจสอบว่าควรข้าม alert นี้หรือไม่ (cooldown)"""
        now = datetime.now()
        for prev_alert in self.alert_history[-10:]:  # ดู 10 alert ล่าสุด
            if (prev_alert.level == alert.level and 
                prev_alert.message == alert.message):
                prev_time = datetime.fromisoformat(prev_alert.timestamp)
                if (now - prev_time).total_seconds() < self.cooldown_period:
                    return True
        return False


Email Alert Handler

def email_alert_handler(alert: BudgetAlert):
    """ส่ง alert ทาง Email"""
    smtp_server = "smtp.gmail.com"
    smtp_port = 587
    sender_email = "[email protected]"
    sender_password = "your-app-password"
    recipient_emails = ["[email protected]", "[email protected]"]
    
    msg = MIMEMultipart("alternative")
    msg["Subject"] = f"[{alert.level}] AI API Budget Alert - ${alert.current_spend:.2f}"
    msg["From"] = sender_email
    
    colors = {
        "WARNING": "#FFA500",
        "CRITICAL": "#FF0000",
        "EMERGENCY": "#8B0000"
    }
    
    html_content = f"""
    
    
        

            
⚠️ AI API Budget Alert

        

        

            
ระดับความรุนแรง: {alert.level}

            
ข้อความ: {alert.message}

            
ค่าใช้จ่ายปัจจุบัน: ${alert.current_spend:.4f}

            
Threshold: ${alert.threshold:.2f}

            
ใช้ไปแล้ว: {alert.percentage:.1f}%

            
เวลา: {alert.timestamp}

        

        

            
                เปิด Dashboard
            
        

    
    
    """
    
    msg.attach(MIMEText(html_content, "html"))
    
    try:
        with smtplib.SMTP(smtp_server, smtp_port) as server:
            server.starttls()
            server.login(sender_email, sender_password)
            for recipient in recipient_emails:
                msg["To"] = recipient
                server.sendmail(sender_email, recipient, msg.as_string())
        logging.info(f"Email alert sent to {recipient_emails}")
    except Exception as e:
        logging.error(f"Failed to send email: {e}")


LINE Webhook Handler

def line_webhook_handler(alert: BudgetAlert):
    """ส่ง alert ไปยัง LINE Notify"""
    line_token = "your-line-notify-token"
    
    emoji_map = {
        "WARNING": "⚠️",
        "CRITICAL": "🚨",
        "EMERGENCY": "🔴"
    }
    
    message = f"""
{emoji_map.get(alert.level, '📢')} *AI API Budget Alert*

*ระดับ:* {alert.level}
*ข้อความ:* {alert.message}
*ค่าใช้จ่าย:* ${alert.current_spend:.4f}
*Threshold:* ${alert.threshold:.2f}
*ใช้ไป:* {alert.percentage:.1f}%
"""
    
    response = requests.post(
        "https://notify-api.line.me/api/notify",
        headers={"Authorization": f"Bearer {line_token}"},
        data={"message": message}
    )
    return response.status_code == 200


ตัวอย่างการใช้งานระบบ Alert

async def monitor_budget_loop():
    """Loop หลักสำหรับตรวจสอบงบประมาณ"""
    tracker = HolySheepCostTracker(api_key="YOUR_HOLYSHEEP_API_KEY")
    alert_system = BudgetAlertSystem()
    
    # เพิ่ม handlers
    alert_system.add_handler(email_alert_handler)
    alert_system.add_handler(line_webhook_handler)
    
    # ตั้งค่า thresholds
    daily_budget = 20.00
    monthly_budget = 200.00
    
    while True:
        # ตรวจสอบงบประมาณ
        status = tracker.check_budget_alert(daily_budget, monthly_budget)
        
        if status['has_alerts']:
            for alert_data in status['alerts']:
                alert = BudgetAlert(
                    level=alert_data['level'],
                    message=alert_data['message'],
                    current_spend=alert_data.get('today_spend', 
                                      alert_data.get('monthly_spend', 0)),
                    threshold=alert_data.get('limit', monthly_budget),
                    percentage=(alert_data.get('today_spend', 0) / 
                               alert_data.get('limit', monthly_budget) * 100),
                    timestamp=datetime.now().isoformat()
                )
                await alert_system.send_alert(alert)
        
        # รอ 5 นาที
        await asyncio.sleep(300)


รัน monitor

if __name__ == "__main__":
    asyncio.run(monitor_budget_loop())

Dashboard สำหรับแสดงผลการใช้งานแบบ Real-time

from prometheus_client import Counter, Gauge, Histogram, start_http_server
from flask import Flask, jsonify, render_template
import threading


Prometheus metrics

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

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

CURRENT_SPEND = Gauge(
    'ai_api_current_spend_dollars',
    'Current spend in dollars',
    ['period']  # daily, monthly
)

TOKEN_USAGE = Counter(
    'ai_api_tokens_total',
    'Total tokens used',
    ['model', 'type']  # type: prompt, completion
)


Flask Dashboard

app = Flask(__name__)

@app.route('/')
def dashboard():
    """หน้า Dashboard หลัก"""
    return render_template('dashboard.html')

@app.route('/api/cost-summary')
def cost_summary():
    """API สำหรับดึงข้อมูลสรุปค่าใช้จ่าย"""
    return jsonify({
        "daily_spend": CURRENT_SPEND.labels(period='daily')._value.get(),
        "monthly_spend": CURRENT_SPEND.labels(period='monthly')._value.get(),
        "requests_today": REQUEST_COUNT.labels(model='all', status='success')._value.get()
    })

@app.route('/api/usage-by-model')
def usage_by_model():
    """API สำหรับดึงข้อมูลการใช้งานแยกตาม model"""
    # ดึงข้อมูลจาก tracker
    tracker = HolySheepCostTracker(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    models_data = []
    for model, data in tracker.model_costs.items():
        models_data.append({
            "model": model,
            "requests": data["requests"],
            "cost": round(data["cost"], 4)
        })
    
    return jsonify({"models": models_data})


HTML Template สำหรับ Dashboard

DASHBOARD_HTML = '''



    
    AI API Cost Dashboard
    
    


    

        

            📊 AI API Cost Dashboard - HolySheep AI
        

        
        
        

            

                
ค่าใช้จ่ายวันนี้

                
$0.00

            

            

                
ค่าใช้จ่ายเดือนนี้

                
$0.00

            

            

                
จำนวน Requests

                
0

            

            

                
Budget Remaining

                
$0.00

            

        

        
        
        

            

                
ค่าใช้จ่ายรายวัน

                
            

            

                
ค่าใช้จ่ายแยกตาม Model

                
            

        

        
        
        

            
Recent API Requests

            

                
                    

                        Timestamp
                        Model
                        Prompt Tokens
                        Completion Tokens
                        Cost
                    

                
                
                
            

        

    

    
    


'''

app.template_folder = 'templates'
import os
os.makedirs('templates', exist_ok=True)
with open('templates/dashboard.html', 'w', encoding='utf-8') as f:
    f.write(DASHBOARD_HTML)

if __name__ == '__main__':
    # Start Prometheus metrics server on port 9090
    start_http_server(9090)
    
    # Start Flask dashboard on port 5000
    app.run(host='0.0.0.0', port=5000, debug=False)

การผสาน Cost Tracking เข้ากับ API Wrapper

import time
import functools
from typing import Any, Dict, Optional

class TrackedAPIWrapper:
    """
    Wrapper สำหรับ AI API ที่มีระบบติดตามค่าใช้จ่ายในตัว
    ใช้กับ HolySheep AI (base_url: https://api.holysheep.ai/v1)
    """
    
    def __init__(self, api_key: str, cost_tracker: HolySheepCostTracker):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.cost_tracker = cost_tracker
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def track_cost(self, model: str):
        """Decorator สำหรับ track ค่าใช้จ่ายของ API call"""
        def decorator(func):
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                start_time = time.time()
                response = None
                error = None
                
                try:
                    response = func(*args, **kwargs)
                    return response
                except Exception as e:
                    error = e
                    raise
                finally:
                    # คำนวณค่าใช้จ่าย
                    latency = time.time() - start_time
                    
                    # ดึง token usage จาก response
                    if response and hasattr(response, 'json'):
                        try:
                            data = response.json()
                            usage = data.get('usage', {})
                            prompt_tokens = usage.get('prompt_tokens', 0)
                            completion_tokens = usage.get('completion_tokens', 0)
                            
                            # บันทึกลง tracker
                            record = self.cost_tracker.log_usage(
                                model=model,
                                prompt_tokens=prompt_tokens,
                                completion_tokens=completion_tokens,
                                request_id=data.get('id', f"auto_{int(time.time())}")
                            )
                            
                            # อัปเดต Prometheus metrics
                            REQUEST_COUNT.labels(model=model, status='success').inc()
                            TOKEN_USAGE.labels(model=model, type='prompt').inc(prompt_tokens)
                            TOKEN_USAGE.labels(model=model, type='completion').inc(completion_tokens)
                            
                        except Exception:
                            pass
                    
                    REQUEST_LATENCY.labels(model=model).observe(latency)
                    
                    if error:
                        REQUEST_COUNT.labels(model=model, status='error').inc()
            
            return wrapper
        return decorator
    
    @track_cost

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