ในโปรเจ็กต์ล่าสุดของผม ทีมเซลส์ของลูกค้าอีคอมเมิร์ซรายใหญ่ประสบปัญหาค่าใช้จ่าย AI แชทบอทพุ่งสูงขึ้น 320% ในช่วง Super Sale ทำให้บิลรายเดือนพุ่งจาก $800 เป็น $3,500 ภายใน 48 ชั่วโมง ปัญหานี้สอนผมว่าการไม่มีระบบติดตามที่ดีเทียบกับการขับรถบนทางด่วนโดยไม่มีมาตรวัดความเร็ว — คุณจะรู้ว่ามีปัญหาก็ต่อเมื่อมาถึงปลายทางแล้ว

บทความนี้จะสอนวิธีสร้างระบบมอนิเตอร์ที่ครอบคลุม ตั้งแต่การเก็บข้อมูลการใช้งาน ไปจนถึงการสร้าง Dashboard ที่แสดง Cost per Request, Token Usage และ Latency แบบเรียลไทม์ โดยใช้ HolySheep AI ซึ่งมีความเร็วตอบสนองต่ำกว่า 50ms และอัตราค่าบริการที่ประหยัดกว่า 85% เมื่อเทียบกับผู้ให้บริการรายอื่น

ทำไมต้องมีระบบติดตามค่าใช้จ่าย AI API

จากประสบการณ์ของผมในการดูแลระบบ AI สำหรับองค์กรต่างๆ มี 3 เหตุผลหลักที่ทำให้การมอนิเตอร์มีความสำคัญ:

โครงสร้างพื้นฐานของระบบมอนิเตอร์

ระบบที่ดีต้องเก็บข้อมูล 4 ประเภทหลัก:

การสร้าง API Wrapper พร้อมระบบ Logging

ขั้นตอนแรกคือการสร้าง Wrapper ที่ครอบ API Call ทั้งหมดเพื่อเก็บข้อมูลอัตโนมัติ:

import requests
import time
import json
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
import sqlite3

@dataclass
class APIUsageRecord:
    request_id: str
    timestamp: str
    model: str
    endpoint: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    cost_usd: float
    latency_ms: float
    status_code: int
    error_message: Optional[str] = None


ราคาต่อ Million Tokens (2026)

MODEL_PRICING = {
    "gpt-4.1": {"input": 2.0, "output": 6.0},           # $8/MTok total
    "claude-sonnet-4.5": {"input": 3.0, "output": 12.0},  # $15/MTok total
    "gemini-2.5-flash": {"input": 0.35, "output": 2.15},  # $2.50/MTok total
    "deepseek-v3.2": {"input": 0.27, "output": 0.15},     # $0.42/MTok total
}

