ในโลกของ AI-powered application ทุกอย่างดูราบรื่นในขณะที่ prototype อยู่บน local machine แต่เมื่อขึ้น production เต็มรูปแบบ ความจริงที่โหดร้ายรอคุณอยู่: AI API สามารถล่มได้ตลอดเวลา ความหน่วง (latency) อาจพุ่งสูงถึง 30 วินาที หรือ token cost อาจบวมจนทำให้ project ขาดทุนในเดือนเดียว บทความนี้จะแบ่งปันแนวทางปฏิบัติจริงจากประสบการณ์ในการสร้าง AI service layer ที่ทำงานได้อย่างมั่นใจ 24/7 ครับ
ทำไมต้อง Graceful Degradation?
จากสถิติของระบบที่ผมดูแลมา พบว่า AI API มี uptime ประมาณ 99.5% เท่านั้น ฟังดูเยอะใช่มั้ยครับ? แต่ลองคำนวณดู — 0.5% ของ 1 วัน = 7.2 นาที ของ downtime ต่อวัน หรือประมาณ 2.5 ชั่วโมงต่อเดือน ถ้าระบบของคุณมี 1,000 requests ต่อวัน นั่นหมายถึง 5 requests ที่ล้มเหลวทุกวันโดยไม่มีการเตรียมรับมือ
สำหรับใครที่กำลังมองหา AI API ที่เสถียรและประหยัด ผมแนะนำ สมัครที่นี่ — ผู้ให้บริการ AI API ราคาประหยัด อัตราแลกเปลี่ยน ¥1=$1 ทำให้ประหยัดได้มากกว่า 85% เมื่อเทียบกับผู้ให้บริการอื่น แถมมี latency ต่ำกว่า 50ms อีกด้วยครับ
สถาปัตยกรรม AI Service Layer พร้อม Graceful Degradation
"""
AI Service Layer with Graceful Degradation
Production-ready implementation using HolySheep AI
"""
import asyncio
import time
import logging
from typing import Optional, Any, Dict, List
from dataclasses import dataclass, field
from enum import Enum
from collections import OrderedDict
import hashlib
Third-party imports
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ServiceHealth(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
UNHEALTHY = "unhealthy"
CIRCUIT_OPEN = "circuit_open"
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject requests
HALF_OPEN = "half_open" # Testing if service recovered
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Failures before opening circuit
success_threshold: int = 2 # Successes to close circuit
timeout: float = 30.0 # Seconds before half-open
half_open_max_calls: int = 3 # Max test calls in half-open
@dataclass
class RateLimitConfig:
requests_per_minute: int = 60
tokens_per_minute: int = 100000
burst_size: int = 10
@dataclass
class AIRequest:
model: str
messages: List[Dict[str, str]]
temperature: float = 0.7
max_tokens: int = 1000
fallback_model: Optional[str] = None
@dataclass
class AIResponse:
content: str
model: str
tokens_used: int
latency_ms: float
source: str # "primary", "fallback", "cache", "degraded"
class TokenBucket:
"""Token bucket for rate limiting with fine-grained control"""
def __init__(self, rate: float, capacity: float):
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last_update = time.time()
self._lock = asyncio.Lock()
async def acquire(self, tokens: float = 1.0) -> bool:
async with self._lock:
now = time.time()
elapsed = now - self.last_update
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_update = now
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def refill_rate(self, tokens: float) -> float:
"""Calculate time until enough tokens are available"""
needed = tokens - self.tokens
if needed <= 0:
return 0.0
return needed / self.rate
class CircuitBreaker:
"""Circuit breaker implementation for AI API resilience"""
def __init__(self, config: CircuitBreakerConfig):
self.config = config
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: Optional[float] = None
self.half_open_calls = 0
self._lock = asyncio.Lock()
async def call(self, func, *args, **kwargs):
async with self._lock:
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
logger.info("Circuit breaker entering HALF_OPEN state")
else:
raise CircuitOpenError(
f"Circuit breaker is OPEN. Retry after "
f"{self.config.timeout - (time.time() - self.last_failure_time):.1f}s"
)
if self.state == CircuitState.HALF_OPEN:
if self.half_open_calls >= self.config.half_open_max_calls:
raise CircuitOpenError("Circuit breaker half-open limit reached")
self.half_open_calls += 1
try:
result = await func(*args, **kwargs)
await self._on_success()
return result
except Exception as e:
await self._on_failure()
raise
def _should_attempt_reset(self) -> bool:
if self.last_failure_time is None:
return True
return time.time() - self.last_failure_time >= self.config.timeout
async def _on_success(self):
async with self._lock:
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self.state = CircuitState.CLOSED
self.success_count = 0
logger.info("Circuit breaker CLOSED - Service recovered")
async def _on_failure(self):
async with self._lock:
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
logger.warning("Circuit breaker OPENED - Service still failing")
elif self.failure_count >= self.config.failure_threshold:
self.state = CircuitState.OPEN
logger.warning(
f"Circuit breaker OPENED after {self.failure_count} failures"
)
class CircuitOpenError(Exception):
pass
class AICache:
"""Simple LRU cache for AI responses with TTL"""
def __init__(self, max_size: int = 1000, ttl_seconds: float = 3600):
self.cache: OrderedDict = OrderedDict()
self.max_size = max_size
self.ttl = ttl_seconds
self.hits = 0
self.misses = 0
self._lock = asyncio.Lock()
def _make_key(self, request: AIRequest) -> str:
content = f"{request.model}:{request.messages}:{request.temperature}"
return hashlib.sha256(content.encode()).hexdigest()[:32]
async def get(self, request: AIRequest) -> Optional[str]:
key = self._make_key(request)
async with self._lock:
if key in self.cache:
cached_data, timestamp = self.cache[key]
if time.time() - timestamp < self.ttl:
self.cache.move_to_end(key)
self.hits += 1
return cached_data
else:
del self.cache[key]
self.misses += 1
return None
async def set(self, request: AIRequest, response: str):
key = self._make_key(request)
async with self._lock:
if key in self.cache:
self.cache.move_to_end(key)
self.cache[key] = (response, time.time())
if len(self.cache) > self.max_size:
self.cache.popitem(last=False)
def stats(self) -> Dict[str, Any]:
total = self.hits + self.misses
hit_rate = self.hits / total if total > 0 else 0
return {
"hits": self.hits,
"misses": self.misses,
"hit_rate": f"{hit_rate:.2%}",
"size": len(self.cache)
}
class AIServiceLayer:
"""
Production AI Service Layer with Graceful Degradation
Features:
- Circuit breaker for fault isolation
- Multi-tier fallback (cache -> fallback model -> degraded response)
- Token bucket rate limiting
- Response caching
- Cost optimization through model routing
"""
BASE_URL = "https://api.holysheep.ai/v1"
# Model pricing (per 1M tokens) for cost optimization
MODEL_COSTS = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42,
}
# Latency characteristics (typical p50 in ms)
MODEL_LATENCY = {
"gpt-4.1": 800,
"claude-sonnet-4.5": 950,
"gemini-2.5-flash": 150,
"deepseek-v3.2": 200,
}
def __init__(
self,
api_key: str,
primary_model: str = "deepseek-v3.2",
circuit_breaker_config: Optional[CircuitBreakerConfig] = None,
rate_limit_config: Optional[RateLimitConfig] = None,
):
self.api_key = api_key
self.primary_model = primary_model
self.circuit_breaker = CircuitBreaker(
circuit_breaker_config or CircuitBreakerConfig()
)
self.rate_limiter = TokenBucket(
rate=rate_limit_config.requests_per_minute / 60.0,
capacity=rate_limit_config.burst_size or 10
)
self.cache = AICache(max_size=500, ttl_seconds=1800)
self.fallback_models = ["gemini-2.5-flash", "deepseek-v3.2"]
# Concurrency control
self._semaphore = asyncio.Semaphore(20) # Max concurrent requests
self._active_requests = 0
# Metrics
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"cache_hits": 0,
"fallback_usage": 0,
"circuit_breaker_trips": 0,
}
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 1000,
enable_cache: bool = True,
) -> AIResponse:
"""
Main entry point for AI chat completion with graceful degradation
"""
model = model or self.primary_model
self.metrics["total_requests"] += 1
request = AIRequest(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
fallback_model=self._get_fallback_model(model),
)
start_time = time.time()
# Try cache first
if enable_cache:
cached = await self.cache.get(request)
if cached:
self.metrics["cache_hits"] += 1
return AIResponse(
content=cached,
model=model,
tokens_used=0,
latency_ms=0,
source="cache",
)
# Rate limiting check
if not await self.rate_limiter.acquire():
logger.warning("Rate limit hit, queuing request")
wait_time = self.rate_limiter.refill_rate(1.0)
await asyncio.sleep(min(wait_time, 5.0)) # Max 5s wait
# Concurrency control
async with self._semaphore:
self._active_requests += 1
try:
# Try primary model with circuit breaker
try:
response = await self._call_with_retry(request)
self.metrics["successful_requests"] += 1
# Cache successful response
if enable_cache:
await self.cache.set(request, response.content)
return response
except (CircuitOpenError, httpx.HTTPStatusError) as e:
logger.warning(f"Primary model failed: {e}")
self.metrics["failed_requests"] += 1
# Fallback to alternative model
return await self._fallback_request(request)
finally:
self._active_requests -= 1
async def _call_with_retry(self, request: AIRequest) -> AIResponse:
"""Call AI API with circuit breaker and retry logic"""
async def _do_call():
start = time.time()
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{self.BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
},
json={
"model": request.model,
"messages": request.messages,
"temperature": request.temperature,
"max_tokens": request.max_tokens,
},
)
if response.status_code == 429:
raise httpx.HTTPStatusError(
"Rate limited", request=response.request, response=response
)
response.raise_for_status()
data = response.json()
latency_ms = (time.time() - start) * 1000
tokens_used = data.get("usage", {}).get("total_tokens", 0)
return AIResponse(
content=data["choices"][0]["message"]["content"],
model=request.model,
tokens_used=tokens_used,
latency_ms=latency_ms,
source="primary",
)
return await self.circuit_breaker.call(_do_call)
async def _fallback_request(self, request: AIRequest) -> AIResponse:
"""Handle fallback to alternative model or degraded response"""
for fallback_model in self.fallback_models:
if fallback_model == request.model:
continue
logger.info(f"Trying fallback model: {fallback_model}")
self.metrics["fallback_usage"] += 1
try:
fallback_request = AIRequest(
model=fallback_model,
messages=request.messages,
temperature=request.temperature,
max_tokens=request.max_tokens,
)
return await self._call_with_retry(fallback_request)
except Exception as e:
logger.warning(f"Fallback {fallback_model} failed: {e}")
continue
# All models failed, return degraded but helpful response
logger.error("All AI models failed, returning degraded response")
return await self._degraded_response(request)
async def _degraded_response(self, request: AIRequest) -> AIResponse:
"""
Return a degraded but still helpful response when all AI services fail
"""
return AIResponse(
content=(
"ขออภัยครับ ระบบ AI ขณะนี้ไม่สามารถประมวลผลได้ "
"กรุณาลองใหม่อีกครั้งในอีกไม่กี่นาที หรือติดต่อฝ่ายสนับสนุน"
),
model="none",
tokens_used=0,
latency_ms=0,
source="degraded",
)
def _get_fallback_model(self, model: str) -> Optional[str]:
"""Get the next best fallback model based on cost-latency tradeoff"""
try:
idx = self.fallback_models.index(model)
if idx + 1 < len(self.fallback_models):
return self.fallback_models[idx + 1]
except ValueError:
pass
return self.fallback_models[0] if self.fallback_models else None
def estimate_cost(self, model: str, tokens: int) -> float:
"""Estimate cost for a request in USD"""
cost_per_million = self.MODEL_COSTS.get(model, 1.0)
return (tokens / 1_000_000) * cost_per_million
def select_optimal_model(
self,
required_quality: str = "medium",
max_latency_ms: float = 500,
) -> str:
"""
Select the most cost-effective model based on requirements
Quality levels:
- high: Use best model regardless of cost
- medium: Balance cost and quality
- fast: Use fastest/cheapest model
"""
if required_quality == "high":
return "gpt-4.1"
candidates = [
m for m in self.fallback_models
if self.MODEL_LATENCY.get(m, 999) <= max_latency_ms
]
if not candidates:
candidates = self.fallback_models
# Sort by cost (ascending)
candidates.sort(key=lambda m: self.MODEL_COSTS.get(m, 999))
return candidates[0]
def get_metrics(self) -> Dict[str, Any]:
"""Get current service metrics"""
cache_stats = self.cache.stats()
return {
**self.metrics,
**cache_stats,
"active_requests": self._active_requests,
"circuit_breaker_state": self.circuit_breaker.state.value,
"estimated_monthly_cost_usd": self._estimate_monthly_cost(),
}
def _estimate_monthly_cost(self) -> float:
"""Estimate monthly cost based on current metrics"""
avg_tokens_per_request = 500 # Rough estimate
monthly_requests = self.metrics["total_requests"] * 30
total_tokens = monthly_requests * avg_tokens_per_request
# Use cheapest model price as baseline
min_cost = min(self.MODEL_COSTS.values())
return (total_tokens / 1_000_000) * min_cost
Example usage
async def main():
"""Example demonstrating the AI service layer"""
service = AIServiceLayer(
api_key="YOUR_HOLYSHEEP_API_KEY",
primary_model="deepseek-v3.2", # Most cost-effective
)
messages = [
{"role": "system", "content": "คุณเป็นผู้ช่วย AI ที่เป็นมิตร"},
{"role": "user", "content": "อธิบายเรื่อง Graceful Degradation ให้เข้าใจง่าย"},
]
# Select optimal model based on requirements
model = service.select_optimal_model(
required_quality="medium",
max_latency_ms=300,
)
print(f"Selected model: {model}")
# Make request
response = await service.chat_completion(
messages=messages,
model=model,
temperature=0.7,
)
print(f"Response from: {response.source}")
print(f"Latency: {response.latency_ms:.2f}ms")
print(f"Content: {response.content[:100]}...")
# Check metrics
print(f"\nMetrics: {service.get_metrics()}")
if __name__ == "__main__":
asyncio.run(main())
กลยุทธ์ Fallback หลายระดับ (Multi-Tier Fallback)
จากโค้ดข้างต้น ระบบมีการจัดการ fallback เป็นลำดับชั้นดังนี้ครับ:
- Cache Hit — ถ้าคำถามเดิมเคยถามมาก่อน ตอบจาก cache ทันที (latency ~0ms, ค่าใช้จ่าย $0)
- Primary Model — ลอง model หลักก่อน พร้อม circuit breaker ป้องกัน
- Fallback Model — ถ้า primary ล้มเหลว ลอง model ทดแทนตามลำดับ
- Degraded Response — ถ้าทุกอย่างล้มเหลว ส่งข้อความแจ้งผู้ใช้อย่างสุภาพ
"""
Benchmark Script - วัดประสิทธิภาพ Graceful Degradation
Run this script to test system behavior under various failure scenarios
"""
import asyncio
import time
import statistics
from typing import List, Tuple
import sys
sys.path.append('.')
from ai_service_layer import AIServiceLayer, CircuitBreakerConfig
async def benchmark_cache_performance(service: AIServiceLayer):
"""Benchmark cache hit rate with repeated queries"""
print("\n" + "="*60)
print("BENCHMARK 1: Cache Performance")
print("="*60)
messages = [
{"role": "user", "content": "วิธีทำกาแฟเย็นแบบง่ายๆ"}
]
# First request - cache miss expected
start = time.time()
response1 = await service.chat_completion(messages, enable_cache=True)
time1 = (time.time() - start) * 1000
# Subsequent requests - should hit cache
times = []
for i in range(10):
await asyncio.sleep(0.1)
start = time.time()
response = await service.chat_completion(messages, enable_cache=True)
times.append((time.time() - start) * 1000)
avg_cache_time = statistics.mean(times)
print(f"First request (cache miss): {time1:.2f}ms")
print(f"Subsequent requests (cache hit) avg: {avg_cache_time:.2f}ms")
print(f"Cache speedup: {time1/avg_cache_time:.1f}x faster")
metrics = service.get_metrics()
print(f"Cache hit rate: {metrics['hit_rate']}")
async def simulate_primary_failure(service: AIServiceLayer):
"""Simulate primary model failure and measure fallback behavior"""
print("\n" + "="*60)
print("BENCHMARK 2: Fallback Behavior (Simulated)")
print("="*60)
# Temporarily break the primary model
original_model = service.primary_model
service.primary_model = "non-existent-model"
messages = [
{"role": "user", "content": "ทดสอบระบบ fallback"}
]
start = time.time()
response = await service.chat_completion(messages)
elapsed = (time.time() - start) * 1000
print(f"Fallback response source: {response.source}")
print(f"Total time (with fallback): {elapsed:.2f}ms")
# Restore
service.primary_model = original_model
async def benchmark_concurrent_load(service: AIServiceLayer):
"""Benchmark system behavior under concurrent load"""
print("\n" + "="*60)
print("BENCHMARK 3: Concurrent Load Test")
print("="*60)
messages = [
{"role": "user", "content": f"ทดสอบ concurrent request ที่ {i}"}
for i in range(50)
]
start = time.time()
tasks = [
service.chat_completion([msg])
for msg in messages
]
responses = await asyncio.gather(*tasks, return_exceptions=True)
elapsed = time.time() - start
success_count = sum(1 for r in responses if not isinstance(r, Exception))
print(f"Total requests: 50")
print(f"Successful: {success_count}")
print(f"Failed: {50 - success_count}")
print(f"Total time: {elapsed:.2f}s")
print(f"Throughput: {50/elapsed:.1f} req/s")
print(f"Avg latency: {elapsed/50*1000:.2f}ms per request")
async def benchmark_cost_estimation(service: AIServiceLayer):
"""Benchmark cost estimation accuracy"""
print("\n" + "="*60)
print("BENCHMARK 4: Cost Estimation")
print("="*60)
test_cases = [
("deepseek-v3.2", 1000),
("gpt-4.1", 1000),
("gemini-2.5-flash", 1000),
("claude-sonnet-4.5", 1000),
]
print(f"{'Model':<25} {'Tokens':<10} {'Est. Cost':<12} {'$/M Token'}")
print("-" * 60)
for model, tokens in test_cases:
cost = service.estimate_cost(model, tokens)
per_million = service.MODEL_COSTS.get(model, 0)
print(f"{model:<25} {tokens:<10} ${cost:<11.4f} ${per_million}")
# Calculate savings
expensive = service.estimate_cost("claude-sonnet-4.5", 1000000)
cheap = service.estimate_cost("deepseek-v3.2", 1000000)
savings = ((expensive - cheap) / expensive) * 100
print(f"\n💰 Cost savings with DeepSeek V3.2 vs Claude: {savings:.1f}%")
async def benchmark_model_selection():
"""Benchmark model selection algorithm"""
print("\n" + "="*60)
print("BENCHMARK 5: Model Selection Strategy")
print("="*60)
service = AIServiceLayer(api_key="test")
scenarios = [
("Fast response needed", "fast", 200),
("Balance quality/speed", "medium", 500),
("Best quality needed", "high", 2000),
]
print(f"{'Scenario':<25} {'Quality':<10} {'Max Latency':<15} {'Selected Model'}")
print("-" * 75)
for desc, quality, max_latency in scenarios:
model = service.select_optimal_model(quality, max_latency)
cost = service.MODEL_COSTS.get(model, 0)
latency = service.MODEL_LATENCY.get(model, 0)
print(f"{desc:<25} {quality:<10} {max_latency}ms{'':<8} {model:<20} (${cost}/M, ~{latency}ms)")
async def run_all_benchmarks():
"""Run all benchmarks"""
print("\n" + "="*60)
print("🚀 AI SERVICE LAYER - BENCHMARK SUITE")
print("="*60)
print("Testing Graceful Degradation Implementation")
print("Using HolySheep AI: https://api.holysheep.ai/v1")
service = AIServiceLayer(
api_key="YOUR_HOLYSHEEP_API_KEY",
primary_model="deepseek-v3.2",
circuit_breaker_config=CircuitBreakerConfig(
failure_threshold=3,
timeout=10.0,
),
)
# Run benchmarks (skip actual API calls for model selection)
await benchmark_cache_performance(service)
await simulate_primary_failure(service)
await benchmark
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง