เมื่อวานนี้ผมเกือบจะเสียเงินไป 200 ดอลลาร์จากการทดสอบ AI API ในโปรเจกต์ที่ใช้ GPT-4.1 อย่างไม่หวังดี ตื่นมาเช้าวันหนึ่งเจออีเมลจากผู้ให้บริการว่า "ค่าใช้จ่ายของคุณเกินงบประมาณ $150 แล้ว" — ตอนนั้นหัวใจหล่นไปเลยครับ จากเหตุการณ์ครั้งนั้น ผมเลยอยากแชร์วิธีสร้างระบบ monitor ค่าใช้จ่าย AI API ที่ใช้งานได้จริง พร้อมตั้งค่าการแจ้งเตือนก่อนที่บัญชีจะบวมเกินไป โดยใช้ HolySheep AI เป็นตัวอย่างหลัก ซึ่งมีอัตรา ¥1=$1 ประหยัดได้ถึง 85% เมื่อเทียบกับผู้ให้บริการอื่น แถมมีเครดิตฟรีเมื่อลงทะเบียน
ทำไมต้อง Monitor ค่าใช้จ่าย AI API?
ปัญหาหลักของนักพัฒนาหลายคนคือ ไม่มีใครรู้ว่าโค้ดของตัวเองใช้ token ไปเท่าไหร่ จนกว่าจะได้ใบแจ้งหนี้ตอนสิ้นเดือน การสร้าง dashboard ติดตามค่าใช้จ่ายจึงสำคัญมาก โดยเฉพาะเมื่อใช้โมเดลราคาสูงอย่าง GPT-4.1 ($8/MTok) หรือ Claude Sonnet 4.5 ($15/MTok) ซึ่งถ้าปล่อยให้วิ่งโดยไม่มีการควบคุม ค่าไฟฟ้าที่เป็นค่า API อย่างเดียวอาจเกินหลายร้อยดอลลาร์ได้ในชั่วข้ามคืน
โครงสร้างพื้นฐานของระบบ Monitor
ก่อนจะเข้าเนื้อหา ผมอยากให้ทุกคนเข้าใจ flow การทำงานของระบบ monitor ค่าใช้จ่าย AI API ก่อน ซึ่งประกอบด้วย 4 ส่วนหลัก ได้แก่ request logging, token counting, budget tracking และ alert system ซึ่งแต่ละส่วนต้องทำงานร่วมกันอย่างไร้รอยต่อ ผมจึงสร้าง class wrapper ที่ครอบ request ไว้แล้ว track ทุกอย่างโดยอัตโนมัติ
สร้าง Token Tracking Decorator สำหรับ API Calls
ผมเริ่มจากการสร้าง decorator ที่จะ wrap ทุก API call เพื่อนับ token และคำนวณค่าใช้จ่ายโดยอัตโนมัติ โดยใช้ HolySheep AI เป็น endpoint หลัก ซึ่งมี latency เพียง <50ms ทำให้การ monitor ไม่กระทบประสิทธิภาพการทำงานของ app มากนัก
import time
import json
import sqlite3
from datetime import datetime, timedelta
from functools import wraps
from dataclasses import dataclass, field
from typing import Optional, Dict, Any, Callable
import threading
import os
@dataclass
class TokenUsage:
model: str
prompt_tokens: int
completion_tokens: int
total_tokens: int
cost: float
timestamp: datetime = field(default_factory=datetime.now)
request_id: Optional[str] = None
response_time_ms: float = 0.0
@dataclass
class BudgetConfig:
daily_limit: float = 10.0
weekly_limit: float = 50.0
monthly_limit: float = 200.0
alert_threshold_percent: float = 0.8
class AIAPICostTracker:
"""
ระบบติดตามค่าใช้จ่าย AI API แบบครบวงจร
รองรับ HolySheep AI, OpenAI และ Anthropic
"""
# ราคาต่อล้าน tokens (อัปเดตล่าสุด 2026)
PRICING = {
'gpt-4.1': {'prompt': 8.0, 'completion': 8.0},
'gpt-4.1-nano': {'prompt': 0.30, 'completion': 1.20},
'claude-sonnet-4.5': {'prompt': 15.0, 'completion': 15.0},
'claude-3-5-haiku': {'prompt': 0.80, 'completion': 4.0},
'gemini-2.5-flash': {'prompt': 2.50, 'completion': 2.50},
'deepseek-v3.2': {'prompt': 0.42, 'completion': 0.42},
}
def __init__(self, db_path: str = 'api_costs.db', base_url: str = 'https://api.holysheep.ai/v1'):
self.db_path = db_path
self.base_url = base_url
self._lock = threading.Lock()
self._init_database()
def _init_database(self):
"""สร้างตารางสำหรับเก็บข้อมูลการใช้งาน"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS token_usage (
id INTEGER PRIMARY KEY AUTOINCREMENT,
model TEXT NOT NULL,
prompt_tokens INTEGER NOT NULL,
completion_tokens INTEGER NOT NULL,
total_tokens INTEGER NOT NULL,
cost REAL NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
request_id TEXT,
response_time_ms REAL,
session_id TEXT
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS budget_alerts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
threshold_type TEXT NOT NULL,
threshold_amount REAL NOT NULL,
actual_amount REAL NOT NULL,
triggered_at DATETIME DEFAULT CURRENT_TIMESTAMP,
acknowledged BOOLEAN DEFAULT FALSE
)
''')
cursor.execute('''
CREATE INDEX IF NOT EXISTS idx_timestamp ON token_usage(timestamp)
''')
conn.commit()
def calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> float:
"""คำนวณค่าใช้จ่ายจากจำนวน tokens"""
if model not in self.PRICING:
# กรณีโมเดลใหม่ที่ยังไม่มีในระบบ ใช้ราคาเฉลี่ย
return (prompt_tokens + completion_tokens) * 0.00001
pricing = self.PRICING[model]
prompt_cost = (prompt_tokens / 1_000_000) * pricing['prompt']
completion_cost = (completion_tokens / 1_000_000) * pricing['completion']
return round(prompt_cost + completion_cost, 6)
def log_usage(self, usage: TokenUsage, session_id: Optional[str] = None):
"""บันทึกข้อมูลการใช้งานลงฐานข้อมูล"""
with self._lock:
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT INTO token_usage
(model, prompt_tokens, completion_tokens, total_tokens, cost, timestamp, request_id, response_time_ms, session_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (
usage.model,
usage.prompt_tokens,
usage.completion_tokens,
usage.total_tokens,
usage.cost,
usage.timestamp.isoformat(),
usage.request_id,
usage.response_time_ms,
session_id
))
conn.commit()
def get_spending(self, period: str = 'daily') -> float:
"""ดึงข้อมูลค่าใช้จ่ายตามช่วงเวลา"""
with sqlite3.connect(self.db_path) as conn:
cursor = conn.cursor()
if period == 'daily':
start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
elif period == 'weekly':
start = datetime.now() - timedelta(days=7)
elif period == 'monthly':
start = datetime.now() - timedelta(days=30)
else:
start = datetime.min
cursor.execute('''
SELECT SUM(cost) FROM token_usage WHERE timestamp >= ?
''', (start.isoformat(),))
result = cursor.fetchone()
return result[0] if result[0] else 0.0
def get_usage_by_model(self, limit: int = 10) -> list:
"""ดึงสถิติการใช้งานแยกตามโมเดล"""
with sqlite3.connect(self.db_path) as conn:
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute('''
SELECT
model,
COUNT(*) as request_count,
SUM(prompt_tokens) as total_prompt,
SUM(completion_tokens) as total_completion,
SUM(total_tokens) as total_tokens,
SUM(cost) as total_cost
FROM token_usage
GROUP BY model
ORDER BY total_cost DESC
LIMIT ?
''', (limit,))
return [dict(row) for row in cursor.fetchall()]สร้าง Client Wrapper พร้อม Request Logging
หลังจากมีระบบ track ค่าใช้จ่ายแล้ว ต่อไปเราต้องสร้าง client wrapper ที่จะ intercept request และ response ทุกครั้งเพื่อดึงข้อมูล token usage ออกมา ซึ่งวิธีนี้ทำให้เราไม่ต้องแก้ไขโค้ดที่ใช้ API อยู่แล้วมากมาย สามารถใช้งานได้ทันทีหลัง import เข้ามา สิ่งสำคัญคือต้อง handle error response ด้วย เพราะบางครั้ง API อาจ return error กลับมาโดยไม่มี usage info
import requests
from typing import Optional, Dict, Any, List
import hashlib
class HolySheepAPIClient:
"""
HolySheep AI API Client พร้อมระบบติดตามค่าใช้จ่าย
base_url: https://api.holysheep.ai/v1
"""
def __init__(self, api_key: str, tracker: Optional[AIAPICostTracker] = None):
self.api_key = api_key
self.base_url = 'https://api.holysheep.ai/v1'
self.tracker = tracker or AIAPICostTracker()
self.session_id = hashlib.md5(str(datetime.now()).encode()).hexdigest()[:8]
def _get_headers(self) -> Dict[str, str]:
return {
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
}
def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: Optional[int] = None,
**kwargs
) -> Dict[str, Any]:
"""ส่ง request ไปยัง chat completion API พร้อม track ค่าใช้จ่าย"""
start_time = time.time()
request_id = hashlib.md5(f"{self.session_id}{time.time()}".encode()).hexdigest()
payload = {
'model': model,
'messages': messages,
'temperature': temperature
}
if max_tokens:
payload['max_tokens'] = max_tokens
payload.update(kwargs)
try:
response = requests.post(
f'{self.base_url}/chat/completions',
headers=self._get_headers(),
json=payload,
timeout=30
)
response_time = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
usage = data.get('usage', {})
prompt_tokens = usage.get('prompt_tokens', 0)
completion_tokens = usage.get('completion_tokens', 0)
total_tokens = usage.get('total_tokens', 0)
token_usage = TokenUsage(
model=model,
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=total_tokens,
cost=self.tracker.calculate_cost(model, prompt_tokens, completion_tokens),
timestamp=datetime.now(),
request_id=request_id,
response_time_ms=response_time
)
self.tracker.log_usage(token_usage, self.session_id)
# ตรวจสอบงบประมาณหลัง request
self._check_budget_alert(model)
return data
else:
# Log error response แต่ยังคง raise exception
print(f"API Error: {response.status_code} - {response.text}")
response.raise_for_status()
except requests.exceptions.RequestException as e:
print(f"Request failed: {str(e)}")
raise
def embedding(self, model: str, input_text: str) -> Dict[str, Any]:
"""สร้าง embedding พร้อม track ค่าใช้จ่าย"""
start_time = time.time()
payload = {
'model': model,
'input': input_text
}
response = requests.post(
f'{self.base_url}/embeddings',
headers=self._get_headers(),
json=payload,
timeout=30
)
response_time = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
usage = data.get('usage', {})
# Embedding ใช้ prompt_tokens เป็น input_tokens
input_tokens = usage.get('prompt_tokens', 0)
total_tokens = usage.get('total_tokens', input_tokens)
# คำนวณค่า embedding (ปกติจะถูกกว่า text generation)
cost = (total_tokens / 1_000_000) * 0.10
token_usage = TokenUsage(
model=model,
prompt_tokens=input_tokens,
completion_tokens=0,
total_tokens=total_tokens,
cost=cost,
timestamp=datetime.now(),
response_time_ms=response_time
)
self.tracker.log_usage(token_usage, self.session_id)
return data
else:
response.raise_for_status()
def _check_budget_alert(self, model: str):
"""ตรวจสอบและส่งการแจ้งเตือนเมื่อใช้งานเกิน threshold"""
daily_spending = self.tracker.get_spending('daily')
# ส่งการแจ้งเตือนผ่าน Slack/Email/PagerDuty
# ตั้งค่า threshold ที่ 80% ของงบประมาณรายวัน
alert_threshold = 8.0 # $8
if daily_spending >= alert_threshold:
self._send_alert(
alert_type='daily_budget_warning',
message=f'ค่าใช้จ่ายรายวันถึง ${daily_spending:.2f} แล้ว'
)
def _send_alert(self, alert_type: str, message: str):
"""ส่งการแจ้งเตือนผ่านช่องทางต่างๆ"""
# Integration กับ Slack
# Integration กับ Email
# Integration กับ PagerDuty
print(f"[ALERT] {alert_type}: {message}")สร้าง Dashboard สำหรับแสดงผลแบบ Real-time
ตอนนี้เรามีข้อมูลค่าใช้จ่ายเก็บในฐานข้อมูลแล้ว ต่อไปมาสร้าง dashboard สำหรับแสดงผลกัน ซึ่งจะช่วยให้เราเห็นภาพรวมของการใช้งานได้ง่ายขึ้น ผมใช้ terminal output แบบ simple ก่อน แต่สามารถสร้าง web dashboard ด้วย Flask/FastAPI เพิ่มเติมได้ตามต้องการ
import matplotlib.pyplot as plt
from typing import List
import os
class CostDashboard:
"""
Dashboard แสดงผลค่าใช้จ่าย AI API แบบ real-time
"""
def __init__(self, tracker: AIAPICostTracker):
self.tracker = tracker
def print_summary(self):
"""พิมพ์สรุปค่าใช้จ่ายแบบ text"""
print("=" * 60)
print("🤖 AI API COST SUMMARY")
print("=" * 60)
# ค่าใช้จ่ายวันนี้
daily = self.tracker.get_spending('daily')
print(f"📅 วันนี้: ${daily:.4f}")
# ค่าใช้จ่ายสัปดาห์นี้
weekly = self.tracker.get_spending('weekly')
print(f"📆 สัปดาห์นี้: ${weekly:.4f}")
# ค่าใช้จ่ายเดือนนี้
monthly = self.tracker.get_spending('monthly')
print(f"🗓️ เดือนนี้: ${monthly:.4f}")
print("\n" + "-" * 60)
print("📊 การใช้งานแยกตามโมเดล:")
print("-" * 60)
usage = self.tracker.get_usage_by_model()
for item in usage:
print(f"\n🔹 {item['model']}")
print(f" Requests: {item['request_count']}")
print(f" Total Tokens: {item['total_tokens']:,}")
print(f" ค่าใช้จ่าย: ${item['total_cost']:.4f}")
# แสดงเปอร์เซ็นต์ของค่าใช้จ่ายทั้งหมด
if monthly > 0:
percent = (item['total_cost'] / monthly) * 100
print(f" สัดส่วน: {percent:.1f}%")
def get_hourly_spending(self, hours: int = 24) -> List[tuple]:
"""ดึงข้อมูลค่าใช้จ่ายรายชั่วโมง"""
with sqlite3.connect(self.tracker.db_path) as conn:
cursor = conn.cursor()
since = datetime.now() - timedelta(hours=hours)
cursor.execute('''
SELECT
strftime('%Y-%m-%d %H:00', timestamp) as hour,
SUM(cost) as total_cost,
SUM(total_tokens) as total_tokens
FROM token_usage
WHERE timestamp >= ?
GROUP BY hour
ORDER BY hour
''', (since.isoformat(),))
return cursor.fetchall()
def plot_spending_trend(self, hours: int = 168):
"""สร้างกราฟแนวโน้มค่าใช้จ่าย"""
data = self.get_hourly_spending(hours)
if not data:
print("ไม่มีข้อมูลสำหรับแสดงกราฟ")
return
hours_list = [row[0] for row in data]
costs_list = [row[1] for row in data]
plt.figure(figsize=(14, 6))
plt.plot(range(len(hours_list)), costs_list, marker='o', linewidth=2)
plt.title('AI API Spending Trend (Last 7 Days)')
plt.xlabel('Hours')
plt.ylabel('Cost ($)')
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('spending_trend.png', dpi=150)
print("📈 กราฟถูกบันทึกที่ spending_trend.png")
ตัวอย่างการใช้งาน
if __name__ == '__main__':
tracker = AIAPICostTracker()
dashboard = CostDashboard(tracker)
# สร้าง API client
client = HolySheepAPIClient(
api_key='YOUR_HOLYSHEEP_API_KEY',
tracker=tracker
)
# ตัวอย่างการใช้งาน
messages = [
{'role': 'system', 'content': 'You are a helpful assistant.'},
{'role': 'user', 'content': 'Explain quantum computing in simple terms.'}
]
# เรียกใช้ API
response = client.chat_completion(
model='deepseek-v3.2',
messages=messages,
temperature=0.7
)
# แสดง dashboard
dashboard.print_summary()ตั้งค่า Alert System ผ่าน Webhook
การแจ้งเตือนที่ดีต้องส่งไปถึงมือเราทันทีก่อนที่จะเกิดความเสียหาย ผมเลยสร้างระบบ alert ที่เชื่อมต่อกับ Slack, Discord หรือ email ได้โดยตรง ซึ่งสามารถกำหนดเงื่อนไขได้หลายแบบ เช่น แจ้งเตือนเมื่อค่าใช้จ่ายเกิน threshold, แจ้งเมื่อใช้ token เกินจำนวนที่กำหนด หรือแจ้งเมื่อโมเดลบางตัวถูกใช้งานผิดปกติ
import smtplib
import json
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from typing import Optional, List
class AlertSystem:
"""
ระบบแจ้งเตือนค่าใช้จ่าย AI API
รองรับ Slack, Discord, Email, LINE Notify
"""
def __init__(self):
self.channels = []
def add_slack_webhook(self, webhook_url: str, channel: str = '#ai-alerts'):
self.channels.append({
'type': 'slack',
'webhook_url': webhook_url,
'channel': channel
})
def add_discord_webhook(self, webhook_url: str):
self.channels.append({
'type': 'discord',
'webhook_url': webhook_url
})
def add_email_alert(
self,
smtp_server: str,
smtp_port: int,
sender_email: str,
sender_password: str,
recipient_emails: List[str]
):
self.channels.append({
'type': 'email',
'smtp_server': smtp_server,
'smtp_port': smtp_port,
'sender_email': sender_email,
'sender_password': sender_password,
'recipient_emails': recipient_emails
})
def send_alert(
self,
title: str,
message: str,
severity: str = 'warning',
metadata: Optional[Dict] = None
):
"""ส่งการแจ้งเตือนไปยังทุกช่องทางที่ตั้งค่าไว้"""
emoji_map = {
'info': 'ℹ️',
'warning': '⚠️',
'critical': '🚨',
'success': '✅'
}
emoji = emoji_map.get(severity, 'ℹ️')
formatted_message = f"{emoji} *{title}*\n\n{message}"
if metadata:
formatted_message += "\n\n📋 *Details:*"
for key, value in metadata.items():
formatted_message += f"\n• {key}: {value}"
for channel in self.channels:
try:
if channel['type'] == 'slack':
self._send_slack(channel, formatted_message)
elif channel['type'] == 'discord':
self._send_discord(channel, title, formatted_message, severity)
elif channel['type'] == 'email':
self._send_email(channel, title, formatted_message)
except Exception as e:
print(f"Failed to send alert via {channel['type']}: {e}")
def _send_slack(self, channel: dict, message: str):
"""ส่งการแจ้งเตือนไป Slack"""
payload = {
'channel': channel['channel'],
'text': message,
'username': 'AI Cost Monitor',
'icon_emoji': ':money_with_wings:'
}
requests.post(channel['webhook_url'], json=payload)
def _send_discord(self, channel: dict, title: str, message: str, severity: str):
"""ส่งการแจ้งเตือนไป Discord"""
color_map = {
'info': 3447003,
'warning': 16776960,
'critical': 15158332,
'success': 3066993
}
payload = {
'embeds': [{
'title': title,
'description': message,
'color': color_map.get(severity, 3447003)
}]
}
requests.post(channel['webhook_url'], json=payload)
def _send_email(self, channel: dict, title: str, message: str):
"""ส่งการแจ้งเตือนทาง Email"""
msg = MIMEMultipart()
msg['From'] = channel['sender_email']
msg['To'] = ', '.join(channel['recipient_emails'])
msg['Subject'] = f"[AI Cost Alert] {title}"
msg.attach(MIMEText(message, 'plain'))
with smtplib.SMTP(channel['smtp_server'], channel['smtp_port']) as server:
server.starttls()
server.login(channel['sender_email'], channel['sender_password'])
server.send_message(msg)
ตัวอย่างการใช้งาน Alert System
def setup_alerts() -> AlertSystem:
"""ตั้งค่าระบบแจ้งเตือน"""
alert_system = AlertSystem()
# เพิ่ม Slack webhook
# alert_system.add_slack_webhook('https://hooks.slack.com/services/xxx')
# เพิ่ม Discord webhook
# alert_system.add_discord_webhook('https://discord.com/api/webhooks/xxx')
# เพิ่ม Email
# alert_system.add_email_alert(
# smtp_server='smtp.gmail.com',
# smtp_port=587,
# sender_email='[email protected]',
# sender_password='your-app-password',
# recipient_emails=['[email protected]']
# )
return alert_system
class