การใช้งาน AI API ในปี 2026 เติบโตอย่างก้าวกระโดด แต่ต้นทุนที่พุ่งสูงขึ้นก็เป็นความท้าทายสำคับสำหรับทีมพัฒนา บทความนี้จะสอนวิธีสร้างระบบ API Cost Governance ที่ช่วยควบคุมค่าใช้จ่าย แยกวิเคราะห์ และแจ้งเตือนอัตโนมัติ เหมาะสำหรับองค์กรที่ใช้ AI หลายโมเดลพร้อมกัน
ภาพรวมต้นทุน AI API ปี 2026
ก่อนจะลงมือสร้างระบบจัดการต้นทุน เรามาดูข้อมูลราคาที่ตรวจสอบแล้วของโมเดลยอดนิยมในปี 2026 กันก่อน
ราคา Output ต่อ Million Tokens
| โมเดล | ราคา/MTok | ต้นทุน 10M tokens/เดือน | ประเภท |
|---|---|---|---|
| GPT-4.1 | $8.00 | $80.00 | Premium |
| Claude Sonnet 4.5 | $15.00 | $150.00 | Premium |
| Gemini 2.5 Flash | $2.50 | $25.00 | Mid-Range |
| DeepSeek V3.2 | $0.42 | $4.20 | Budget |
สรุป: หากใช้งาน 10 ล้าน tokens ต่อเดือน การเลือก DeepSeek V3.2 แทน Claude Sonnet 4.5 จะประหยัดได้ถึง $145.80 หรือคิดเป็น 97%
ทำไมต้องมีระบบ Cost Governance
- ควบคุมงบประมาณ: ป้องกันการบวมค่าใช้จ่ายจากการเรียก API มากเกินไป
- วิเคราะห์พฤติกรรม: รู้ว่าแผนกไหนใช้โมเดลอะไร เท่าไหร่
- เพิ่มประสิทธิภาพ: ย้ายงานที่เหมาะสมไปใช้โมเดลราคาถูกกว่า
- แจ้งเตือนเร็ว: ตรวจจับความผิดปกติก่อนที่บิลจะพุ่งแรง
การแยกค่าใช้จ่ายตามผู้เรียก/โมเดล/ช่วงเวลา
ระบบ Cost Governance ที่ดีต้องสามารถ track ค่าใช้จ่ายได้ละเอียด มาดูวิธีสร้างกัน
โครงสร้างข้อมูลสำหรับเก็บ Usage Logs
"""
API Cost Tracking System
เก็บข้อมูลการใช้งาน API แยกตาม caller, model, timestamp
"""
import sqlite3
from datetime import datetime, timedelta
from typing import Optional
import hashlib
class CostTracker:
def __init__(self, db_path: str = "cost_tracking.db"):
self.conn = sqlite3.connect(db_path)
self._init_database()
def _init_database(self):
"""สร้างตารางสำหรับเก็บ cost logs"""
cursor = self.conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS api_usage (
id INTEGER PRIMARY KEY AUTOINCREMENT,
caller_id TEXT NOT NULL,
model TEXT NOT NULL,
input_tokens INTEGER,
output_tokens INTEGER,
cost_usd REAL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
request_id TEXT UNIQUE,
metadata TEXT
)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_caller_time
ON api_usage(caller_id, timestamp)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_model_time
ON api_usage(model, timestamp)
""")
self.conn.commit()
def log_usage(
self,
caller_id: str,
model: str,
input_tokens: int,
output_tokens: int,
cost_usd: float,
request_id: Optional[str] = None,
metadata: Optional[dict] = None
):
"""บันทึกการใช้งาน API"""
if request_id is None:
request_id = hashlib.md5(
f"{caller_id}{model}{datetime.now()}".encode()
).hexdigest()
cursor = self.conn.cursor()
cursor.execute("""
INSERT INTO api_usage
(caller_id, model, input_tokens, output_tokens, cost_usd,
request_id, metadata)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
caller_id, model, input_tokens, output_tokens,
cost_usd, request_id, str(metadata) if metadata else None
))
self.conn.commit()
def get_cost_by_caller(
self,
start_date: datetime,
end_date: datetime,
caller_id: Optional[str] = None
) -> list:
"""ดึงค่าใช้จ่ายตามผู้เรียก"""
cursor = self.conn.cursor()
query = """
SELECT
caller_id,
model,
SUM(input_tokens) as total_input,
SUM(output_tokens) as total_output,
SUM(cost_usd) as total_cost,
COUNT(*) as request_count
FROM api_usage
WHERE timestamp BETWEEN ? AND ?
"""
params = [start_date, end_date]
if caller_id:
query += " AND caller_id = ?"
params.append(caller_id)
query += " GROUP BY caller_id, model ORDER BY total_cost DESC"
cursor.execute(query, params)
return cursor.fetchall()
def get_cost_by_model(
self,
start_date: datetime,
end_date: datetime
) -> list:
"""ดึงค่าใช้จ่ายตามโมเดล"""
cursor = self.conn.cursor()
cursor.execute("""
SELECT
model,
SUM(input_tokens) as total_input,
SUM(output_tokens) as total_output,
SUM(cost_usd) as total_cost,
COUNT(*) as request_count,
AVG(cost_usd) as avg_cost_per_request
FROM api_usage
WHERE timestamp BETWEEN ? AND ?
GROUP BY model
ORDER BY total_cost DESC
""", [start_date, end_date])
return cursor.fetchall()
def get_cost_by_hour(self, date: datetime) -> list:
"""ดึงค่าใช้จ่ายตามชั่วโมง"""
cursor = self.conn.cursor()
cursor.execute("""
SELECT
strftime('%H', timestamp) as hour,
SUM(cost_usd) as total_cost,
COUNT(*) as request_count
FROM api_usage
WHERE date(timestamp) = date(?)
GROUP BY hour
ORDER BY hour
""", [date])
return cursor.fetchall()
def close(self):
self.conn.close()
ตัวอย่างการใช้งาน
if __name__ == "__main__":
tracker = CostTracker()
# บันทึกการใช้งานตัวอย่าง
tracker.log_usage(
caller_id="backend-service-1",
model="gpt-4.1",
input_tokens=1500,
output_tokens=500,
cost_usd=0.016, # (1500+500)/1M * $8
metadata={"endpoint": "/chat/completions"}
)
# ดึงรายงาน
end = datetime.now()
start = end - timedelta(days=30)
print("=== ค่าใช้จ่ายตามผู้เรียก ===")
for row in tracker.get_cost_by_caller(start, end):
print(f"Caller: {row[0]}, Model: {row[1]}, Cost: ${row[4]:.2f}")
print("\n=== ค่าใช้จ่ายตามโมเดล ===")
for row in tracker.get_cost_by_model(start, end):
print(f"Model: {row[0]}, Cost: ${row[3]:.2f}, Requests: {row[4]}")
tracker.close()
การคำนวณต้นทุนอัตโนมัติ
"""
ตารางอัตราค่าบริการสำหรับคำนวณต้นทุน
อัปเดต: พฤษภาคม 2026
"""
หน่วย: USD ต่อ Million Tokens (Output)
MODEL_PRICING = {
"gpt-4.1": {
"input_per_mtok": 2.00,
"output_per_mtok": 8.00,
"provider": "OpenAI Compatible"
},
"claude-sonnet-4.5": {
"input_per_mtok": 3.00,
"output_per_mtok": 15.00,
"provider": "Anthropic Compatible"
},
"gemini-2.5-flash": {
"input_per_mtok": 0.30,
"output_per_mtok": 2.50,
"provider": "Google Compatible"
},
"deepseek-v3.2": {
"input_per_mtok": 0.10,
"output_per_mtok": 0.42,
"provider": "DeepSeek Compatible"
}
}
def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
"""คำนวณค่าใช้จ่ายจากจำนวน tokens"""
if model not in MODEL_PRICING:
raise ValueError(f"Unknown model: {model}")
pricing = MODEL_PRICING[model]
input_cost = (input_tokens / 1_000_000) * pricing["input_per_mtok"]
output_cost = (output_tokens / 1_000_000) * pricing["output_per_mtok"]
return round(input_cost + output_cost, 6)
def calculate_monthly_budget(
daily_request_count: int,
avg_input_tokens: int,
avg_output_tokens: int,
model: str
) -> dict:
"""คำนวณงบประมาณรายเดือนโดยประมาณ"""
daily_cost = daily_request_count * calculate_cost(
model, avg_input_tokens, avg_output_tokens
)
monthly_cost = daily_cost * 30
return {
"model": model,
"daily_requests": daily_request_count,
"avg_input": avg_input_tokens,
"avg_output": avg_output_tokens,
"daily_cost_usd": round(daily_cost, 2),
"monthly_cost_usd": round(monthly_cost, 2),
"yearly_cost_usd": round(monthly_cost * 12, 2)
}
ตัวอย่างการคำนวณ
if __name__ == "__main__":
# สมมติ: วันละ 1000 คำขอ, input เฉลี่ย 500 tokens, output เฉลี่ย 200 tokens
print("=== ค่าใช้จ่ายรายเดือนโดยประมาณ ===\n")
for model in MODEL_PRICING:
budget = calculate_monthly_budget(
daily_request_count=1000,
avg_input_tokens=500,
avg_output_tokens=200,
model=model
)
print(f"โมเดล: {model}")
print(f" ค่าใช้จ่ายรายวัน: ${budget['daily_cost_usd']}")
print(f" ค่าใช้จ่ายรายเดือน: ${budget['monthly_cost_usd']}")
print(f" ค่าใช้จ่ายรายปี: ${budget['yearly_cost_usd']}")
print()
# คำนวณ savings จากการใช้ DeepSeek แทน Claude
claude_monthly = calculate_monthly_budget(1000, 500, 200, "claude-sonnet-4.5")
deepseek_monthly = calculate_monthly_budget(1000, 500, 200, "deepseek-v3.2")
savings = claude_monthly['monthly_cost_usd'] - deepseek_monthly['monthly_cost_usd']
savings_pct = (savings / claude_monthly['monthly_cost_usd']) * 100
print(f"💰 หากเปลี่ยนจาก Claude → DeepSeek:")
print(f" ประหยัดได้: ${savings:.2f}/เดือน ({savings_pct:.1f}%)")
การตั้งค่าระบบแจ้งเตือนงบประมาณ
การตั้ง Budget Alert ที่เหมาะสมช่วยป้องกันค่าใช้จ่ายที่ไม่คาดคิด เราจะสร้างระบบที่แจ้งเตือนเมื่อใช้งานเกิน threshold ที่กำหนด
"""
ระบบ Budget Alert สำหรับ API Cost Management
แจ้งเตือนเมื่อใช้งานเกินงบประมาณที่กำหนด
"""
import sqlite3
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Optional, List, Callable
import json
@dataclass
class BudgetThreshold:
"""กำหนดค่า threshold สำหรับการแจ้งเตือน"""
caller_id: str
model: Optional[str] # None = ทุกโมเดล
daily_limit_usd: float
weekly_limit_usd: float
monthly_limit_usd: float
daily_warning_pct: float = 0.70 # แจ้งเตือนเมื่อใช้ 70%
weekly_warning_pct: float = 0.70
monthly_warning_pct: float = 0.70
class BudgetAlertSystem:
def __init__(self, db_path: str = "cost_tracking.db"):
self.conn = sqlite3.connect(db_path)
self.thresholds: List[BudgetThreshold] = []
self.alert_callbacks: List[Callable] = []
self._load_thresholds()
def _load_thresholds(self):
"""โหลด threshold จากฐานข้อมูลหรือ config"""
# ตัวอย่าง threshold
self.thresholds = [
BudgetThreshold(
caller_id="analytics-team",
model=None, # ทุกโมเดล
daily_limit_usd=50.0,
weekly_limit_usd=300.0,
monthly_limit_usd=1000.0
),
BudgetThreshold(
caller_id="chatbot-service",
model="deepseek-v3.2",
daily_limit_usd=20.0,
weekly_limit_usd=100.0,
monthly_limit_usd=400.0
),
BudgetThreshold(
caller_id="chatbot-service",
model="claude-sonnet-4.5",
daily_limit_usd=100.0,
weekly_limit_usd=500.0,
monthly_limit_usd=2000.0
),
]
def add_alert_callback(self, callback: Callable):
"""เพิ่ม function สำหรับส่ง alert"""
self.alert_callbacks.append(callback)
def check_budget(self, caller_id: str, model: str = None) -> dict:
"""ตรวจสอบงบประมาณปัจจุบัน"""
alerts = []
now = datetime.now()
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
week_start = today_start - timedelta(days=now.weekday())
month_start = today_start.replace(day=1)
# หา threshold ที่ match
matching_thresholds = [
t for t in self.thresholds
if t.caller_id == caller_id and (t.model is None or t.model == model)
]
for threshold in matching_thresholds:
# คำนวณ usage ปัจจุบัน
cursor = self.conn.cursor()
# Daily
cursor.execute("""
SELECT COALESCE(SUM(cost_usd), 0)
FROM api_usage
WHERE caller_id = ?
AND timestamp >= ?
AND (? IS NULL OR model = ?)
""", [caller_id, today_start, threshold.model, threshold.model])
daily_usage = cursor.fetchone()[0]
# Weekly
cursor.execute("""
SELECT COALESCE(SUM(cost_usd), 0)
FROM api_usage
WHERE caller_id = ?
AND timestamp >= ?
AND (? IS NULL OR model = ?)
""", [caller_id, week_start, threshold.model, threshold.model])
weekly_usage = cursor.fetchone()[0]
# Monthly
cursor.execute("""
SELECT COALESCE(SUM(cost_usd), 0)
FROM api_usage
WHERE caller_id = ?
AND timestamp >= ?
AND (? IS NULL OR model = ?)
""", [caller_id, month_start, threshold.model, threshold.model])
monthly_usage = cursor.fetchone()[0]
# ตรวจสอบ threshold
alert = self._check_threshold(
threshold, daily_usage, weekly_usage, monthly_usage, model
)
if alert:
alerts.extend(alert)
return {
"caller_id": caller_id,
"model": model,
"timestamp": now.isoformat(),
"alerts": alerts,
"has_critical": any(a["level"] == "critical" for a in alerts)
}
def _check_threshold(
self,
threshold: BudgetThreshold,
daily: float,
weekly: float,
monthly: float
) -> List[dict]:
"""ตรวจสอบว่าเกิน threshold หรือไม่"""
alerts = []
# Daily check
daily_pct = daily / threshold.daily_limit_usd
if daily_pct >= 1.0:
alerts.append({
"level": "critical",
"period": "daily",
"used": daily,
"limit": threshold.daily_limit_usd,
"message": f"⚠️ Daily budget EXCEEDED: ${daily:.2f} / ${threshold.daily_limit_usd}"
})
elif daily_pct >= threshold.daily_warning_pct:
alerts.append({
"level": "warning",
"period": "daily",
"used": daily,
"limit": threshold.daily_limit_usd,
"message": f"⚡ Daily budget warning: ${daily:.2f} / ${threshold.daily_limit_usd} ({daily_pct*100:.0f}%)"
})
# Weekly check
weekly_pct = weekly / threshold.weekly_limit_usd
if weekly_pct >= 1.0:
alerts.append({
"level": "critical",
"period": "weekly",
"used": weekly,
"limit": threshold.weekly_limit_usd,
"message": f"🚨 Weekly budget EXCEEDED: ${weekly:.2f} / ${threshold.weekly_limit_usd}"
})
elif weekly_pct >= threshold.weekly_warning_pct:
alerts.append({
"level": "warning",
"period": "weekly",
"used": weekly,
"limit": threshold.weekly_limit_usd,
"message": f"📊 Weekly budget warning: ${weekly:.2f} / ${threshold.weekly_limit_usd} ({weekly_pct*100:.0f}%)"
})
# Monthly check
monthly_pct = monthly / threshold.monthly_limit_usd
if monthly_pct >= 1.0:
alerts.append({
"level": "critical",
"period": "monthly",
"used": monthly,
"limit": threshold.monthly_limit_usd,
"message": f"💸 Monthly budget EXCEEDED: ${monthly:.2f} / ${threshold.monthly_limit_usd}"
})
elif monthly_pct >= threshold.monthly_warning_pct:
alerts.append({
"level": "warning",
"period": "monthly",
"used": monthly,
"limit": threshold.monthly_limit_usd,
"message": f"📈 Monthly budget warning: ${monthly:.2f} / ${threshold.monthly_limit_usd} ({monthly_pct*100:.0f}%)"
})
return alerts
def check_all_budgets(self) -> List[dict]:
"""ตรวจสอบทุก caller"""
cursor = self.conn.cursor()
cursor.execute("SELECT DISTINCT caller_id FROM api_usage")
callers = [row[0] for row in cursor.fetchall()]
results = []
for caller in callers:
result = self.check_budget(caller)
if result["alerts"]:
results.append(result)
return results
def close(self):
self.conn.close()
ตัวอย่างการใช้งาน
if __name__ == "__main__":
alert_system = BudgetAlertSystem()
# เพิ่ม callback สำหรับส่ง alert
def print_alert(alert_data):
print(f"\n📢 ALERT RECEIVED:")
for alert in alert_data["alerts"]:
print(f" {alert['message']}")
alert_system.add_alert_callback(print_alert)
# ตรวจสอบ budget ของทุก caller
all_alerts = alert_system.check_all_budgets()
for result in all_alerts:
for callback in alert_system.alert_callbacks:
callback(result)
alert_system.close()
ระบบรายงานประจำเดือนอัตโนมัติ
การสร้างรายงานอัตโนมัติช่วยให้เห็นภาพรวมการใช้งานและวางแผนงบประมาณล่วงหน้า
"""
Monthly Token Consumption Report Generator
สร้างรายงานการใช้งาน token รายเดือนอัตโนมัติ
"""
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
import json
@dataclass
class MonthlyReport:
"""โครงสร้างข้อมูลรายงานรายเดือน"""
month: str
total_cost_usd: float
total_input_tokens: int
total_output_tokens: int
total_requests: int
by_caller: Dict
by_model: Dict
by_day: Dict
top_spenders: List
recommendations: List
class MonthlyReportGenerator:
def __init__(self, db_path: str = "cost_tracking.db"):
self.conn = sqlite3.connect(db_path)
def generate_report(self, year: int, month: int) -> MonthlyReport:
"""สร้างรายงานสำหรับเดือนที่ระบุ"""
month_str = f"{year}-{month:02d}"
start_date = datetime(year, month, 1)
# หาวันสิ้นเดือน
if month == 12:
end_date = datetime(year + 1, 1, 1)
else:
end_date = datetime(year, month + 1, 1)
cursor = self.conn.cursor()
# Total summary
cursor.execute("""
SELECT
COALESCE(SUM(cost_usd), 0),
COALESCE(SUM(input_tokens), 0),
COALESCE(SUM(output_tokens), 0),
COUNT(*)
FROM api_usage
WHERE timestamp >= ? AND timestamp < ?
""", [start_date, end_date])
total_row = cursor.fetchone()
# By caller
cursor.execute("""
SELECT
caller_id,
SUM(cost_usd) as cost,
SUM(input_tokens) as input_tok,
SUM(output_tokens) as output_tok,
COUNT(*) as requests
FROM api_usage
WHERE timestamp >= ? AND timestamp < ?
GROUP BY caller_id
ORDER BY cost DESC
""", [start_date, end_date])
by_caller = {
row[0]: {
"cost": row[1],
"input_tokens": row[2],
"output_tokens": row[3],
"requests": row[4],
"pct_of_total": 0 # จะคำนวณ later
}
for row in cursor.fetchall()
}
# Calculate percentage
total_cost = total_row[0] if total_row[0] else 0
for caller in by_caller:
by_caller[caller]["pct_of_total"] = (
(by_caller[caller]["cost"] / total_cost * 100)
if total_cost > 0 else 0
)
# By model
cursor.execute("""
SELECT