class HolySheepMonitor:
    def __init__(self, api_key: str, db_path: str = "usage.db"):
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = api_key
        self.db_path = db_path
        self._init_database()
    
    def _init_database(self):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS api_usage (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                request_id TEXT UNIQUE,
                timestamp TEXT,
                model TEXT,
                endpoint TEXT,
                prompt_tokens INTEGER,
                completion_tokens INTEGER,
                total_tokens INTEGER,
                cost_usd REAL,
                latency_ms REAL,
                status_code INTEGER,
                error_message TEXT
            )
        """)
        cursor.execute("""
            CREATE INDEX IF NOT EXISTS idx_timestamp ON api_usage(timestamp)
        """)
        cursor.execute("""
            CREATE INDEX IF NOT EXISTS idx_model ON api_usage(model)
        """)
        conn.commit()
        conn.close()
    
    def calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> float:
        if model not in MODEL_PRICING:
            return 0.0
        pricing = MODEL_PRICING[model]
        input_cost = (prompt_tokens / 1_000_000) * pricing["input"]
        output_cost = (completion_tokens / 1_000_000) * pricing["output"]
        return round(input_cost + output_cost, 6)
    
    def chat_completions(self, model: str, messages: List[Dict], 
                         temperature: float = 0.7, max_tokens: int = 1000) -> Dict:
        import uuid
        
        request_id = str(uuid.uuid4())
        start_time = time.time()
        error_message = None
        status_code = 200
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            latency_ms = round((time.time() - start_time) * 1000, 2)
            status_code = response.status_code
            
            if response.status_code == 200:
                result = response.json()
                usage = result.get("usage", {})
                prompt_tokens = usage.get("prompt_tokens", 0)
                completion_tokens = usage.get("completion_tokens", 0)
                total_tokens = usage.get("total_tokens", 0)
                cost_usd = self.calculate_cost(model, prompt_tokens, completion_tokens)
                
                record = APIUsageRecord(
                    request_id=request_id,
                    timestamp=datetime.now().isoformat(),
                    model=model,
                    endpoint="/v1/chat/completions",
                    prompt_tokens=prompt_tokens,
                    completion_tokens=completion_tokens,
                    total_tokens=total_tokens,
                    cost_usd=cost_usd,
                    latency_ms=latency_ms,
                    status_code=status_code
                )
                self._save_record(record)
                return result
            else:
                error_message = response.text
                raise Exception(f"API Error: {response.status_code}")
                
        except Exception as e:
            latency_ms = round((time.time() - start_time) * 1000, 2)
            error_message = str(e)
            record = APIUsageRecord(
                request_id=request_id,
                timestamp=datetime.now().isoformat(),
                model=model,
                endpoint="/v1/chat/completions",
                prompt_tokens=0,
                completion_tokens=0,
                total_tokens=0,
                cost_usd=0.0,
                latency_ms=latency_ms,
                status_code=status_code if status_code != 200 else 500,
                error_message=error_message
            )
            self._save_record(record)
            raise
    
    def _save_record(self, record: APIUsageRecord):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute("""
            INSERT OR REPLACE INTO api_usage 
            (request_id, timestamp, model, endpoint, prompt_tokens, 
             completion_tokens, total_tokens, cost_usd, latency_ms, 
             status_code, error_message)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            record.request_id, record.timestamp, record.model, 
            record.endpoint, record.prompt_tokens, record.completion_tokens,
            record.total_tokens, record.cost_usd, record.latency_ms,
            record.status_code, record.error_message
        ))
        conn.commit()
        conn.close()


วิธีใช้งาน

monitor = HolySheepMonitor(api_key="YOUR_HOLYSHEEP_API_KEY")

response = monitor.chat_completions(
    model="deepseek-v3.2",
    messages=[{"role": "user", "content": "อธิบายเรื่อง SEO"}]
)

print(f"Response: {response['choices'][0]['message']['content']}")

Dashboard ด้วย Streamlit

หลังจากเก็บข้อมูลแล้ว มาสร้าง Dashboard แสดงผลแบบเรียลไทม์กัน:

import streamlit as st
import pandas as pd
import sqlite3
from datetime import datetime, timedelta
import plotly.express as px
import plotly.graph_objects as go

st.set_page_config(page_title="AI Usage Monitor", layout="wide")

def load_data(db_path: str, days: int = 7) -> pd.DataFrame:
    conn = sqlite3.connect(db_path)
    query = f"""
        SELECT * FROM api_usage 
        WHERE timestamp >= datetime('now', '-{days} days')
        ORDER BY timestamp DESC
    """
    df = pd.read_sql_query(query, conn)
    conn.close()
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    return df

def calculate_savings(df: pd.DataFrame) -> dict:
    """คำนวณการประหยัดเมื่อเทียบกับ OpenAI"""
    if df.empty:
        return {"total_cost": 0, "openai_cost": 0, "savings": 0, "savings_percent": 0}
    
    # สมมติใช้ GPT-4o mini เป็น baseline ($0.15/MTok input, $0.60/MTok output)
    baseline_rate = 0.75  # avg $/MTok
    
    total_tokens = df['total_tokens'].sum()
    our_cost = df['cost_usd'].sum()
    baseline_cost = (total_tokens / 1_000_000) * baseline_rate
    
    savings = baseline_cost - our_cost
    savings_percent = (savings / baseline_cost * 100) if baseline_cost > 0 else 0
    
    return {
        "total_cost": our_cost,
        "openai_cost": baseline_cost,
        "savings": savings,
        "savings_percent": savings_percent
    }


