การดูและควบคุมต้นทุน API เป็นหัวใจสำคัญของการพัฒนาระบบที่ยั่งยืน โดยเฉพาะเมื่อเราพูดถึง LLM API ที่มีราคาสูงและมีแนวโน้มเพิ่มขึ้นทุกวัน บทความนี้จะพาคุณสร้างระบบ API Cost Monitoring และ Alerting ที่ production-ready ตั้งแต่การออกแบบสถาปัตยกรรม การติดตั้ง monitoring pipeline ไปจนถึงการตั้งค่า alert rules ที่ฉลาดและแม่นยำ
ทำไมต้องมี API Cost Monitoring?
ในระบบ Production ที่ใช้ LLM API จาก HolySheep AI ซึ่งมีอัตราที่ประหยัดมากกว่า 85% เมื่อเทียบกับผู้ให้บริการรายอื่น (¥1 = $1 สำหรับ GPT-4.1, Claude Sonnet 4.5 และโมเดลอื่นๆ) การมีระบบ monitoring ที่ดีจะช่วยให้เรา:
- ป้องกันการบวมต้นทุน — ระบบจะแจ้งเตือนก่อนที่ค่าใช้จ่ายจะพุ่งสูงเกินควบคุม
- ตรวจจับ abnormal usage — รู้ทันทีเมื่อมี request ผิดปกติหรือ potential abuse
- เพิ่มประสิทธิภาพการใช้งาน — วิเคราะห์ patterns เพื่อหาจุดที่ต้อง optimize
- Budget forecasting — ประมาณการค่าใช้จ่ายล่วงหน้าได้แม่นยำขึ้น
สถาปัตยกรรมระบบ Monitoring
ระบบ monitoring ที่ดีต้องมีองค์ประกอบหลัก 4 ส่วน ได้แก่ Data Collection Layer สำหรับการเก็บ request metadata, Time-Series Database สำหรับจัดเก็บข้อมูลอนุกรมเวลา, Alert Engine สำหรับประมวลผลเงื่อนไขการแจ้งเตือน และ Notification Channel สำหรับส่ง alert ไปยังช่องทางที่ต้องการ
High-Level Architecture
+------------------+ +-------------------+ +------------------+
| Your App/Service | --> | HolySheep API | --> | Monitoring Layer|
| (with proxy) | | api.holysheep.ai | | |
+------------------+ +-------------------+ +------------------+
|
+-------------------+ +-------v--------+
| Prometheus/ | <-- | Alert Manager |
| InfluxDB | | |
+-------------------+ +-------+---------+
|
+-------------------+ |
| Slack/Email/ | <------+
| PagerDuty |
+-------------------+การสร้าง API Proxy พร้อม Cost Tracking
วิธีที่มีประสิทธิภาพที่สุดในการ monitor API cost คือการสร้าง proxy layer ที่ครอบ request ทั้งหมด โดย Python เป็นตัวเลือกที่ยอดเยี่ยมเนื่องจาก ecosystem ที่ครบครันและความเร็วในการพัฒนา ตัวอย่างต่อไปนี้เป็น production-ready proxy ที่รองรับ concurrent requests ได้อย่างมีประสิทธิภาพ
"""
HolySheep AI Cost Monitoring Proxy
Production-grade API proxy with real-time cost tracking
"""
import asyncio
import aiohttp
import time
import json
from dataclasses import dataclass, asdict
from typing import Optional, Dict, List
from datetime import datetime, timedelta
from collections import defaultdict
from threading import Lock
import logging
from logging.handlers import RotatingFileHandler
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
RotatingFileHandler('/var/log/api_proxy.log', maxBytes=10_000_000, backupCount=5),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
HolySheep AI Configuration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
MODEL_PRICING = {
# ราคาต่อ 1M tokens (2026)
"gpt-4.1": {"input": 8.0, "output": 8.0, "currency": "USD"},
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0, "currency": "USD"},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50, "currency": "USD"},
"deepseek-v3.2": {"input": 0.42, "output": 0.42, "currency": "USD"},
}
@dataclass
class TokenUsage:
"""โครงสร้างข้อมูลการใช้งาน token"""
timestamp: float
model: str
prompt_tokens: int
completion_tokens: int
total_tokens: int
cost_usd: float
request_id: str
user_id: Optional[str] = None
endpoint: str = "/chat/completions"
@dataclass
class AlertThreshold:
"""กำหนดค่า threshold สำหรับ alert"""
name: str
metric: str # 'cost_per_minute', 'cost_per_hour', 'request_count'
threshold: float
window_seconds: int
operator: str # 'gt', 'lt', 'eq'
severity: str # 'info', 'warning', 'critical'
class CostTracker:
"""ระบบติดตามต้นทุน API แบบ real-time"""
def __init__(self):
self._usage_buffer: List[TokenUsage] = []
self._daily_cost: Dict[str, float] = defaultdict(float)
self._hourly_cost: Dict[str, List[float]] = defaultdict(lambda: [0.0] * 60)
self._lock = Lock()
self._alerts: List[Dict] = []
# Default thresholds
self._thresholds = [
AlertThreshold("high_cost_per_min", "cost_per_minute", 10.0, 60, "gt", "warning"),
AlertThreshold("critical_cost_burst", "cost_per_minute", 50.0, 60, "gt", "critical"),
AlertThreshold("hourly_budget_warning", "cost_per_hour", 100.0, 3600, "gt", "warning"),
AlertThreshold("hourly_budget_critical", "cost_per_hour", 500.0, 3600, "gt", "critical"),
]
def record_usage(self, usage: TokenUsage):
"""บันทึกการใช้งาน API"""
with self._lock:
self._usage_buffer.append(usage)
self._daily_cost[usage.model] += usage.cost_usd
# Update hourly cost bucket
minute = int(usage.timestamp / 60) % 60
self._hourly_cost[usage.model][minute] += usage.cost_usd
# Check alerts
self._check_alerts(usage)
# Cleanup old data (keep last hour)
cutoff = time.time() - 3600
self._usage_buffer = [u for u in self._usage_buffer if u.timestamp > cutoff]
def calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> float:
"""คำนวณต้นทุนจากจำนวน tokens"""
pricing = MODEL_PRICING.get(model, MODEL_PRICING["deepseek-v3.2"])
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 get_cost_per_minute(self, model: str) -> float:
"""ดึงค่าใช้จ่ายต่อนาที"""
with self._lock:
minute = int(time.time() / 60) % 60
return self._hourly_cost.get(model, [0] * 60)[minute]
def get_cost_per_hour(self, model: str) -> float:
"""ดึงค่าใช้จ่ายต่อชั่วโมง"""
with self._lock:
return sum(self._hourly_cost.get(model, [0] * 60))
def get_daily_cost(self, model: str) -> float:
"""ดึงค่าใช้จ่ายวันนี้"""
with self._lock:
return self._daily_cost.get(model, 0.0)
def get_all_metrics(self) -> Dict:
"""ดึง metrics ทั้งหมด"""
with self._lock:
return {
"daily_costs": dict(self._daily_cost),
"total_daily_cost": sum(self._daily_cost.values()),
"request_count": len(self._usage_buffer),
"timestamp": datetime.utcnow().isoformat()
}
def _check_alerts(self, usage: TokenUsage):
"""ตรวจสอบเงื่อนไข alert"""
for threshold in self._thresholds:
current_value = self._get_metric_value(threshold.metric, usage.model)
triggered = False
if threshold.operator == "gt":
triggered = current_value > threshold.threshold
elif threshold.operator == "lt":
triggered = current_value < threshold.threshold
if triggered:
alert = {
"timestamp": datetime.utcnow().isoformat(),
"name": threshold.name,
"severity": threshold.severity,
"metric": threshold.metric,
"value": current_value,
"threshold": threshold.threshold,
"model": usage.model
}
self._alerts.append(alert)
logger.warning(f"ALERT TRIGGERED: {alert}")
def _get_metric_value(self, metric: str, model: str) -> float:
if metric == "cost_per_minute":
return self.get_cost_per_minute(model)
elif metric == "cost_per_hour":
return self.get_cost_per_hour(model)
return 0.0
def get_pending_alerts(self) -> List[Dict]:
return self._alerts.copy()
def clear_alerts(self):
with self._lock:
self._alerts.clear()
Singleton instance
cost_tracker = CostTracker()
class HolySheepProxy:
"""Proxy สำหรับ HolySheep AI API พร้อม cost tracking"""
def __init__(self, api_key: