ในฐานะ SRE ที่ดูแลระบบ AI API มาหลายปี ผมเคยเจอสถานการณ์วิกฤตมาแล้วมากมาย เช่น ระบบแชทบอทลูกค้าสัมพันธ์ของอีคอมเมิร์ซที่ล่มกลางดึกเพราะไม่มีการติดตาม SLO หรือระบบ RAG องค์กรที่ response time พุ่งสูงผิดปกติโดยไม่มีใครสังเกตเห็นจนลูกค้า投诉มาถึง CEO บทความนี้จะสอนวิธีกำหนด SLO สำหรับ AI API อย่างมืออาชีพ และสร้างระบบติดตามที่ครอบคลุม โดยใช้ HolySheep AI เป็นตัวอย่างการ implement จริง
ทำไม AI API ต้องมี SLO เฉพาะ
AI API แตกต่างจาก API ทั่วไปตรงที่มีความไม่แน่นอนสูง (latency variance) เพราะขึ้นอยู่กับปัจจัยหลายอย่าง ได้แก่:
- Model complexity — โมเดลใหญ่ต้องใช้เวลาประมวลผลนานกว่า
- Queue depth — จำนวน request ที่รอในคิวส่งผลต่อ response time
- Context length — prompt ยาวใช้ทรัพยากรมากขึ้นแบบ exponential
- Rate limiting — การจำกัด request rate ของ provider
จากประสบการณ์ตรงที่ implement ระบบ monitor มาหลายระบบ ผมพบว่า AI API ที่ไม่มี SLO ที่ชัดเจนจะมีปัญหา "silent degradation" คือ performance แย่ลงเรื่อยๆ โดยไม่มีใครสังเกตจนสร้างความเสียหายให้ธุรกิจ
กรณีศึกษา: ระบบ AI ลูกค้าสัมพันธ์อีคอมเมิร์ซ
ผมเคยดูแลระบบแชทบอทตอบคำถามลูกค้าของร้านค้าออนไลน์ขนาดใหญ่ ช่วง Black Friday ระบบรับ request พุ่งสูงผิดปกติจาก 1,000 req/min เป็น 50,000 req/min ภายใน 1 ชั่วโมง ถ้าไม่มี SLO dashboard เราคงไม่รู้ว่า:
- P50 latency เพิ่มจาก 800ms เป็น 4,500ms
- P99 latency พุ่งเกิน 30 วินาที
- Error rate เพิ่มจาก 0.1% เป็น 8.5%
หลังจาก implement SLO tracking ด้วย Prometheus + Grafana เราสามารถตั้ง alert และ scale out ระบบทันก่อนลูกค้าได้รับประสบการณ์ที่แย่
กำหนด SLO สำหรับ AI API
SLO ที่ดีต้องครอบคลุม 4 มิติหลัก เรียกว่า "ALEE":
- Availability — uptime เป็นเปอร์เซ็นต์
- Latency — response time ที่แต่ละ percentile
- Error rate — เปอร์เซ็นต์ของ request ที่ล้มเหลว
- Exactness — ความถูกต้องของ output (สำหรับ AI พิเศษมาก)
ตัวอย่าง SLO Dashboard
โค้ดต่อไปนี้แสดงการสร้าง SLO tracker สำหรับ HolySheep AI API ด้วย Python:
#!/usr/bin/env python3
"""
AI API SLO Tracker - HolySheep AI Edition
Author: HolySheep AI Team
"""
import time
import httpx
import asyncio
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta
import statistics
@dataclass
class SLOReport:
"""โครงสร้างข้อมูลรายงาน SLO"""
service_name: str
timestamp: datetime
total_requests: int
successful_requests: int
failed_requests: int
# Latency percentiles (milliseconds)
latency_p50: float
latency_p95: float
latency_p99: float
# SLO compliance
availability_slo: float # เป้าหมาย uptime เช่น 99.9%
latency_slo: float # เป้าหมาย P99 latency เช่น 2000ms
error_rate_slo: float # เป้าหมาย error rate เช่น 1.0%
@property
def actual_availability(self) -> float:
return (self.successful_requests / self.total_requests * 100)
if self.total_requests > 0 else 0.0
@property
def availability_compliance(self) -> bool:
return self.actual_availability >= self.availability_slo
@property
def latency_compliance(self) -> bool:
return self.latency_p99 <= self.latency_slo
@property
def error_rate_compliance(self) -> bool:
actual_error_rate = (self.failed_requests / self.total_requests * 100)
if self.total_requests > 0 else 0.0
return actual_error_rate <= self.error_rate_slo
def to_markdown(self) -> str:
return f"""
SLO Report: {self.service_name}
**Timestamp:** {self.timestamp.isoformat()}
Request Statistics
- Total Requests: {self.total_requests:,}
- Successful: {self.successful_requests:,}
- Failed: {self.failed_requests:,}
Latency (ms)
| Percentile | Value | SLO Target | Status |
|------------|-------|------------|--------|
| P50 | {self.latency_p50:.2f} | - | {'✅' if self.latency_p50 <= self.latency_slo * 0.5 else '⚠️'} |
| P95 | {self.latency_p95:.2f} | - | {'✅' if self.latency_p95 <= self.latency_slo * 0.8 else '⚠️'} |
| P99 | {self.latency_p99:.2f} | {self.latency_slo} | {'✅' if self.latency_compliance else '❌'} |
SLO Compliance
| Metric | Actual | Target | Status |
|--------|--------|--------|--------|
| Availability | {self.actual_availability:.4f}% | {self.availability_slo}% | {'✅' if self.availability_compliance else '❌'} |
| Error Rate | {self.failed_requests/self.total_requests*100 if self.total_requests > 0 else 0:.4f}% | {self.error_rate_slo}% | {'✅' if self.error_rate_compliance else '❌'} |
"""
class AISLOService:
"""บริการติดตาม SLO สำหรับ AI API"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, service_name: str = "default"):
self.api_key = api_key
self.service_name = service_name
self.request_log: List[dict] = []
# SLO targets - กำหนดตาม business requirement
self.slo_targets = {
"availability": 99.9, # 99.9% uptime
"latency_p99": 2000, # P99 < 2 วินาที
"error_rate": 1.0, # error < 1%
}
async def call_ai_api(
self,
prompt: str,
model: str = "gpt-4.1",
max_tokens: int = 1000
) -> dict:
"""เรียก HolyShehe AI API พร้อมวัด latency"""
start_time = time.perf_counter()
async with httpx.AsyncClient(timeout=30.0) as client:
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens
}
try:
response = await client.post(
f"{self.BASE_URL}/chat/completions",
headers=headers,
json=payload
)
latency_ms = (time.perf_counter() - start_time) * 1000
result = {
"timestamp": datetime.now(),
"latency_ms": latency_ms,
"status_code": response.status_code,
"success": response.status_code == 200,
"model": model
}
self.request_log.append(result)
return result
except httpx.TimeoutException:
latency_ms = (time.perf_counter() - start_time) * 1000
result = {
"timestamp": datetime.now(),
"latency_ms": latency_ms,
"status_code": 408,
"success": False,
"model": model,
"error": "Timeout"
}
self.request_log.append(result)
return result
except Exception as e:
latency_ms = (time.perf_counter() - start_time) * 1000
result = {
"timestamp": datetime.now(),
"latency_ms": latency_ms,
"status_code": 500,
"success": False,
"model": model,
"error": str(e)
}
self.request_log.append(result)
return result
def generate_slo_report(self, window_minutes: int = 60) -> SLOReport:
"""สร้างรายงาน SLO จาก request log"""
cutoff_time = datetime.now() - timedelta(minutes=window_minutes)
recent_logs = [
log for log in self.request_log
if log["timestamp"] >= cutoff_time
]
total = len(recent_logs)
successful = sum(1 for log in recent_logs if log["success"])
failed = total - successful
latencies = [log["latency_ms"] for log in recent_logs]
def percentile(data: List[float], p: float) -> float:
if not data:
return 0.0
sorted_data = sorted(data)
index = int(len(sorted_data) * p / 100)
return sorted_data[min(index, len(sorted_data) - 1)]
return SLOReport(
service_name=self.service_name,
timestamp=datetime.now(),
total_requests=total,
successful_requests=successful,
failed_requests=failed,
latency_p50=percentile(latencies, 50),
latency_p95=percentile(latencies, 95),
latency_p99=percentile(latencies, 99),
availability_slo=self.slo_targets["availability"],
latency_slo=self.slo_targets["latency_p99"],
error_rate_slo=self.slo_targets["error_rate"]
)
ตัวอย่างการใช้งาน
async def main():
slo_service = AISLOService(
api_key="YOUR_HOLYSHEEP_API_KEY",
service_name="ecommerce-chatbot"
)
# จำลอง load test
tasks = []
for i in range(100):
prompt = f"ตอบคำถามลูกค้าหมายเลข {i}: สินค้านี้มีสีอะไรบ้าง?"
tasks.append(slo_service.call_ai_api(prompt))
results = await asyncio.gather(*tasks)
# สร้างรายงาน SLO
report = slo_service.generate_slo_report(window_minutes=5)
print(report.to_markdown())
if __name__ == "__main__":
asyncio.run(main())
Prometheus Metrics สำหรับ SLO Tracking
โค้ดต่อไปนี้แสดงการสร้าง Prometheus metrics exporter ที่รวมเข้ากับ Grafana dashboard ได้เลย:
#!/usr/bin/env python3
"""
Prometheus Metrics Exporter for AI API SLO
Works with HolySheep AI API
"""
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from flask import Flask, Response
import time
import random
app = Flask(__name__)
Define Prometheus metrics
ai_api_requests_total = Counter(
'ai_api_requests_total',
'Total number of AI API requests',
['model', 'status']
)
ai_api_latency_seconds = Histogram(
'ai_api_latency_seconds',
'AI API request latency in seconds',
['model'],
buckets=[0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)
ai_api_slo_availability = Gauge(
'ai_api_slo_availability_percent',
'Current availability SLO compliance'
)
ai_api_slo_latency_p99 = Gauge(
'ai_api_slo_latency_p99_ms',
'Current P99 latency SLO status'
)
ai_api_error_rate = Gauge(
'ai_api_error_rate_percent',
'Current error rate percentage'
)
SLO targets (จาก business requirement)
SLO_TARGETS = {
"availability": 99.9,
"latency_p99_ms": 2000,
"error_rate_percent": 1.0
}
def call_holysheep_api(prompt: str, model: str = "gpt-4.1") -> dict:
"""
เรียก HolySheep AI API พร้อม record metrics
base_url: https://api.holysheep.ai/v1
"""
import httpx
start_time = time.perf_counter()
status = "success"
try:
with httpx.Client(timeout=30.0) as client:
response = client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}]
}
)
if response.status_code != 200:
status = "error"
except httpx.TimeoutException:
status = "timeout"
except Exception:
status = "error"
latency = time.perf_counter() - start_time
# Record metrics
ai_api_requests_total.labels(model=model, status=status).inc()
ai_api_latency_seconds.labels(model=model).observe(latency)
return {"status": status, "latency": latency}
def calculate_slo_metrics():
"""
คำนวณ SLO metrics ปัจจุบัน