Main Dashboard

st.title("📊 AI API Usage Dashboard")


Sidebar filters

with st.sidebar:
    st.header("ตัวกรอง")
    db_path = st.text_input("Database Path", value="usage.db")
    days = st.slider("แสดงข้อมูลย้อนหลัง (วัน)", 1, 30, 7)
    selected_models = st.multiselect(
        "เลือก Model",
        ["deepseek-v3.2", "gemini-2.5-flash", "gpt-4.1", "claude-sonnet-4.5"],
        default=["deepseek-v3.2", "gemini-2.5-flash"]
    )


Load data

try:
    df = load_data(db_path, days)
    
    if not selected_models:
        st.error("กรุณาเลือกอย่างน้อย 1 Model")
    else:
        df_filtered = df[df['model'].isin(selected_models)]
        
        # KPI Cards
        col1, col2, col3, col4 = st.columns(4)
        
        savings = calculate_savings(df_filtered)
        
        col1.metric(
            "💰 ค่าใช้จ่ายรวม (HolySheep)", 
            f"${savings['total_cost']:.4f}"
        )
        col2.metric(
            "📈 ค่าใช้จ่าย (OpenAI ประมาณการ)", 
            f"${savings['openai_cost']:.4f}"
        )
        col3.metric(
            "✅ ประหยัดได้", 
            f"${savings['savings']:.4f}",
            delta=f"{savings['savings_percent']:.1f}%"
        )
        col4.metric(
            "🔢 Total Requests", 
            f"{len(df_filtered):,}"
        )
        
        # Charts
        tab1, tab2, tab3 = st.tabs(["ค่าใช้จ่าย", "Token Usage", "Latency"])
        
        with tab1:
            fig = px.bar(
                df_filtered.groupby('model')['cost_usd'].sum().reset_index(),
                x='model', y='cost_usd',
                title='ค่าใช้จ่ายแยกตาม Model',
                labels={'cost_usd': 'Cost (USD)', 'model': 'Model'}
            )
            st.plotly_chart(fig, use_container_width=True)
            
            # Cost over time
            df_filtered['date'] = df_filtered['timestamp'].dt.date
            fig2 = px.line(
                df_filtered.groupby('date')['cost_usd'].sum().reset_index(),
                x='date', y='cost_usd',
                title='ค่าใช้จ่ายรายวัน'
            )
            st.plotly_chart(fig2, use_container_width=True)
        
        with tab2:
            fig3 = px.bar(
                df_filtered.groupby('model')[['prompt_tokens', 'completion_tokens']].sum().reset_index(),
                x='model', y=['prompt_tokens', 'completion_tokens'],
                title='Token Usage แยกตาม Model',
                barmode='group'
            )
            st.plotly_chart(fig3, use_container_width=True)
        
        with tab3:
            # Filter only successful requests
            df_success = df_filtered[df_filtered['status_code'] == 200]
            
            fig4 = go.Figure()
            for model in selected_models:
                model_data = df_success[df_success['model'] == model]['latency_ms']
                fig4.add_trace(go.Histogram(
                    x=model_data,
                    name=model,
                    nbinsx=30
                ))
            fig4.update_layout(
                title='Latency Distribution',
                xaxis_title='Latency (ms)',
                yaxis_title='Count'
            )
            st.plotly_chart(fig4, use_container_width=True)
            
            # Avg latency by model
            avg_latency = df_success.groupby('model')['latency_ms'].mean()
            st.dataframe(avg_latency.rename("Avg Latency (ms)"))
        
        # Recent requests table
        st.subheader("Recent Requests")
        display_cols = ['timestamp', 'model', 'total_tokens', 'cost_usd', 'latency_ms', 'status_code']
        st.dataframe(df_filtered[display_cols].head(50), use_container_width=True)

except Exception as e:
    st.error(f"Error loading data: {e}")
    st.info("ตรวจสอบว่าไฟล์ database มีอยู่จริงและมีข้อมูล")


Run with: streamlit run dashboard.py

Alert System สำหรับ Cost Spike

นี่คือส่วนที่ช่วยป้องกันปัญหา Bill Shock โดยส่งการแจ้งเตือนเมื่อค่าใช้จ่ายหรือจำนวน Request สูงผิดปกติ:

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from dataclasses import dataclass
from typing import List
import sqlite3
from datetime import datetime, timedelta

@dataclass
class AlertConfig:
    daily_budget_usd: float = 50.0
    hourly_request_threshold: int = 1000
    avg_latency_threshold_ms: float = 2000.0
    error_rate_threshold: float = 0.05  # 5%

class CostAlertSystem:
    def __init__(self, db_path: str, config: AlertConfig):
        self.db_path = db_path
        self.config = config
    
    def check_daily_budget(self) -> List[dict]:
        conn = sqlite3.connect(self.db_path)
        query = """
            SELECT 
                DATE(timestamp) as date,
                SUM(cost_usd) as total_cost,
                COUNT(*) as total_requests
            FROM api_usage
            WHERE timestamp >= datetime('now', '-1 day')
            GROUP BY DATE(timestamp)
        """
        df = pd.read_sql_query(query, conn)
        conn.close()
        
        alerts = []
        for _, row in df.iterrows():
            if row['total_cost'] > self.config.daily_budget_usd:
                alerts.append({
                    'type': 'daily_budget',
                    'message': f"ค่าใช้จ่ายรายวัน ${row['total_cost']:.2f} เกิน Budget ${self.config.daily_budget_usd}",
                    'severity': 'high' if row['total_cost'] > self.config.daily_budget_usd * 2 else 'medium'
                })
        return alerts
    
    def check_hourly_spike(self) -> List[dict]:
        conn = sqlite3.connect(self.db_path)
        query = """
            SELECT 
                strftime('%Y-%m-%d %H:00', timestamp) as hour,
                COUNT(*) as request_count
            FROM api_usage
            WHERE timestamp >= datetime('now', '-2 hours')
            GROUP BY strftime('%Y-%m-%d %H:00', timestamp)
            ORDER BY hour DESC
        """
        df = pd.read_sql_query(query, conn)
        conn.close()
        
        alerts = []
        if len(df) >= 2:
            current_hour = df.iloc[0]['request_count']
            previous_hour = df.iloc[1]['request_count']
            
            if previous_hour > 0:
                spike_ratio = current_hour / previous_hour
                if spike_ratio > 3.0:  # 3x increase
                    alerts.append({
                        'type': 'request_spike',
                        'message': f"Request พุ่ง {spike_ratio:.1f}x จาก {previous_hour} เป็น {current_hour}",
                        'severity': 'high'
                    })
        return alerts
    
    def check_latency(self) -> List[dict]:
        conn = sqlite3.connect(self.db_path)
        query = """
            SELECT 
                model,
                AVG(latency_ms) as avg_latency,
                MAX(latency_ms) as max_latency
            FROM api_usage
            WHERE timestamp >= datetime('now', '-1 hour')
            AND status_code = 200
            GROUP BY model
        """
        df = pd.read_sql_query(query, conn)
        conn.close()
        
        alerts = []
        for _, row in df.iterrows():
            if row['avg_latency'] > self.config.avg_latency_threshold_ms:
                alerts.append({
                    'type': 'high_latency',
                    'message': f"{row['model']}: Avg {row['avg_latency']:.0f}ms, Max {row['max_latency']:.0f}ms",
                    'severity': 'medium'
                })
        return alerts
    
    def check_error_rate(self) -> List[dict]:
        conn = sqlite3.connect(self.db_path)
        query = """
            SELECT 
                COUNT(*) FILTER (WHERE status_code != 200) as error_count,
                COUNT(*) as total_count
            FROM api_usage
            WHERE timestamp >= datetime('now', '-1 hour')
        """
        cursor = conn.execute(query)
        result = cursor.fetchone()
        conn.close()
        
        if result[1] > 0:
            error_rate = result[0] / result[1]
            if error_rate > self.config.error_rate_threshold:
                return [{
                    'type': 'high_error_rate',
                    'message': f"Error Rate {error_rate*100:.1f}% ({result[0]}/{result[1]} requests)",
                    'severity': 'high' if error_rate > 0.1 else 'medium'
                }]
        return []
    
    def send_alert_email(self, alerts: List[dict], recipients: List[str]):
        if not alerts:
            return
        
        # สร้าง HTML content
        html_content = """
        
        
🚨 AI API Usage Alert

        
        """
        for alert in alerts:
            color = {'high': 'red', 'medium': 'orange'}.get(alert['severity'], 'black')
            html_content += f"""
            
            """
        html_content += "

            
