ในยุคที่ AI API กลายเป็นหัวใจสำคัญของแอปพลิเคชันธุรกิจ การตรวจสอบการใช้งานและควบคุมต้นทุนจึงเป็นสิ่งจำเป็นอย่างยิ่ง บทความนี้จะอธิบายวิธีสร้างระบบ Audit Log และ Cost Monitoring สำหรับ AI API อย่างครบวงจร พร้อมโค้ดตัวอย่างที่ใช้งานได้จริง
ทำไมต้องมีระบบ Audit Log และ Cost Monitoring
จากประสบการณ์ในการพัฒนาระบบ Enterprise หลายโปรเจกต์ พบว่าปัญหาที่พบบ่อยที่สุดคือ:
- ค่าใช้จ่ายที่บานปลาย — ไม่สามารถระบุได้ว่า Token ถูกใช้ไปกับอะไร
- ความปลอดภัย — ไม่มี Log สำหรับตรวจสอบการละเมิด API Key
- การ Debug — ยากที่จะติดตามปัญหาเมื่อเกิด Error
- Compliance — ขาดหลักฐานสำหรับ Audit Trail
ตารางเปรียบเทียบบริการ AI API
| เกณฑ์ | HolySheep AI | API อย่างเป็นทางการ | บริการ Relay อื่นๆ |
|---|---|---|---|
| ราคา GPT-4.1 | $8/MTok | $60/MTok | $15-30/MTok |
| ราคา Claude Sonnet 4.5 | $15/MTok | $90/MTok | $25-45/MTok |
| ราคา Gemini 2.5 Flash | $2.50/MTok | $10/MTok | $5-8/MTok |
| ราคา DeepSeek V3.2 | $0.42/MTok | $1/MTok | $0.60-0.80/MTok |
| Latency | <50ms | 80-200ms | 60-150ms |
| การจัดการ Cost | มี Dashboard ในตัว | ต้องใช้บริการเพิ่ม | มีบางส่วน |
| Audit Log | ครบถ้วน | พื้นฐาน | แตกต่างกัน |
| ช่องทางชำระเงิน | WeChat/Alipay | บัตรเครดิต | แตกต่างกัน |
| เครดิตฟรี | มีเมื่อลงทะเบียน | $5-18 | น้อยหรือไม่มี |
เหมาะกับใคร / ไม่เหมาะกับใคร
✅ เหมาะกับใคร
- องค์กรที่ต้องการประหยัดค่าใช้จ่าย AI API มากกว่า 85%
- ทีมพัฒนาที่ต้องการระบบ Audit Log ที่ครบถ้วน
- ธุรกิจในตลาดเอเชียที่ใช้ WeChat/Alipay
- ผู้ที่ต้องการ Latency ต่ำ (<50ms)
- Startup ที่ต้องการเริ่มต้นด้วยเครดิตฟรี
❌ ไม่เหมาะกับใคร
- องค์กรที่ต้องการ Support 24/7 แบบ Dedicated
- โปรเจกต์ที่ต้องการ SLA สูงมาก (99.99%+)
- ผู้ที่ต้องการใช้บริการเฉพาะของผู้ให้บริการโดยตรงเท่านั้น
ราคาและ ROI
การใช้ HolySheep AI สามารถประหยัดค่าใช้จ่ายได้อย่างมหาศาล:
- GPT-4.1: ประหยัด 86.7% ($60 → $8 ต่อ MTok)
- Claude Sonnet 4.5: ประหยัด 83.3% ($90 → $15 ต่อ MTok)
- DeepSeek V3.2: ประหยัด 58% ($1 → $0.42 ต่อ MTok)
สำหรับองค์กรที่ใช้งาน 100 ล้าน Token ต่อเดือน การใช้ HolySheep จะประหยัดได้หลายหมื่นบาทต่อเดือน
โครงสร้างระบบ Audit Log พื้นฐาน
ก่อนอื่น มาดูโครงสร้างพื้นฐานของระบบ Audit Log ที่ควรมี:
"""
AI API Audit Logging System
โครงสร้างพื้นฐานสำหรับบันทึกการใช้งาน AI API
"""
import json
import sqlite3
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import Optional, Dict, Any
from contextlib import contextmanager
@dataclass
class AuditLogEntry:
"""โครงสร้างข้อมูลสำหรับบันทึกการใช้งาน API"""
id: Optional[int] = None
timestamp: str = ""
request_id: str = ""
api_key_hash: str = ""
model: str = ""
input_tokens: int = 0
output_tokens: int = 0
total_tokens: int = 0
cost_usd: float = 0.0
latency_ms: float = 0.0
status: str = "success"
error_message: Optional[str] = None
user_id: Optional[str] = None
endpoint: str = ""
request_body_hash: Optional[str] = None
response_body_hash: Optional[str] = None
ip_address: Optional[str] = None
metadata: Optional[str] = None
class AuditLogger:
"""คลาสสำหรับจัดการ Audit Log ของ AI API"""
def __init__(self, db_path: str = "audit_logs.db"):
self.db_path = db_path
self._init_database()
def _init_database(self):
"""สร้างตารางฐานข้อมูลถ้ายังไม่มี"""
with self._get_connection() as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS audit_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
request_id TEXT UNIQUE NOT NULL,
api_key_hash TEXT NOT NULL,
model TEXT NOT NULL,
input_tokens INTEGER DEFAULT 0,
output_tokens INTEGER DEFAULT 0,
total_tokens INTEGER DEFAULT 0,
cost_usd REAL DEFAULT 0.0,
latency_ms REAL DEFAULT 0.0,
status TEXT DEFAULT 'success',
error_message TEXT,
user_id TEXT,
endpoint TEXT NOT NULL,
request_body_hash TEXT,
response_body_hash TEXT,
ip_address TEXT,
metadata TEXT,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# สร้าง Index สำหรับค้นหาเร็ว
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_timestamp
ON audit_logs(timestamp)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_api_key_hash
ON audit_logs(api_key_hash)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_user_id
ON audit_logs(user_id)
""")
@contextmanager
def _get_connection(self):
"""Context Manager สำหรับเชื่อมต่อฐานข้อมูล"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def log_request(self, entry: AuditLogEntry) -> int:
"""บันทึกการใช้งาน API"""
entry.timestamp = datetime.utcnow().isoformat()
with self._get_connection() as conn:
cursor = conn.execute("""
INSERT INTO audit_logs (
timestamp, request_id, api_key_hash, model,
input_tokens, output_tokens, total_tokens,
cost_usd, latency_ms, status, error_message,
user_id, endpoint, request_body_hash,
response_body_hash, ip_address, metadata
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
entry.timestamp, entry.request_id, entry.api_key_hash,
entry.model, entry.input_tokens, entry.output_tokens,
entry.total_tokens, entry.cost_usd, entry.latency_ms,
entry.status, entry.error_message, entry.user_id,
entry.endpoint, entry.request_body_hash,
entry.response_body_hash, entry.ip_address, entry.metadata
))
return cursor.lastrowid
def get_daily_usage(self, start_date: str, end_date: str) -> list:
"""ดึงข้อมูลการใช้งานรายวัน"""
with self._get_connection() as conn:
cursor = conn.execute("""
SELECT
DATE(timestamp) as date,
model,
COUNT(*) as request_count,
SUM(input_tokens) as total_input_tokens,
SUM(output_tokens) as total_output_tokens,
SUM(total_tokens) as total_tokens,
SUM(cost_usd) as total_cost
FROM audit_logs
WHERE timestamp BETWEEN ? AND ?
GROUP BY DATE(timestamp), model
ORDER BY date DESC
""", (start_date, end_date))
return [dict(row) for row in cursor.fetchall()]
def get_user_usage(self, user_id: str, limit: int = 100) -> list:
"""ดึงข้อมูลการใช้งานของผู้ใช้รายบุคคล"""
with self._get_connection() as conn:
cursor = conn.execute("""
SELECT * FROM audit_logs
WHERE user_id = ?
ORDER BY timestamp DESC
LIMIT ?
""", (user_id, limit))
return [dict(row) for row in cursor.fetchall()]
def get_cost_summary(self, start_date: str, end_date: str) -> dict:
"""ดึงสรุปค่าใช้จ่าย"""
with self._get_connection() as conn:
cursor = conn.execute("""
SELECT
COUNT(*) as total_requests,
SUM(total_tokens) as total_tokens,
SUM(cost_usd) as total_cost,
AVG(latency_ms) as avg_latency,
SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as error_count
FROM audit_logs
WHERE timestamp BETWEEN ? AND ?
""", (start_date, end_date))
row = cursor.fetchone()
return dict(row) if row else {}
ตัวอย่างการใช้งาน
if __name__ == "__main__":
logger = AuditLogger("production_audit.db")
# บันทึกตัวอย่าง
entry = AuditLogEntry(
request_id="req_abc123",
api_key_hash="hash_xxx",
model="gpt-4.1",
input_tokens=1000,
output_tokens=500,
total_tokens=1500,
cost_usd=0.012,
latency_ms=45.2,
user_id="user_001",
endpoint="/v1/chat/completions"
)
log_id = logger.log_request(entry)
print(f"บันทึกสำเร็จ: ID={log_id}")
# ดึงสรุปค่าใช้จ่าย
summary = logger.get_cost_summary("2025-01-01", "2025-12-31")
print(f"สรุปค่าใช้จ่าย: {summary}")ระบบ Cost Monitoring สำหรับ HolySheep API
ต่อไปจะเป็นระบบ Cost Monitoring ที่ใช้งานได้จริงกับ HolySheep AI:
"""
AI API Cost Monitoring System
ระบบตรวจสอบต้นทุนและการใช้งานสำหรับ HolySheep AI
"""
import hashlib
import time
import json
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
import requests
===== การตั้งค่า =====
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
===== โครงสร้างราคา (อัปเดต 2026) =====
MODEL_PRICING = {
"gpt-4.1": {"input": 2.0, "output": 8.0}, # $8/MTok output
"claude-sonnet-4.5": {"input": 3.0, "output": 15.0}, # $15/MTok output
"gemini-2.5-flash": {"input": 0.125, "output": 2.50}, # $2.50/MTok output
"deepseek-v3.2": {"input": 0.14, "output": 0.42}, # $0.42/MTok output
}
@dataclass
class CostAlert:
"""การแจ้งเตือนเมื่อค่าใช้จ่ายเกินกำหนด"""
level: str # info, warning, critical
message: str
threshold_type: str
current_value: float
threshold_value: float
class CostMonitor:
"""ระบบตรวจสอบต้นทุน AI API"""
def __init__(self, db_path: str = "cost_monitor.db"):
self.db_path = db_path
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
})
self._init_database()
def _init_database(self):
"""สร้างฐานข้อมูลสำหรับเก็บข้อมูลต้นทุน"""
with self._get_connection() as conn:
# ตารางเก็บข้อมูลการใช้งานราย request
conn.execute("""
CREATE TABLE IF NOT EXISTS request_logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
request_id TEXT UNIQUE,
model TEXT NOT NULL,
input_tokens INTEGER,
output_tokens INTEGER,
total_tokens INTEGER,
cost_usd REAL,
latency_ms REAL,
status_code INTEGER,
success BOOLEAN,
error_detail TEXT,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# ตารางเก็บข้อมูล Budget
conn.execute("""
CREATE TABLE IF NOT EXISTS budgets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
limit_usd REAL NOT NULL,
period_type TEXT NOT NULL, -- daily, weekly, monthly
alert_threshold REAL DEFAULT 0.8,
active BOOLEAN DEFAULT 1,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# ตารางเก็บข้อมูล Alert History
conn.execute("""
CREATE TABLE IF NOT EXISTS alert_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
alert_level TEXT NOT NULL,
message TEXT NOT NULL,
current_spend_usd REAL,
budget_limit_usd REAL,
acknowledged BOOLEAN DEFAULT 0
)
""")
@property
def connection(self):
return self._get_connection()
def _get_connection(self):
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""คำนวณค่าใช้จ่ายจากจำนวน Token"""
pricing = MODEL_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)
def _hash_api_key(self) -> str:
"""สร้าง Hash ของ API Key เพื่อเก็บใน Log"""
return hashlib.sha256(API_KEY.encode()).hexdigest()[:16]
def call_api_with_logging(
self,
model: str,
messages: List[Dict],
max_tokens: int = 1000,
temperature: float = 0.7,
user_id: Optional[str] = None
) -> Tuple[Optional[Dict], float]:
"""
เรียก API พร้อมบันทึกข้อมูลการใช้งาน
Returns:
Tuple[response_data, cost]
"""
start_time = time.time()
request_id = f"req_{int(start_time * 1000)}"
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature
}
try:
response = self.session.post(
f"{BASE_URL}/chat/completions",
json=payload,
timeout=30
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", 0)
cost = self._calculate_cost(model, input_tokens, output_tokens)
# บันทึกลงฐานข้อมูล
self._log_request(
request_id=request_id,
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
total_tokens=total_tokens,
cost_usd=cost,
latency_ms=latency_ms,
status_code=response.status_code,
success=True
)
return data, cost
else:
# บันทึก Error
self._log_request(
request_id=request_id,
model=model,
input_tokens=0,
output_tokens=0,
total_tokens=0,
cost_usd=0,
latency_ms=latency_ms,
status_code=response.status_code,
success=False,
error_detail=response.text
)
return None, 0.0
except Exception as e:
latency_ms = (time.time() - start_time) * 1000
self._log_request(
request_id=request_id,
model=model,
input_tokens=0,
output_tokens=0,
total_tokens=0,
cost_usd=0,
latency_ms=latency_ms,
status_code=0,
success=False,
error_detail=str(e)
)
raise
def _log_request(
self,
request_id: str,
model: str,
input_tokens: int,
output_tokens: int,
total_tokens: int,
cost_usd: float,
latency_ms: float,
status_code: int,
success: bool,
error_detail: Optional[str] = None
):
"""บันทึกข้อมูลการใช้งานลงฐานข้อมูล"""
conn = self._get_connection()
try:
conn.execute("""
INSERT INTO request_logs (
timestamp, request_id, model, input_tokens,
output_tokens, total_tokens, cost_usd, latency_ms,
status_code, success, error_detail
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
datetime.utcnow().isoformat(),
request_id, model, input_tokens,
output_tokens, total_tokens, cost_usd, latency_ms,
status_code, success, error_detail
))
conn.commit()
finally:
conn.close()
def get_current_spend(self, period: str = "daily") -> Dict:
"""ดึงค่าใช้จ่ายปัจจุบันตามช่วงเวลา"""
conn = self._get_connection()
try:
if period == "daily":
date_filter = "DATE(timestamp) = DATE('now')"
elif period == "weekly":
date_filter = "timestamp >= DATE('now', '-7 days')"
elif period == "monthly":
date_filter = "timestamp >= DATE('now', 'start of month')"
else:
date_filter = "1=1"
cursor = conn.execute(f"""
SELECT
COUNT(*) as total_requests,
SUM(total_tokens) as total_tokens,
SUM(cost_usd) as total_cost,
AVG(latency_ms) as avg_latency,
SUM(CASE WHEN NOT success THEN 1 ELSE 0 END) as error_count
FROM request_logs
WHERE {date_filter}
""")
row = cursor.fetchone()
return dict(row) if row else {}
finally:
conn.close()
def get_usage_by_model(self, days: int = 30) -> List[Dict]:
"""ดึงข้อมูลการใช้งานแยกตาม Model"""
conn = self._get_connection()
try:
cursor = conn.execute("""
SELECT
model,
COUNT(*) as request_count,
SUM(input_tokens) as total_input,
SUM(output_tokens) as total_output,
SUM(total_tokens) as total_tokens,
SUM(cost_usd) as total_cost,
AVG(latency_ms) as avg_latency
FROM request_logs
WHERE timestamp >= DATE('now', ?)
GROUP BY model
ORDER BY total_cost DESC
""", (f"-{days} days",))
return [dict(row) for row in cursor.fetchall()]
finally:
conn.close()
def check_budget_alerts(self) -> List[CostAlert]:
"""ตรวจสอบและสร้าง Alert ถ้าค่าใช้จ่ายเกิน Budget"""
alerts = []
conn = self._get_connection()
try:
# ดึง Budget ที่ active
cursor = conn.execute("""
SELECT * FROM budgets WHERE active = 1
""")
budgets = [dict(row) for row in cursor.fetchall()]
for budget in budgets:
# คำนวณค่าใช้จ่ายตามช่วงเวลา
if budget["period_type"] == "daily":
current = self.get_current_spend("daily")["total_cost"] or 0
period_name = "วันนี้"
elif budget["period_type"] == "weekly":
current = self.get_current_spend("weekly")["total_cost"] or 0
period_name = "สัปดาห์นี้"
else:
current = self.get_current_spend("monthly")["total_cost"] or 0
period_name = "เดือนนี้"
limit = budget["limit_usd"]
threshold = budget["alert_threshold"]
# ตรวจสอบ Alert Level
ratio = current / limit if limit > 0 else 0
if ratio >= 1.0:
level = "critical"
message = f"⚠️ ค่าใช้จ่ายเกิน Budget! ({period_name}: ${current:.2f}/${limit:.2f})"
elif ratio >= threshold:
level = "warning"
message = f"⚡ ค่าใช้จ่ายใกล้ถึงขีดจำกัด ({ratio*100:.1f}%) - {period_name}: ${current:.2f}/${limit:.2f}"
else:
continue
alerts.append(CostAlert(
level=level,
message=message,
threshold_type=budget["period_type"],
current_value=current,
threshold_value=limit
))
# บันทึก Alert ลงฐานข้อมูล
conn.execute("""
INSERT INTO alert_history
(timestamp, alert_level, message, current_spend_usd, budget_limit_usd)
VALUES (?, ?, ?, ?, ?)
""", (
datetime.utcnow().isoformat(),
level, message, current, limit
))
conn.commit()
finally:
conn.close()
return alerts
def set_budget(self, name: str, limit_usd: float, period: str, threshold: float = 0.8):
"""ตั้งค่า Budget ใหม่"""
conn = self._get_connection()
try:
conn.execute("""
INSERT INTO budgets (name, limit_usd, period_type, alert_threshold)
VALUES (?, ?, ?, ?)
""", (name, limit_usd, period, threshold))
conn.commit()
finally:
conn.close()
def generate_report(self, days: int = 30) -> Dict:
"""สร้างรายงานสรุปการใช้งาน"""
return {
"period": f"{days} วันที่ผ่านมา",
"current_spend": self.get_current_spend("monthly"),
"usage_by_model": self.get_usage_by_model(days),
"alerts": self.check_budget_alerts(),
"recommendations": self._generate_recommendations(days)
}
def _generate_recommendations(self, days: int) -> List[str]:
"""สร้างคำแนะนำจากข้อมูลการใช้งาน"""
recommendations = []
usage = self.get_usage_by_model(days)
if usage:
# หา Model ที่ใช้มากที่สุด
top_model = usage[0]
if top_model["total_cost"] > 100:
recommendations.append(
f"พิจารณาใช้ Model ราคาถูกกว่าสำหรับ Task ที่ไม่ซับซ้อน "
f