ในฐานะวิศวกรที่ดูแลระบบ AI หลายตัวมาหลายปี ผมเคยเจอปัญหา token ล้นงบประมาณจนบิลดิบเป็นหมื่นดอลลาร์ในเดือนเดียว วันนี้ผมจะมาแชร์วิธีที่ผมใช้ HolySheep AI ในการจัดการ token และตั้งค่า budget alert อย่างมีประสิทธิภาพ พร้อมโค้ด production-ready ที่รันอยู่จริงบน production ของผมเอง
ทำไมต้องจัดการ Token อย่างเป็นระบบ
จากประสบการณ์ตรงที่ดูแลระบบ RAG และ AI agent หลายตัว ผมพบว่า 80% ของปัญหาค่าใช้จ่ายที่ไม่คาดคิดมาจาก 3 สาเหตุหลัก:
- Context explosion: Prompt ที่ยาวขึ้นเรื่อยๆ จากการ chain requests
- Retry loops: โค้ดที่ไม่มี exponential backoff ทำให้เกิด request ซ้ำๆ
- Lack of monitoring: ไม่มีระบบ alert เมื่อใช้งานเกิน threshold
สถาปัตยกรรม Token Management System
ระบบที่ผมออกแบบประกอบด้วย 4 ชั้นหลัก:
- Token Tracker: ติดตาม usage แบบ real-time
- Budget Controller: ควบคุมวงเงินตาม time window
- Alert Manager: ส่ง notification เมื่อถึง threshold
- Rate Limiter: จำกัด request rate ตาม plan
การตั้งค่า Token Tracker
เริ่มจากการสร้าง client wrapper ที่ track usage อัตโนมัติทุก request:
"""
HolySheep Token Management Client
Production-ready wrapper พร้อม built-in tracking และ budget control
"""
import time
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, Dict, Callable, Any
from collections import defaultdict
import threading
import httpx
logger = logging.getLogger(__name__)
@dataclass
class TokenUsage:
"""โครงสร้างข้อมูลสำหรับเก็บ token usage"""
model: str
input_tokens: int
output_tokens: int
cost_usd: float
latency_ms: float
timestamp: datetime = field(default_factory=datetime.now)
@dataclass
class BudgetConfig:
"""การตั้งค่างบประมาณ"""
daily_limit: float = 100.0 # ดอลลาร์ต่อวัน
monthly_limit: float = 1000.0 # ดอลลาร์ต่อเดือน
alert_threshold: float = 0.8 # แจ้งเตือนเมื่อใช้ไป 80%
emergency_threshold: float = 0.95 # หยุดเมื่อใช้ไป 95%
class HolySheepTokenManager:
"""
Token Manager สำหรับ HolySheep API
- ติดตาม usage แบบ real-time
- Budget control พร้อม hard/soft limits
- Automatic alert เมื่อเกิน threshold
- Rate limiting ในตัว
"""
BASE_URL = "https://api.holysheep.ai/v1"
# ราคาต่อ million tokens (USD) - อัปเดต 2026
PRICING = {
"gpt-4.1": {"input": 8.0, "output": 8.0},
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42},
}
def __init__(
self,
api_key: str,
budget_config: Optional[BudgetConfig] = None,
on_alert: Optional[Callable[[str, Dict], None]] = None
):
self.api_key = api_key
self.budget_config = budget_config or BudgetConfig()
self.on_alert = on_alert or self._default_alert_handler
# In-memory storage สำหรับ usage tracking
self._usage_log: list[TokenUsage] = []
self._daily_spent: Dict[str, float] = defaultdict(float)
self._monthly_spent: Dict[str, float] = defaultdict(float)
self._lock = threading.Lock()
# Rate limiting
self._request_times: list[float] = []
self._rate_limit = 60 # requests per minute
# Client สำหรับ HTTP requests
self._client = httpx.Client(
timeout=60.0,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
)
def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""คำนวณค่าใช้จ่ายจากจำนวน tokens"""
pricing = self.PRICING.get(model, {"input": 0, "output": 0})
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return round(input_cost + output_cost, 6) # แม่นยำถึง 6 หลักทศนิยม
def _check_budget(self, estimated_cost: float) -> bool:
"""ตรวจสอบว่าอยู่ในงบประมาณหรือไม่"""
today = datetime.now().date()
daily_key = str(today)
daily_total = self._daily_spent[daily_key]
monthly_key = f"{today.year}-{today.month}"
monthly_total = self._monthly_spent[monthly_key]
# Soft limit check - แจ้งเตือน
if daily_total >= self.budget_config.daily_limit * self.budget_config.alert_threshold:
self.on_alert("DAILY_SOFT_LIMIT", {
"spent": daily_total,
"limit": self.budget_config.daily_limit,
"remaining": self.budget_config.daily_limit - daily_total
})
# Hard limit check - ปฏิเสธ request
if daily_total >= self.budget_config.daily_limit * self.budget_config.emergency_threshold:
logger.warning(f"Daily emergency threshold reached: ${daily_total:.2f}")
return False
if monthly_total >= self.budget_config.monthly_limit * self.budget_config.emergency_threshold:
logger.warning(f"Monthly emergency threshold reached: ${monthly_total:.2f}")
return False
return True
def _default_alert_handler(self, alert_type: str, data: Dict):
"""Handler เริ่มต้นสำหรับ alert - พิมพ์ log"""
logger.warning(f"ALERT [{alert_type}]: {data}")
def chat_completion(
self,
model: str,
messages: list,
max_tokens: int = 1024,
temperature: float = 0.7,
**kwargs
) -> Dict[str, Any]:
"""
ส่ง chat completion request พร้อม tracking
Args:
model: ชื่อ model (เช่น 'deepseek-v3.2', 'gpt-4.1')
messages: list of message dicts
max_tokens: maximum output tokens
temperature: sampling temperature
Returns:
Response dict พร้อม usage information
"""
start_time = time.perf_counter()
# Rate limit check
self._check_rate_limit()
# Budget check
# Estimate cost ล่วงหน้า (ใช้ approx 4 chars per token)
estimated_input_tokens = sum(len(str(m)) // 4 for m in messages)
estimated_cost = self._calculate_cost(
model, estimated_input_tokens, max_tokens
)
if not self._check_budget(estimated_cost):
raise BudgetExceededError(
f"Budget limit reached. Daily limit: ${self.budget_config.daily_limit}"
)
# Prepare request
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature,
**kwargs
}
try:
response = self._client.post(
f"{self.BASE_URL}/chat/completions",
json=payload
)
response.raise_for_status()
result = response.json()
# Calculate actual cost
usage = result.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
actual_cost = self._calculate_cost(model, input_tokens, output_tokens)
latency_ms = (time.perf_counter() - start_time) * 1000
# Log usage
token_usage = TokenUsage(
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost_usd=actual_cost,
latency_ms=round(latency_ms, 2) # แม่นยำถึง 2 หลักทศนิยม (มิลลิวินาที)
)
with self._lock:
self._usage_log.append(token_usage)
today = str(datetime.now().date())
monthly_key = f"{datetime.now().year}-{datetime.now().month}"
self._daily_spent[today] += actual_cost
self._monthly_spent[monthly_key] += actual_cost
# Add usage info to response
result["_token_usage"] = token_usage
return result
except httpx.HTTPStatusError as e:
logger.error(f"HTTP Error: {e.response.status_code} - {e.response.text}")
raise
def _check_rate_limit(self):
"""ตรวจสอบ rate limit"""
now = time.time()
with self._lock:
# Remove requests older than 1 minute
self._request_times = [t for t in self._request_times if now - t < 60]
if len(self._request_times) >= self._rate_limit:
sleep_time = 60 - (now - self._request_times[0])
if sleep_time > 0:
logger.info(f"Rate limit reached, sleeping {sleep_time:.2f}s")
time.sleep(sleep_time)
self._request_times.append(now)
def get_usage_report(self, days: int = 7) -> Dict[str, Any]:
"""สร้าง report สรุปการใช้งาน"""
cutoff = datetime.now() - timedelta(days=days)
with self._lock:
recent_usage = [u for u in self._usage_log if u.timestamp > cutoff]
total_cost = sum(u.cost_usd for u in recent_usage)
total_input = sum(u.input_tokens for u in recent_usage)
total_output = sum(u.output_tokens for u in recent_usage)
avg_latency = sum(u.latency_ms for u in recent_usage) / len(recent_usage) if recent_usage else 0
model_breakdown = defaultdict(lambda: {"requests": 0, "cost": 0.0, "tokens": 0})
for u in recent_usage:
model_breakdown[u.model]["requests"] += 1
model_breakdown[u.model]["cost"] += u.cost_usd
model_breakdown[u.model]["tokens"] += u.input_tokens + u.output_tokens
return {
"period_days": days,
"total_requests": len(recent_usage),
"total_cost_usd": round(total_cost, 2),
"total_input_tokens": total_input,
"total_output_tokens": total_output,
"avg_latency_ms": round(avg_latency, 2),
"model_breakdown": dict(model_breakdown),
"daily_budget_remaining": round(
self.budget_config.daily_limit - self._daily_spent[str(datetime.now().date())],
2
),
"monthly_budget_remaining": round(
self.budget_config.monthly_limit - self._monthly_spent[
f"{datetime.now().year}-{datetime.now().month}"
],
2
)
}
class BudgetExceededError(Exception):
"""Exception ที่โยนเมื่อ budget เกิน limit"""
passการตั้งค่า Budget Alert System
ต่อไปคือระบบ alert ที่ทำงานแบบ real-time ผ่าน webhook และ email:
"""
Budget Alert System สำหรับ HolySheep
ส่ง notification ผ่านหลายช่องทางเมื่อถึง threshold
"""
import json
import asyncio
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from dataclasses import dataclass
from typing import Optional, Protocol
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
Alert thresholds แบบ granular
ALERT_LEVELS = {
"INFO": 0.5, # 50% - แจ้งเตือนเบาๆ
"WARNING": 0.75, # 75% - เตือนว่าใกล้ถึง limit
"CRITICAL": 0.90, # 90% - วิกฤติ ต้องดำเนินการ
"EMERGENCY": 0.98 # 98% - ใกล้หมดงบ
}
@dataclass
class AlertMessage:
level: str
title: str
body: str
data: dict
timestamp: datetime
class AlertChannel(Protocol):
"""Protocol สำหรับ alert channels"""
async def send(self, alert: AlertMessage) -> bool: ...
class WebhookAlertChannel:
"""ส่ง alert ไปยัง webhook (Slack, Discord, LINE, etc.)"""
def __init__(self, webhook_url: str, channel_name: str = "HolySheep Alerts"):
self.webhook_url = webhook_url
self.channel_name = channel_name
async def send(self, alert: AlertMessage) -> bool:
import httpx
# Format message ตาม webhook type
payload = {
"username": self.channel_name,
"embeds": [{
"title": f"🚨 {alert.level}: {alert.title}",
"description": alert.body,
"color": self._get_color(alert.level),
"fields": [
{"name": k, "value": str(v), "inline": True}
for k, v in alert.data.items()
],
"timestamp": alert.timestamp.isoformat()
}]
}
try:
async with httpx.AsyncClient() as client:
response = await client.post(
self.webhook_url,
json=payload,
timeout=10.0
)
return response.status_code == 200
except Exception as e:
logger.error(f"Webhook alert failed: {e}")
return False
def _get_color(self, level: str) -> int:
colors = {
"INFO": 0x3498db, # ฟ้า
"WARNING": 0xf39c12, # ส้ม
"CRITICAL": 0xe74c3c, # แดง
"EMERGENCY": 0x9b59b6 # ม่วง
}
return colors.get(level, 0x95a5a6)
class EmailAlertChannel:
"""ส่ง alert ทาง email"""
def __init__(
self,
smtp_host: str,
smtp_port: int,
smtp_user: str,
smtp_password: str,
from_email: str,
to_emails: list[str]
):
self.smtp_host = smtp_host
self.smtp_port = smtp_port
self.smtp_user = smtp_user
self.smtp_password = smtp_password
self.from_email = from_email
self.to_emails = to_emails
async def send(self, alert: AlertMessage) -> bool:
msg = MIMEMultipart("alternative")
msg["Subject"] = f"[{alert.level}] HolySheep Budget Alert: {alert.title}"
msg["From"] = self.from_email
msg["To"] = ", ".join(self.to_emails)
# Plain text version
text_body = f"""
HolySheep Budget Alert
=======================
Level: {alert.level}
Title: {alert.title}
{alert.body}
Details:
{json.dumps(alert.data, indent=2)}
Time: {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S UTC')}
"""
# HTML version
html_body = f"""
<html>
<body style="font-family: Arial, sans-serif;">
<h2 style="color: {'#e74c3c' if alert.level in ['CRITICAL', 'EMERGENCY'] else '#f39c12'};">
🚨 {alert.level}: {alert.title}
</h2>
<p>{alert.body}</p>
<table style="border-collapse: collapse; width: 100%;">
"""
for key, value in alert.data.items():
html_body += f"""
<tr>
<td style="padding: 8px; border: 1px solid #ddd; font-weight: bold;">{key}</td>
<td style="padding: 8px; border: 1px solid #ddd;">{value}</td>
</tr>
"""
html_body += f"""
</table>
<p style="color: #666; font-size: 12px;">
Time: {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S UTC')}
</p>
</body>
</html>
"""
msg.attach(MIMEText(text_body, "plain"))
msg.attach(MIMEText(html_body, "html"))
try:
with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
server.starttls()
server.login(self.smtp_user, self.smtp_password)
server.send_message(msg)
return True
except Exception as e:
logger.error(f"Email alert failed: {e}")
return False
class BudgetAlertManager:
"""
จัดการ budget alerts สำหรับ HolySheep
- ตรวจสอบ threshold หลายระดับ
- ส่งไปยังหลาย channels
- Cooldown เพื่อไม่ให้ส่งซ้ำ
"""
def __init__(self):
self.channels: list[AlertChannel] = []
self._last_alerts: dict[str, datetime] = {}
self._cooldown_seconds = 300 # 5 นาที cooldown ระหว่าง alert ประเภทเดียวกัน
def add_channel(self, channel: AlertChannel):
self.channels.append(channel)
async def check_and_alert(
self,
current_spent: float,
daily_limit: float,
monthly_limit: float,
daily_usage: dict
):
"""ตรวจสอบและส่ง alert ถ้าจำเป็น"""
daily_percentage = current_spent / daily_limit
monthly_percentage = daily_usage.get("monthly_spent", 0) / monthly_limit
# ตรวจสอบแต่ละ level
for level_name, threshold in ALERT_LEVELS.items():
daily_key = f"daily_{level_name}"
# ตรวจสอบ cooldown
if daily_key in self._last_alerts:
elapsed = (datetime.now() - self._last_alerts[daily_key]).total_seconds()
if elapsed < self._cooldown_seconds:
continue
if daily_percentage >= threshold or monthly_percentage >= threshold:
alert = AlertMessage(
level=level_name,
title=f"Budget {level_name}",
body=self._generate_alert_body(
level_name, current_spent, daily_limit,
daily_percentage, monthly_percentage
),
data={
"Spent (USD)": f"${current_spent:.2f}",
"Daily Limit": f"${daily_limit:.2f}",
"Daily %": f"{daily_percentage * 100:.1f}%",
"Monthly %": f"{monthly_percentage * 100:.1f}%",
"Requests Today": daily_usage.get("request_count", 0),
"Avg Cost/Request": f"${daily_usage.get('avg_cost', 0):.4f}"
},
timestamp=datetime.now()
)
# ส่งไปยังทุก channel
results = await asyncio.gather(
*[channel.send(alert) for channel in self.channels],
return_exceptions=True
)
if any(r for r in results):
self._last_alerts[daily_key] = datetime.now()
logger.info(f"Sent {level_name} alert to {sum(1 for r in results if r)} channels")
def _generate_alert_body(
self,
level: str,
spent: float,
limit: float,
daily_pct: float,
monthly_pct: float
) -> str:
messages = {
"INFO": f"คุณใช้งานไป {daily_pct*100:.0f}% ของงบประมาณรายวัน ยังเหลือเวลาในการวางแผน",
"WARNING": f"⚠️ ใช้งานไป {daily_pct*100:.0f}% แล้ว ควรเริ่มจำกัดการใช้งาน",
"CRITICAL": f"🚨 ถึง {daily_pct*100:.0f}% ของงบ ต้องดำเนินการทันที!",
"EMERGENCY": f"🚨🚨 งบประมาณเกือบหมดแล้ว ({daily_pct*100:.0f}%) ระบบจะหยุดทำงานเร็วๆ นี้!"
}
return messages.get(level, "")
ตัวอย่างการใช้งาน
async def main():
# สร้าง alert manager
alert_manager = BudgetAlertManager()
# เพิ่ม webhook channel (Slack/Discord)
alert_manager.add_channel(
WebhookAlertChannel(
webhook_url="https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
channel_name="HolySheep Budget"
)
)
# เพิ่ม email channel
alert_manager.add_channel(
EmailAlertChannel(
smtp_host="smtp.gmail.com",
smtp_port=587,
smtp_user="[email protected]",
smtp_password="your-app-password",
from_email="[email protected]",
to_emails=["[email protected]", "[email protected]"]
)
)
# ในโค้ดหลัก ทำการ check ทุก 5 นาที
while True:
usage = manager.get_usage_report(days=1)
daily_spent = usage.get("daily_budget_remaining", 0)
await alert_manager.check_and_alert(
current_spent=daily_spent,
daily_limit=100.0,
monthly_limit=1000.0,
daily_usage={
"monthly_spent": usage.get("total_cost_usd", 0),
"request_count": usage.get("total_requests", 0),
"avg_cost": usage.get("total_cost_usd", 0) / max(usage.get("total_requests", 1), 1)
}
)
await asyncio.sleep(300) # Check ทุก 5 นาที
if __name__ == "__main__":
asyncio.run(main())การตั้งค่า Production Dashboard
ผมสร้าง dashboard แบบง่ายๆ ด้วย FastAPI สำหรับ monitor usage แบบ real-time:
"""
HolySheep Budget Dashboard - FastAPI Backend
Dashboard สำหรับ monitor token usage และ budget แบบ real-time
"""
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from datetime import datetime, timedelta
from typing import Optional
import uvicorn
app = FastAPI(title="HolySheep Budget Dashboard", version="1.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
Global manager instance
manager: Optional[HolySheepTokenManager] = None
alert_manager: Optional[BudgetAlertManager] = None
def init_managers(api_key: str, daily_limit: float = 100.0, monthly_limit: float = 1000.0):
global manager, alert_manager
budget_config = BudgetConfig(
daily_limit=daily_limit,
monthly_limit=monthly_limit,
alert_threshold=0.75,
emergency_threshold=0.95
)
async def custom_alert(alert_type: str, data: dict):
print(f"🚨 ALERT [{alert_type}]: {data}")
manager = HolySheepTokenManager(
api_key=api_key,
budget_config=budget_config,
on_alert=custom_alert
)
alert_manager = BudgetAlertManager()
# เพิ่ม channels ตามต้องการ
API Models
class ChatRequest(BaseModel):
model: str
messages: list
max_tokens: int = 1024
temperature: float = 0.7
class BudgetUpdate(BaseModel):
daily_limit: Optional[float] = None
monthly_limit: Optional[float] = None
alert_threshold: Optional[float] = None
API Endpoints
@app.get("/")
async def root():
return {"message": "HolySheep Budget Dashboard API", "version": "1.0.0"}
@app.get("/api/usage/current")
async def get_current_usage():
"""ดึงข้อมูล usage ปัจจุบัน"""
if not manager:
raise HTTPException(status_code=500, detail="Manager not initialized")
return manager.get_usage_report(days=1)
@app.get("/api/usage/history")
async def get_usage_history(days: int = 7):
"""ดึงข้อมูล usage ย้อนหลัง"""
if not manager:
raise HTTPException(status_code=500, detail="Manager not initialized")
return manager.get_usage_report(days=days)