TypeMessageSeverity

                {alert['type']}
                {alert['message']}
                {alert['severity'].upper()}
            
"
        
        msg = MIMEMultipart('alternative')
        msg['Subject'] = f"AI Alert: {len(alerts)} issue(s) detected"
        msg['From'] = "[email protected]"
        msg['To'] = ", ".join(recipients)
        msg.attach(MIMEText(html_content, 'html'))
        
        # ส่ง email (ต้องตั้งค่า SMTP server)
        try:
            with smtplib.SMTP('smtp.gmail.com', 587) as server:
                server.starttls()
                # server.login('your-email', 'your-password')
                # server.send_message(msg)
                print("Alert email sent!")
        except Exception as e:
            print(f"Failed to send email: {e}")
    
    def run_all_checks(self) -> List[dict]:
        all_alerts = []
        all_alerts.extend(self.check_daily_budget())
        all_alerts.extend(self.check_hourly_spike())
        all_alerts.extend(self.check_latency())
        all_alerts.extend(self.check_error_rate())
        return all_alerts


ใช้งาน

import pandas as pd

config = AlertConfig(
    daily_budget_usd=100.0,
    hourly_request_threshold=500,
    avg_latency_threshold_ms=1500.0
)

alert_system = CostAlertSystem("usage.db", config)
alerts = alert_system.run_all_checks()

if alerts:
    print(f"พบ {len(alerts)} Alert(s):")
    for alert in alerts:
        print(f"  [{alert['severity'].upper()}] {alert['message']}")
    # alert_system.send_alert_email(alerts, ["[email protected]"])
else:
    print("✓ ไม่พบปัญหาผิดปกติ")

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

1. Error 401: Authentication Failed

สาเหตุ: API Key ไม่ถูกต้องหรือหมดอายุ

# ❌ ผิด - Key มีช่องว่างหรือผิด format
headers = {
    "Authorization": f"Bearer {api_key} ",  # มี space ต่อท้าย
}


✅ ถูก - Key ตรงมาจาก HolySheep Dashboard

headers = {
    "Authorization": f"Bearer {api_key.strip()}",
}


ตรวจสอบ Key format

import re
if not re.match(r'^sk-[a-zA-Z0-9]{32,}$', api_key):
    raise ValueError("Invalid API Key format. Get your key from https://www.holysheep.ai/register")

2. Error 429: Rate Limit Exceeded

สาเหตุ: ส่ง Request เร็วเกินไปหรือเกินโควต้า

import time
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_with_retry(monitor, model, messages):
    try:
        return monitor.chat_completions(model, messages)
    except Exception as e:
        if "429" in str(e) or "rate limit" in str(e).lower():
            print("Rate limited, waiting...")
            time.sleep(5)  # รอ 5 วินาทีก่อน retry
            raise
        raise


