ในระบบ AI ระดับ Production ที่ต้องรับโหลดสูง อัตราความล้มเหลว (Failure Rate) ของ API เป็นตัวชี้วัดสำคัญที่กำหนดคุณภาพบริการโดยตรง บทความนี้จะพาคุณสร้างระบบ Monitoring แบบครบวงจร ตั้งแต่การเก็บสถิติ การวิเคราะห์สาเหตุ ไปจนถึงการสร้างระบบ Fallback ที่ทำงานได้อย่างเสถียร พร้อมโค้ด Production-Ready ที่ใช้งานได้จริง
ทำไมต้องวัดอัตราความล้มเหลวอย่างเป็นระบบ
จากประสบการณ์ในการดูแลระบบที่รับโหลดมากกว่า 100,000 คำขอต่อวัน พบว่า API ที่ไม่มีการ Monitor อย่างเป็นระบบจะมีปัญหา Silent Failure ที่ยากต่อการตรวจจับ การวัดความล้มเหลวอย่างเป็นระบบช่วยให้เราสามารถคาดการณ์ปัญหาได้ก่อนที่จะส่งผลกระทบต่อผู้ใช้
สถาปัตยกรรมระบบติดตามความล้มเหลวแบบ End-to-End
ระบบที่ดีต้องครอบคลุมทั้ง Client-Side และ Server-Side Metrics โดยใช้ OpenTelemetry ร่วมกับ Prometheus และ Grafana สำหรับ Visualization แบบ Real-Time
import asyncio
import aiohttp
import time
from dataclasses import dataclass, field
from typing import Optional, Dict, List
from collections import defaultdict
import logging
from datetime import datetime, timedelta
import hashlib
import json
@dataclass
class APIFailureRecord:
"""โครงสร้างข้อมูลสำหรับบันทึกความล้มเหลวแต่ละครั้ง"""
timestamp: datetime
endpoint: str
error_type: str
error_message: str
status_code: Optional[int]
latency_ms: float
retry_count: int
model: str
request_id: str
metadata: Dict = field(default_factory=dict)
@dataclass
class APIMetrics:
"""โครงสร้าง Metrics สำหรับสรุปผล"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_latency_ms: float = 0.0
error_counts: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
latency_percentiles: Dict[str, float] = field(default_factory=dict)
@property
def failure_rate(self) -> float:
"""คำนวณอัตราความล้มเหลวเป็นเปอร์เซ็นต์"""
if self.total_requests == 0:
return 0.0
return (self.failed_requests / self.total_requests) * 100
@property
def success_rate(self) -> float:
"""คำนวณอัตราความสำเร็จเป็นเปอร์เซ็นต์"""
return 100.0 - self.failure_rate
@property
def average_latency_ms(self) -> float:
"""คำนวณความหน่วงเฉลี่ย"""
if self.total_requests == 0:
return 0.0
return self.total_latency_ms / self.total_requests
class AIAPIFailureTracker:
"""
ระบบติดตามและวิเคราะห์ความล้มเหลวของ AI API
รองรับ Multi-Provider, Retry Logic และ Circuit Breaker
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, enable_logging: bool = True):
self.api_key = api_key
self.enable_logging = enable_logging
self.logger = logging.getLogger(__name__)
# ที่เก็บข้อมูล Metrics
self.metrics: Dict[str, APIMetrics] = defaultdict(
lambda: APIMetrics()
)
self.failure_records: List[APIFailureRecord] = []
self.max_records = 10000 # จำกัดจำนวน records ใน memory
# Circuit Breaker State
self.circuit_state: Dict[str, str] = defaultdict(lambda: "CLOSED")
self.circuit_failures: Dict[str, int] = defaultdict(int)
self.circuit_last_failure: Dict[str, datetime] = {}
self.circuit_threshold = 5 # จำนวนความล้มเหลวที่จะเปิด Circuit
self.circuit_timeout = 60 # วินาทีก่อนลองใหม่
# Error Categories
self.error_categories = {
'rate_limit': [429],
'server_error': [500, 502, 503, 504],
'auth_error': [401, 403],
'client_error': [400],
'timeout': ['timeout'],
'connection': ['connection', 'network']
}
def _categorize_error(self, error: Exception, status_code: Optional[int]) -> str:
"""จำแนกประเภทของ Error"""
error_str = str(error).lower()
for category, patterns in self.error_categories.items():
if category in ['rate_limit', 'server_error', 'auth_error', 'client_error']:
if status_code in patterns:
return category
else:
if any(p in error_str for p in patterns):
return category
return 'unknown'
def _generate_request_id(self, endpoint: str, payload: dict) -> str:
"""สร้าง Request ID ที่ไม่ซ้ำกัน"""
content = f"{endpoint}:{json.dumps(payload, sort_keys=True)}:{time.time()}"
return hashlib.md5(content.encode()).hexdigest()[:16]
def _should_allow_request(self, model: str) -> bool:
"""ตรวจสอบ Circuit Breaker"""
state = self.circuit_state[model]
if state == "CLOSED":
return True
if state == "HALF_OPEN":
# อนุญาตให้ลอง Request แค่ 1 ครั้ง
return True
# State = OPEN
last_failure = self.circuit_last_failure.get(model)
if last_failure:
elapsed = (datetime.now() - last_failure).total_seconds()
if elapsed >= self.circuit_timeout:
self.circuit_state[model] = "HALF_OPEN"
self.logger.info(f"Circuit Breaker for {model} transitioning to HALF_OPEN")
return True
return False
def _record_failure(self, record: APIFailureRecord):
"""บันทึกความล้มเหลว"""
self.failure_records.append(record)
# ลบ records เก่าออกถ้าเกิน limit
if len(self.failure_records) > self.max_records:
self.failure_records = self.failure_records[-self.max_records:]
# อัพเดท Circuit Breaker
model = record.model
self.circuit_failures[model] += 1
self.circuit_last_failure[model] = record.timestamp
if self.circuit_failures[model] >= self.circuit_threshold:
if self.circuit_state[model] == "CLOSED":
self.circuit_state[model] = "OPEN"
self.logger.warning(
f"Circuit Breaker OPENED for {model} after {self.circuit_failures[model]} failures"
)
async def call_with_tracking(
self,
model: str,
prompt: str,
max_retries: int = 3,
timeout: int = 60
) -> tuple[Optional[dict], APIFailureRecord]:
"""
เรียก API พร้อมติดตามความล้มเหลว
Returns:
(response, failure_record) - ถ้าสำเร็จ failure_record จะเป็น None
"""
endpoint = "/chat/completions"
url = f"{self.BASE_URL}{endpoint}"
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7,
"max_tokens": 2000
}
request_id = self._generate_request_id(endpoint, payload)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# ตรวจสอบ Circuit Breaker
if not self._should_allow_request(model):
failure_record = APIFailureRecord(
timestamp=datetime.now(),
endpoint=endpoint,
error_type="circuit_breaker",
error_message=f"Circuit Breaker is OPEN for {model}",
status_code=None,
latency_ms=0,
retry_count=0,
model=model,
request_id=request_id
)
self._record_failure(failure_record)
return None, failure_record
for retry_count in range(max_retries):
start_time = time.time()
try:
async with aiohttp.ClientSession() as session:
async with session.post(
url,
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=timeout)
) as response:
latency_ms = (time.time() - start_time) * 1000
# อัพเดท Metrics
metrics = self.metrics[model]
metrics.total_requests += 1
metrics.total_latency_ms += latency_ms
if response.status == 200:
metrics.successful_requests += 1
# Reset Circuit Breaker ถ้าสำเร็จ
if self.circuit_state[model] != "CLOSED":
self.circuit_state[model] = "CLOSED"
self.circuit_failures[model] = 0
result = await response.json()
return result, None
else:
error_text = await response.text()
metrics.failed_requests += 1
error_type = self._categorize_error(
Exception(error_text),
response.status
)
metrics.error_counts[error_type] += 1
failure_record = APIFailureRecord(
timestamp=datetime.now(),
endpoint=endpoint,
error_type=error_type,
error_message=error_text[:500],
status_code=response.status,
latency_ms=latency_ms,
retry_count=retry_count,
model=model,
request_id=request_id
)
# Retry กับบาง Error Types
if response.status in [429, 500, 502, 503, 504]:
wait_time = (2 ** retry_count) * 0.5
await asyncio.sleep(wait_time)
continue
self._record_failure(failure_record)
return None, failure_record
except asyncio.TimeoutError:
latency_ms = (time.time() - start_time) * 1000
error_type = "timeout"
except aiohttp.ClientError as e:
latency_ms = (time.time() - start_time) * 1000
error_type = self._categorize_error(e, None)
except Exception as e:
latency_ms = (time.time() - start_time) * 1000
error_type = self._categorize_error(e, None)
if self.enable_logging:
self.logger.error(f"Unexpected error: {str(e)}")
# บันทึกความล้มเหลว
metrics = self.metrics[model]
metrics.total_requests += 1
metrics.failed_requests += 1
metrics.total_latency_ms += latency_ms
metrics.error_counts[error_type] += 1
# Exponential Backoff
wait_time = (2 ** retry_count) * 0.5
await asyncio.sleep(wait_time)
# ครบจำนวน Retry แล้ว
failure_record = APIFailureRecord(
timestamp=datetime.now(),
endpoint=endpoint,
error_type="max_retries_exceeded",
error_message=f"Failed after {max_retries} retries",
status_code=None,
latency_ms=0,
retry_count=max_retries,
model=model,
request_id=request_id
)
self._record_failure(failure_record)
return None, failure_record
def get_metrics_summary(self, model: Optional[str] = None) -> Dict:
"""สร้าง Summary ของ Metrics"""
if model:
metrics = self.metrics[model]
return {
"model": model,
"total_requests": metrics.total_requests,
"successful_requests": metrics.successful_requests,
"failed_requests": metrics.failed_requests,
"failure_rate_percent": round(metrics.failure_rate, 4),
"success_rate_percent": round(metrics.success_rate, 4),
"average_latency_ms": round(metrics.average_latency_ms, 2),
"error_breakdown": dict(metrics.error_counts),
"circuit_state": self.circuit_state[model]
}
# Summary ทั้งหมด
return {
"models": {
model: {
"total_requests": m.total_requests,
"failure_rate_percent": round(m.failure_rate, 4),
"success_rate_percent": round(m.success_rate, 4),
"average_latency_ms": round(m.average_latency_ms, 2)
}
for model, m in self.metrics.items()
},
"total_failure_records": len(self.failure_records)
}
def get_recent_failures(self, minutes: int = 30, model: Optional[str] = None) -> List[Dict]:
"""ดึงความล้มเหลวล่าสุด"""
cutoff = datetime.now() - timedelta(minutes=minutes)
failures = [
{
"timestamp": r.timestamp.isoformat(),
"model": r.model,
"error_type": r.error_type,
"error_message": r.error_message,
"status_code": r.status_code,
"latency_ms": round(r.latency_ms, 2),
"retry_count": r.retry_count
}
for r in self.failure_records
if r.timestamp >= cutoff and (model is None or r.model == model)
]
return sorted(failures, key=lambda x: x["timestamp"], reverse=True)
ตัวอย่างการใช้งาน
async def main():
tracker = AIAPIFailureTracker(
api_key="YOUR_HOLYSHEEP_API_KEY",
enable_logging=True
)
# ทดสอบหลาย Models
test_prompts = [
("เขียนโค้ด Python สำหรับ Fibonacci", "gpt-4.1"),
("อธิบาย Quantum Computing", "claude-sonnet-4.5"),
("สรุปบทความนี้", "gemini-2.5-flash"),
("คำนวณ 2+2", "deepseek-v3.2")
]
for prompt, model in test_prompts:
response, failure = await tracker.call_with_tracking(model, prompt)
if failure:
print(f"❌ {model}: {failure.error_type} - {failure.error_message[:50]}")
else:
print(f"✅ {model}: Success")
# แสดง Summary
print("\n📊 Metrics Summary:")
print(json.dumps(tracker.get_metrics_summary(), indent=2))
if __name__ == "__main__":
asyncio.run(main())การทำ Benchmark เปรียบเทียบความน่าเชื่อถือระหว่าง Models
การเลือก Model ที่เหมาะสมต้องดูไม่ใช่แค่ราคา แต่ต้องพิจารณา Reliability และ Latency ด้วย ด้านล่างคือระบบ Benchmark ที่ทดสอบ Models ต่างๆ พร้อมวัดผลอย่างละเอียด
import asyncio
import aiohttp
import time
import statistics
from dataclasses import dataclass
from typing import List, Dict
from concurrent.futures import ThreadPoolExecutor
import random
@dataclass
class BenchmarkResult:
"""ผลลัพธ์ของการ Benchmark แต่ละ Model"""
model: str
total_requests: int
successful: int
failed: int
# Latency Statistics (milliseconds)
min_latency: float
max_latency: float
avg_latency: float
p50_latency: float
p95_latency: float
p99_latency: float
std_latency: float
# Reliability
success_rate: float
failure_rate: float
availability_score: float # คะแนนความพร้อมใช้งาน
# Cost Efficiency
cost_per_1k_success: float
effective_cost: float
# Error Breakdown
error_types: Dict[str, int]
def __str__(self):
return f"""
╔══════════════════════════════════════════════════════════════╗
║ Model: {self.model:<50} ║
╠══════════════════════════════════════════════════════════════╣
║ Total Requests: {self.total_requests:<41} ║
║ Success/Failed: {self.successful}/{self.failed:<42} ║
║ Success Rate: {self.success_rate:.2f}%{' '*42} ║
╠══════════════════════════════════════════════════════════════╣
║ LATENCY (ms){' '*47} ║
║ Min/Avg/Max: {self.min_latency:.1f}/{self.avg_latency:.1f}/{self.max_latency:.1f}{' '*25} ║
║ P50/P95/P99: {self.p50_latency:.1f}/{self.p95_latency:.1f}/{self.p99_latency:.1f}{' '*25} ║
║ Std Dev: {self.std_latency:.2f}ms{' '*43} ║
╠══════════════════════════════════════════════════════════════╣
║ COST EFFICIENCY{' '*44} ║
║ Cost per 1K Success: ${self.cost_per_1k_success:.4f}{' '*33} ║
║ Availability Score: {self.availability_score:.2f}/100{' '*34} ║
╚══════════════════════════════════════════════════════════════╝
"""
class ModelReliabilityBenchmark:
"""
ระบบ Benchmark เปรียบเทียบความน่าเชื่อถือของ AI Models
วัดทั้ง Latency, Success Rate, และ Cost Efficiency
"""
BASE_URL = "https://api.holysheep.ai/v1"
# ราคาเป็น USD ต่อ Million Tokens (จาก HolySheep 2026)
MODEL_PRICING = {
"gpt-4.1": {"input": 8.0, "output": 8.0}, # $8/MTok
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0}, # $15/MTok
"gemini-2.5-flash": {"input": 2.50, "output": 2.50}, # $2.50/MTok
"deepseek-v3.2": {"input": 0.42, "output": 0.42} # $0.42/MTok
}
# Test Prompts หลากหลายระดับความยาว
TEST_PROMPTS = [
"ทักทาย",
"อธิบายการสังเคราะห์แสง",
"เขียน Python function สำหรับ merge sort",
"วิเคราะห์ข้อดีข้อเสียของ Solar Energy",
"แปลโค้ดนี้เป็น TypeScript: def hello(name): return f'Hello {name}'",
] * 20 # ทำซ้ำ 20 ครั้ง = 100 requests ต่อ model
def __init__(self, api_key: str):
self.api_key = api_key
self.results: Dict[str, BenchmarkResult] = {}
async def _make_request(
self,
session: aiohttp.ClientSession,
model: str,
prompt: str,
timeout: int = 30
) -> tuple[bool, float, str]:
"""ส่ง Request และวัดผล"""
url = f"{self.BASE_URL}/chat/completions"
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.7,
"max_tokens": 500
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
start = time.time()
try:
async with session.post(
url,
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=timeout)
) as response:
latency_ms = (time.time() - start) * 1000
if response.status == 200:
data = await response.json()
tokens_used = data.get('usage', {}).get('total_tokens', 0)
return True, latency_ms, ""
else:
error = await response.text()
return False, latency_ms, f"HTTP {response.status}"
except asyncio.TimeoutError:
return False, (time.time() - start) * 1000, "Timeout"
except Exception as e:
return False, (time.time() - start) * 1000, str(type(e).__name__)
async def benchmark_model(
self,
model: str,
requests: int = 100,
concurrency: int = 10
) -> BenchmarkResult:
"""Benchmark Model เดียว"""
print(f"\n🔄 Benchmarking {model}...")
latencies: List[float] = []
success_count = 0
failure_count = 0
error_types: Dict[str, int] = {}
# ใช้ Semaphore เพื่อควบคุม Concurrency
semaphore = asyncio.Semaphore(concurrency)
async def bounded_request(prompt: str):
async with semaphore:
async with aiohttp.ClientSession() as session:
return await self._make_request(session, model, prompt)
# ส่ง Requests ทั้งหมด
tasks = [
bounded_request(random.choice(self.TEST_PROMPTS))
for _ in range(requests)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for result in results:
if isinstance(result, Exception):
failure_count += 1
error_types[str(type(result).__name__)] = \
error_types.get(str(type(result).__name__), 0) + 1
else:
success, latency, error = result
if success:
success_count += 1
latencies.append(latency)
else:
failure_count += 1
error_types[error] = error_types.get(error, 0) + 1
# คำนวณ Statistics
latencies.sort()
n = len(latencies)
if n > 0:
avg_lat = statistics.mean(latencies)
std_lat = statistics.stdev(latencies) if n > 1 else 0
min_lat = min(latencies)
max_lat = max(latencies)
p50 = latencies[int(n * 0.50)]
p95 = latencies[int(n * 0.95)]
p99 = latencies[int(n * 0.99)]
else:
avg_lat = std_lat = min_lat = max_lat = p50 = p95 = p99 = 0
success_rate = (success_count / requests * 100) if requests > 0 else 0
failure_rate = 100 - success_rate
# คำนวณ Availability Score
# คิดจาก Success Rate (40%) + Latency (30%) + Consistency (30%)
latency_score = max(0, 100 - (avg_lat / 10)) # Penalty สำหรับ latency สูง
consistency_score = max(0, 100 - (std_lat / 5)) # Penalty สำหรับ latency ไม่คงที่
availability = (success_rate * 0.4) + (latency_score * 0.3) + (consistency_score * 0.3)
# คำนวณ Cost per Success
pricing = self.MODEL_PRICING.get(model, {"input": 10, "output": 10})
# ประมาณการ tokens ต่อ request
avg_tokens_per_request = 300 # input + output
if success_count > 0:
cost_per_1k_success = (
(pricing["input"] + pricing["output"]) * avg_tokens_per_request
* 1000 / 1_000_000
)
else:
cost_per_1k_success = float('inf')
result = BenchmarkResult(
model=model,
total_requests=requests,
successful=success_count,
failed=failure_count,
min_latency=min_lat,
max_latency=max_lat,
avg_latency=avg_lat,
p50_latency=p50,
p95_latency=p95,
p99_latency=p99,
std_latency=std_lat,
success_rate=success_rate,
failure_rate=failure_rate,
availability_score=availability,
cost_per_1k_success=cost_per_1k_success,
effective_cost=cost_per_1k_success / (success_rate / 100) if success_rate > 0 else float('inf'),
error_types=error_types
)
self.results[model] = result
return result
async def run_full_benchmark(self) -> Dict[str, BenchmarkResult]:
"""Run Benchmark สำหรับ Models ทั้งหมด"""
models = list(self.MODEL_PRICING.keys())
print("=" * 60)
print("Starting Comprehensive AI Model Benchmark")
print(f"Models: {', '.join(models)}")
print("=" * 60)
# Benchmark ทุก Model พร้อมกัน
tasks = [self.benchmark_model(model) for model in models]
await asyncio.gather(*tasks)
# แสดงผลเปรียบเทียบ
print("\n" + "=" * 60)
print("BENCHMARK RESULTS SUMMARY")
print("=" * 60)
for model in models:
print(self.results[model])
# คำแนะนำการเลือก Model
print("\n" + "=" * 60)
print("RECOMMENDATIONS")
print("=" * 60)
sorted_by_reliability = sorted(
self.results.items(),
key=lambda x: x[1].availability_score,
reverse=True
)
sorted_by_cost = sorted(
self.results.items(),
key=lambda x: x[1].effective_cost
)
print("\n🏆 Best Reliability:")
for i, (model, result) in enumerate(sorted_by_reliability[:3], 1):
print(f" {i}. {model}: {result.availability_score:.1f}/100")
print("\n💰 Most Cost-Effective:")
for i, (model, result) in enumerate(sorted_by_cost[:3], 1):
print(f" {i}. {model}: ${result.effective_cost:.4f}/1K successful")
return self.results
ตัวอย่างการรัน Benchmark
async def main():
benchmark = ModelReliability