หรือใช้ Exponential Backoff

def call_with_backoff(monitor, model, messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            return monitor.chat_completions(model, messages)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            wait_time = 2 ** attempt
            print(f"Retry {attempt + 1}/{max_retries} after {wait_time}s")
            time.sleep(wait_time)

3. Cost สูงผิดปกติจาก Token Counting

สาเหตุ: Model name ไม่ตรงกับ pricing table หรือใช้ Model ที่มีราคาสูงโดยไม่รู้ตัว

# ❌ ผิด - ใช้ชื่อ Model ผิด
MODEL_PRICING = {
    "gpt-4": {"input": 2.0, "output": 6.0},  # gpt-4 ไม่มีใน HolySheep
}


✅ ถูก - ใช้ Model name ที่ถูกต้องจาก HolySheep

MODEL_PRICING = {
    "deepseek-v3.2": {"input": 0.27, "output": 0.15},
    "gemini-2.5-flash": {"input": 0.35, "output": 2.15},
    "gpt-4.1": {"input": 2.0, "output": 6.0},
}


ตรวจสอบ Model ก่อนเรียกใช้

AVAILABLE_MODELS = ["deepseek-v3.2", "gemini-2.5-flash", "gpt-4.1", "claude-sonnet-4.5"]

def validate_model(model: str):
    if model not in AVAILABLE_MODELS:
        raise ValueError(f"Model '{model}' not available. Choose from: {AVAILABLE_MODELS}")
    
    # ประเมินค่าใช้จ่ายล่วงหน้า
    estimated_cost_per_1k_tokens = (
        MODEL_PRICING[model]["input"] + MODEL_PRICING[model]["output"]
    ) / 1000  # per 1K tokens
    return estimated_cost_per_1k_tokens


ทดสอบ

cost = validate_model("deepseek-v3.2")
print(f"Estimated cost: ${cost:.6f} per 1K tokens")

Output: Estimated cost: $0.000420 per 1K tokens

4. Database Lock Error เมื่อมีหลาย Process

สาเหตุ: SQLite ไม่รองรับการเขียนพร้อมกันจากหลาย Process

import sqlite3
import threading
import queue

class ThreadSafeUsageDB:
    def __init__(self, db_path: str):
        self.db_path = db_path
        self.lock = threading.Lock()
        self._init_database()
    
    def _init_database(self):
        conn = sqlite3.connect(self.db_path, check_same_thread=False)
        conn.execute("PRAGMA journal_mode=WAL")  # Write-Ahead Logging
        conn.execute("PRAGMA busy_timeout=5000")  # รอ 5 วินาทีเมื่อ locked
        conn.close()
    
    def save_record(self, record: APIUsageRecord):
        with self.lock:
            conn = sqlite3.connect(self.db_path, timeout=10.0)
            try:
                conn.execute("""
                    INSERT OR REPLACE INTO api_usage 
                    (request_id, timestamp, model, endpoint, prompt_tokens, 
                     completion_tokens, total_tokens, cost_usd, latency_ms, 
                     status_code, error_message)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """, (
                    record.request_id, record.timestamp, record.model,
                    record.endpoint, record.prompt_tokens, record.completion_tokens,
                    record.total_tokens, record.cost_usd, record.latency_ms,
                    record.status_code, record.error_message
                ))
                conn.commit()
            finally:
                conn.close()
    
    def batch_save(self, records: List[APIUsageRecord]):
        with self.lock:
            conn = sqlite3.connect(self.db_path, timeout=30.0)
            try:
                data = [
                    (r.request_id, r.timestamp, r.model, r.endpoint,
                     r.prompt_tokens, r.completion_tokens, r.total_tokens,
                     r.cost_usd, r.latency_ms, r.status_code, r.error_message)
                    for r in records
                ]
                conn.executemany("""
                    INSERT OR REPLACE INTO api_usage VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """, data)
                conn.commit()
            finally:
                conn.close()
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง