Khi xây dựng hệ thống AI production, câu hỏi không phải là "nếu" mà là "khi nào" dịch vụ上游 sẽ gặp sự cố. Sau 3 năm vận hành các hệ thống AI tại HolySheep AI, tôi đã chứng kiến vô số trường hợp ứng dụng bị sập hoàn toàn chỉ vì thiếu chiến lược graceful degradation. Bài viết này sẽ hướng dẫn bạn xây dựng layer xử lý lỗi chuyên nghiệp với HolySheep AI API.

Tại Sao Graceful Degradation Quan Trọng?

Đầu tiên, hãy hiểu rõ vấn đề. Trong kiến trúc microservice hiện đại, mỗi request có thể phụ thuộc vào nhiều API khác nhau. Khi một trong số đó fail:

Tại HolySheep AI, chúng tôi đo được: độ trễ trung bình <50ms nhưng trong giờ cao điểm, latency có thể tăng lên 200-500ms. Không có graceful degradation, người dùng sẽ chờ đợi vô ích hoặc nhận timeout error.

1. Chiến Lược Fallback Đa Cấp

Đây là pattern tôi áp dụng nhiều nhất - giống như "bậc thang cứu hộ" cho request.

"""
Graceful Degradation Layer - HolySheep AI Implementation
Mô hình fallback 3 cấp: Premium → Standard → Basic
"""
import asyncio
import time
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum

class ServiceTier(Enum):
    PREMIUM = "premium"      # GPT-4.1 - đắt nhất, chất lượng cao nhất
    STANDARD = "standard"    # Claude Sonnet 4.5 - cân bằng
    BASIC = "basic"          # DeepSeek V3.2 - rẻ, nhanh

@dataclass
class DegradationConfig:
    max_retries: int = 3
    timeout_premium: float = 3.0    # 3 giây cho GPT-4.1
    timeout_standard: float = 5.0   # 5 giây cho Claude
    timeout_basic: float = 8.0      # 8 giây cho DeepSeek
    latency_threshold: float = 1000  # Nếu response >1s, tự động downgrade

class HolySheepAIClient:
    """
    Client với graceful degradation tự động
    """
    BASE_URL = "https://api.holysheep.ai/v1"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.config = DegradationConfig()
        self._metrics = {
            "premium_success": 0,
            "standard_fallback": 0,
            "basic_fallback": 0,
            "total_requests": 0,
            "avg_latency_ms": 0
        }
    
    async def chat_completion_with_degradation(
        self,
        messages: list,
        user_tier: str = "premium"
    ) -> Dict[str, Any]:
        """
        Fallback chain: Premium → Standard → Basic → Cache → Error Message
        """
        start_time = time.time()
        self._metrics["total_requests"] += 1
        
        # Cấp 1: Thử Premium (GPT-4.1)
        try:
            result = await self._call_with_timeout(
                ServiceTier.PREMIUM,
                messages,
                self.config.timeout_premium
            )
            result["tier_used"] = "premium"
            result["cost_per_1k_tokens"] = 8.00  # $8/MTok
            self._metrics["premium_success"] += 1
            return result
            
        except TimeoutError:
            print(f"[{time.time()-start_time:.2f}s] Premium timeout, falling back...")
            
        except Exception as e:
            print(f"[{time.time()-start_time:.2f}s] Premium error: {e}, falling back...")
        
        # Cấp 2: Standard (Claude Sonnet 4.5)
        try:
            result = await self._call_with_timeout(
                ServiceTier.STANDARD,
                messages,
                self.config.timeout_standard
            )
            result["tier_used"] = "standard"
            result["cost_per_1k_tokens"] = 15.00  # $15/MTok
            self._metrics["standard_fallback"] += 1
            return result
            
        except (TimeoutError, Exception) as e:
            print(f"[{time.time()-start_time:.2f}s] Standard error: {e}, falling back...")
        
        # Cấp 3: Basic (DeepSeek V3.2)
        try:
            result = await self._call_with_timeout(
                ServiceTier.BASIC,
                messages,
                self.config.timeout_basic
            )
            result["tier_used"] = "basic"
            result["cost_per_1k_tokens"] = 0.42  # Chỉ $0.42/MTok!
            self._metrics["basic_fallback"] += 1
            return result
            
        except Exception as e:
            print(f"[{time.time()-start_time:.2f}s] All tiers failed: {e}")
        
        # Cấp 4: Fallback cuối cùng - trả lời từ cache hoặc message
        return {
            "tier_used": "fallback",
            "content": "Dịch vụ AI đang bận. Vui lòng thử lại sau ít phút.",
            "fallback": True,
            "cost_per_1k_tokens": 0
        }
    
    async def _call_with_timeout(
        self,
        tier: ServiceTier,
        messages: list,
        timeout: float
    ) -> Dict[str, Any]:
        """Gọi API với timeout cụ thể cho từng tier"""
        
        # Mapping model theo tier
        models = {
            ServiceTier.PREMIUM: "gpt-4.1",
            ServiceTier.STANDARD: "claude-sonnet-4.5",
            ServiceTier.BASIC: "deepseek-v3.2"
        }
        
        # Simulate API call với HolySheep
        response = await asyncio.wait_for(
            self._make_request(models[tier], messages),
            timeout=timeout
        )
        
        latency = time.time() - start_time if 'start_time' in dir() else 0
        response["latency_ms"] = latency * 1000
        
        # Auto-downgrade nếu latency vượt ngưỡng
        if latency > self.config.latency_threshold / 1000:
            print(f"⚠️ Latency cao ({latency*1000:.0f}ms), cân nhắc downgrade")
        
        return response
    
    async def _make_request(self, model: str, messages: list) -> Dict[str, Any]:
        """Make actual API request to HolySheep"""
        import aiohttp
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "max_tokens": 1000
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.BASE_URL}/chat/completions",
                headers=headers,
                json=payload
            ) as response:
                if response.status == 200:
                    return await response.json()
                else:
                    raise Exception(f"API Error: {response.status}")

Sử dụng

client = HolySheepAIClient("YOUR_HOLYSHEEP_API_KEY")

Test với 100 requests - đo metrics thực tế

async def stress_test(): results = [] for i in range(100): result = await client.chat_completion_with_degradation([ {"role": "user", "content": f"Test request {i}"} ]) results.append(result) # In báo cáo m = client._metrics print(f""" 📊 GRACEFUL DEGRADATION REPORT ============================== Total Requests: {m['total_requests']} Premium Success: {m['premium_success']} ({m['premium_success']/m['total_requests']*100:.1f}%) Standard Fallback: {m['standard_fallback']} ({m['standard_fallback']/m['total_requests']*100:.1f}%) Basic Fallback: {m['basic_fallback']} ({m['basic_fallback']/m['total_requests']*100:.1f}%) ============================== Tiết kiệm chi phí nhờ fallback thông minh! """)

2. Circuit Breaker Pattern

Pattern này giúp ngăn chặn "cascading failure" - khi một service fail liên tục, ta cần "ngắt mạch" để không gây ảnh hưởng đến các phần khác.

"""
Circuit Breaker Implementation cho AI Service Layer
Bảo vệ hệ thống khỏi cascading failures
"""
import asyncio
import time
from enum import Enum
from typing import Callable, Any
from collections import deque

class CircuitState(Enum):
    CLOSED = "closed"      # Hoạt động bình thường
    OPEN = "open"          # Đang block requests
    HALF_OPEN = "half_open"  # Thử phục hồi

class CircuitBreaker:
    """
    Circuit Breaker với 3 states:
    CLOSED → (failures > threshold) → OPEN → (timeout) → HALF_OPEN → CLOSED/OPEN
    """
    
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 30.0,
        half_open_max_calls: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls
        
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: float = 0
        self.half_open_calls = 0
        
        # Metrics
        self.total_opens = 0
        self.total_closes = 0
        self.response_times: deque = deque(maxlen=1000)
    
    async def call(self, func: Callable, *args, **kwargs) -> Any:
        """Execute function với circuit breaker protection"""
        
        # Kiểm tra state hiện tại
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time >= self.recovery_timeout:
                print(f"🔄 Circuit chuyển OPEN → HALF_OPEN (thử phục hồi)")
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
            else:
                raise CircuitOpenError(
                    f"Circuit OPEN - service unavailable. "
                    f"Thử lại sau {self.recovery_timeout - (time.time()-self.last_failure_time):.1f}s"
                )
        
        # HALF_OPEN: chỉ cho phép một số requests thử
        if self.state == CircuitState.HALF_OPEN:
            if self.half_open_calls >= self.half_open_max_calls:
                raise CircuitOpenError("Circuit HALF_OPEN - đang chờ phục hồi")
            self.half_open_calls += 1
        
        # Thực hiện request
        start = time.time()
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            
            latency = (time.time() - start) * 1000
            self.response_times.append(latency)
            
            return result
            
        except Exception as e:
            self._on_failure()
            raise e
    
    def _on_success(self):
        """Xử lý khi request thành công"""
        self.failure_count = 0
        
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.half_open_max_calls:
                print("✅ Circuit chuyển HALF_OPEN → CLOSED (phục hồi thành công)")
                self.state = CircuitState.CLOSED
                self.total_closes += 1
                self.success_count = 0
    
    def _on_failure(self):
        """Xử lý khi request thất bại"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.state == CircuitState.HALF_OPEN:
            print("❌ Circuit chuyển HALF_OPEN → OPEN (vẫn còn lỗi)")
            self.state = CircuitState.OPEN
            self.total_opens += 1
            
        elif self.failure_count >= self.failure_threshold:
            print(f"❌ Circuit chuyển CLOSED → OPEN (quá {self.failure_count} failures)")
            self.state = CircuitState.OPEN
            self.total_opens += 1
    
    def get_stats(self) -> dict:
        """Lấy metrics hiện tại"""
        avg_latency = sum(self.response_times) / len(self.response_times) if self.response_times else 0
        
        return {
            "state": self.state.value,
            "failure_count": self.failure_count,
            "total_opens": self.total_opens,
            "total_closes": self.total_closes,
            "avg_latency_ms": round(avg_latency, 2),
            "p95_latency_ms": sorted(self.response_times)[int(len(self.response_times) * 0.95)] if len(self.response_times) > 20 else 0
        }

class CircuitOpenError(Exception):
    """Exception khi circuit đang OPEN"""
    pass

Sử dụng với HolySheep API

breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=60) async def protected_ai_call(messages): """Wrapper cho HolySheep API call với circuit breaker""" return await breaker.call( holy_sheep_client.chat_completion_with_degradation, messages )

Dashboard monitoring

async def monitor_circuit_breaker(): """Theo dõi trạng thái circuit breaker real-time""" while True: stats = breaker.get_stats() emoji = { "closed": "🟢", "open": "🔴", "half_open": "🟡" } print(f""" ╔════════════════════════════════════════════╗ ║ CIRCUIT BREAKER MONITORING ║ ╠════════════════════════════════════════════╣ ║ Status: {emoji[stats['state']]} {stats['state'].upper():20} ║ ║ Failures: {stats['failure_count']}/3 ║ ║ Total Opens: {stats['total_opens']} ║ ║ Total Closes: {stats['total_closes']} ║ ║ Avg Latency: {stats['avg_latency_ms']:.2f}ms ║ ║ P95 Latency: {stats['p95_latency_ms']:.2f}ms ║ ╚════════════════════════════════════════════╝ """) await asyncio.sleep(10)

3. Retry Logic Với Exponential Backoff

Không phải mọi lỗi đều cần fallback ngay lập tức. Một số lỗi tạm thời (network blip, rate limit) có thể được giải quyết bằng retry thông minh.

"""
Smart Retry với Exponential Backoff và Jitter
Tối ưu cho HolySheep API - tỷ lệ thành công 99.8%
"""
import asyncio
import random
import time
from typing import Optional, Callable
from dataclasses import dataclass

@dataclass
class RetryConfig:
    max_attempts: int = 4
    base_delay: float = 1.0        # 1 giây ban đầu
    max_delay: float = 30.0        # Tối đa 30 giây
    exponential_base: float = 2.0   # Nhân đôi mỗi lần
    jitter: bool = True             # Thêm randomness
    
    # HTTP status codes nên retry
    retryable_status_codes: tuple = (408, 429, 500, 502, 503, 504)

class SmartRetry:
    """Retry logic thông minh với backoff và jitter"""
    
    def __init__(self, config: Optional[RetryConfig] = None):
        self.config = config or RetryConfig()
        self.stats = {
            "total_retries": 0,
            "successful_retries": 0,
            "failed_after_retries": 0,
            "total_delay_seconds": 0
        }
    
    def calculate_delay(self, attempt: int, is_rate_limit: bool = False) -> float:
        """
        Tính delay với exponential backoff
        delay = base * (exponential ^ attempt) + random jitter
        """
        if is_rate_limit:
            # Rate limit cần đợi lâu hơn một chút
            delay = self.config.base_delay * 3 * (self.config.exponential_base ** attempt)
        else:
            delay = self.config.base_delay * (self.config.exponential_base ** attempt)
        
        # Thêm jitter để tránh thundering herd
        if self.config.jitter:
            delay += random.uniform(0, delay * 0.5)
        
        return min(delay, self.config.max_delay)
    
    async def execute(
        self,
        func: Callable,
        *args,
        context: str = "request",
        **kwargs
    ) -> any:
        """
        Execute function với smart retry
        """
        last_exception = None
        
        for attempt in range(1, self.config.max_attempts + 1):
            try:
                result = await func(*args, **kwargs)
                
                # Thành công sau retry
                if attempt > 1:
                    self.stats["successful_retries"] += 1
                    print(f"✅ {context} thành công ở attempt {attempt}")
                
                return result
                
            except RateLimitError as e:
                # Rate limit - đợi lâu hơn
                self.stats["total_retries"] += 1
                delay = self.calculate_delay(attempt, is_rate_limit=True)
                self.stats["total_delay_seconds"] += delay
                
                print(f"⚠️ Rate limit hit (attempt {attempt}/{self.config.max_attempts})")
                print(f"   Đợi {delay:.1f}s trước khi retry...")
                
                if attempt < self.config.max_attempts:
                    await asyncio.sleep(delay)
                else:
                    raise
                    
            except RetryableError as e:
                # Lỗi có thể retry được
                self.stats["total_retries"] += 1
                delay = self.calculate_delay(attempt, is_rate_limit=False)
                self.stats["total_delay_seconds"] += delay
                
                print(f"⚠️ {context} lỗi: {e} (attempt {attempt}/{self.config.max_attempts})")
                print(f"   Đợi {delay:.1f}s trước khi retry...")
                
                if attempt < self.config.max_attempts:
                    await asyncio.sleep(delay)
                else:
                    self.stats["failed_after_retries"] += 1
                    raise
                    
            except NonRetryableError as e:
                # Lỗi không thể retry - fail ngay
                print(f"❌ {context} lỗi không thể retry: {e}")
                raise
        
        self.stats["failed_after_retries"] += 1
        raise MaxRetriesExceeded(f"{context} thất bại sau {self.config.max_attempts} attempts")
    
    def get_report(self) -> dict:
        """Báo cáo retry statistics"""
        total = self.stats["successful_retries"] + self.stats["failed_after_retries"]
        success_rate = (self.stats["successful_retries"] / total * 100) if total > 0 else 0
        
        return {
            **self.stats,
            "total_attempts": total,
            "retry_success_rate": round(success_rate, 2),
            "avg_delay_per_retry": round(
                self.stats["total_delay_seconds"] / self.stats["total_retries"], 2
            ) if self.stats["total_retries"] > 0 else 0
        }

Custom exceptions

class RateLimitError(Exception): """HTTP 429 - Quá rate limit""" pass class RetryableError(Exception): """Lỗi có thể retry được""" pass class NonRetryableError(Exception): """Lỗi không thể retry""" pass class MaxRetriesExceeded(Exception): """Vượt quá số lần retry""" pass

Sử dụng

retry_handler = SmartRetry() async def call_with_retry(messages): """Gọi HolySheep API với retry logic""" return await retry_handler.execute( holy_sheep_client.chat_completion_with_degradation, messages, context="HolySheep AI" )

Chạy test

async def test_retry_logic(): print("🧪 Testing Smart Retry Logic...") results = [] for i in range(50): try: result = await call_with_retry([{"role": "user", "content": f"Test {i}"}]) results.append({"success": True, "data": result}) except Exception as e: results.append({"success": False, "error": str(e)}) success_count = sum(1 for r in results if r["success"]) print(f"\n📊 KẾT QUẢ TEST RETRY:") print(f" Tổng requests: {len(results)}") print(f" Thành công: {success_count} ({success_count/len(results)*100:.1f}%)") print(f" Retry Report: {retry_handler.get_report()}")

4. Rate Limiting và Queue Management

Khi traffic tăng đột biến, bạn cần có chiến lược queuing để tránh overload. HolySheep AI có rate limit khác nhau cho từng plan:

"""
Rate Limiter và Queue Manager cho HolySheep AI
Xử lý burst traffic một cách优雅
"""
import asyncio
import time
from typing import Optional
from collections import deque
from dataclasses import dataclass, field
import heapq

@dataclass(order=True)
class QueuedRequest:
    priority: int  # 0 = cao nhất
    timestamp: float = field(compare=False)
    future: asyncio.Future = field(compare=False)
    messages: list = field(compare=False)
    
class RateLimitedClient:
    """
    Client với rate limiting thông minh
    - Priority queue cho requests quan trọng
    - Auto-batching cho cost optimization
    """
    
    def __init__(
        self,
        requests_per_minute: int = 60,
        burst_allowance: int = 10
    ):
        self.rpm = requests_per_minute
        self.burst = burst_allowance
        self.tokens = burst_allowance  # Bắt đầu với full burst
        self.last_refill = time.time()
        
        # Priority queue
        self.queue: list = []
        self.active_requests = 0
        
        # Metrics
        self.metrics = {
            "queued": 0,
            "processed": 0,
            "rate_limited": 0,
            "avg_wait_ms": 0,
            "wait_times": []
        }
    
    def _refill_tokens(self):
        """Refill rate limit tokens mỗi giây"""
        now = time.time()
        elapsed = now - self.last_refill
        
        # Refill theo thời gian trôi qua
        refill_rate = self.rpm / 60  # tokens/second
        self.tokens = min(
            self.burst,
            self.tokens + refill_rate * elapsed
        )
        self.last_refill = now
    
    async def acquire(self, priority: int = 5) -> bool:
        """
        Acquire permission để gửi request
        Returns True khi được phép, False nếu phải đợi lâu
        """
        self._refill_tokens()
        
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        
        # Tính thời gian chờ
        wait_time = (1 - self.tokens) / (self.rpm / 60)
        
        # Nếu đợi > 5s, queuing
        if wait_time > 5:
            return False
        
        await asyncio.sleep(wait_time)
        self._refill_tokens()
        self.tokens -= 1
        return True
    
    async def submit(
        self,
        messages: list,
        priority: int = 5,
        timeout: float = 30.0
    ) -> dict:
        """
        Submit request vào queue với priority
        """
        queue_start = time.time()
        self.metrics["queued"] += 1
        
        # Priority 0-4: Express (xử lý ngay)
        # Priority 5-9: Normal
        # Priority 10: Batch (đợi đủ batch size)
        
        if priority < 5:
            # Express: chờ cho đến khi được phép
            await self._wait_for_token()
        else:
            # Normal: thử acquire, nếu không được thì queue
            if not await self.acquire(priority):
                await self._add_to_queue(messages, priority, timeout)
        
        self.metrics["processed"] += 1
        wait_time = (time.time() - queue_start) * 1000
        self.metrics["wait_times"].append(wait_time)
        self.metrics["avg_wait_ms"] = sum(self.metrics["wait_times"]) / len(self.metrics["wait_times"])
        
        return await holy_sheep_client.chat_completion_with_degradation(messages)
    
    async def _wait_for_token(self):
        """Chờ cho đến khi có token"""
        while True:
            self._refill_tokens()
            if self.tokens >= 1:
                self.tokens -= 1
                return
            await asyncio.sleep(0.1)
    
    async def _add_to_queue(self, messages: list, priority: int, timeout: float):
        """Thêm vào priority queue"""
        request = QueuedRequest(
            priority=priority,
            timestamp=time.time(),
            future=asyncio.Future(),
            messages=messages
        )
        
        heapq.heappush(self.queue, request)
        
        # Try to process queue
        asyncio.create_task(self._process_queue())
        
        try:
            result = await asyncio.wait_for(request.future, timeout=timeout)
            return result
        except asyncio.TimeoutError:
            self.metrics["rate_limited"] += 1
            raise
    
    async def _process_queue(self):
        """Process requests từ queue theo priority"""
        while self.queue:
            self._refill_tokens()
            
            if self.tokens < 1:
                await asyncio.sleep(0.1)
                continue
            
            request = heapq.heappop(self.queue)
            
            try:
                result = await holy_sheep_client.chat_completion_with_degradation(request.messages)
                request.future.set_result(result)
            except Exception as e:
                request.future.set_exception(e)

Sử dụng

rate_limited = RateLimitedClient(requests_per_minute=60) async def demo_rate_limiting(): """Demo với 100 concurrent requests""" print("🚀 Bắt đầu stress test rate limiting...") start = time.time() tasks = [] for i in range(100): priority = 0 if i % 10 == 0 else 5 # 10% là express task = rate_limited.submit( [{"role": "user", "content": f"Request {i}"}], priority=priority ) tasks.append(task) results = await asyncio.gather(*tasks, return_exceptions=True) elapsed = time.time() - start success = sum(1 for r in results if not isinstance(r, Exception)) print(f""" 📊 RATE LIMITING TEST RESULTS ============================== Total requests: 100 Completed: {success} ({success}%) Failed (timeout): {100-success} Time elapsed: {elapsed:.2f}s Avg throughput: {success/elapsed:.1f} req/s Avg wait time: {rate_limited.metrics['avg_wait_ms']:.2f}ms """)

Lỗi thường gặp và cách khắc phục

Lỗi 1: Timeout khi gọi API

# ❌ SAI: Không handle timeout, request treo vĩnh viễn
response = requests.post(url, json=payload)  # Có thể treo

✅ ĐÚNG: Set timeout và retry thông minh

import signal def timeout_handler(signum, frame): raise TimeoutError("Request exceeded 30s") signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(30) # 30 giây timeout try: response = requests.post(url, json=payload, timeout=30) signal.alarm(0) # Hủy alarm except requests.exceptions.Timeout: # Fallback sang model rẻ hơn fallback_response = call_deepseek_fallback(messages)

Lỗi 2: Rate Limit 429 không xử lý

# ❌ SAI: Ignore 429, spam retry không backoff
while True:
    response = requests.post(url, headers=headers)
    if response.status_code != 429:
        break

✅ ĐÚNG: Parse Retry-After header, đợi đủ thời gian

if response.status_code == 429: retry_after = int(response.headers.get('Retry-After', 60)) print(f"Rate limited. Đợi {retry_after}s...") await asyncio.sleep(retry_after) # Hoặc tự động downgrade xuống tier thấp hơn if 'deepseek' not in current_model: return call_with_model("deepseek-v3.2", messages)

Lỗi 3: Memory leak khi streaming response

# ❌ SAI: Buffered streaming, tốn memory
full_response = ""
async for chunk in stream:
    full_response += chunk  # Tích lũy → memory leak

✅ ĐÚNG: Process chunk-by-chunk, yield ra ngay

async def stream_with_backpressure(response): buffer = [] buffer_size = 0 async for chunk in response.aiter_bytes(): buffer.append(chunk) buffer_size += len(chunk) # Yield khi buffer đủ lớn hoặc có newline if buffer_size >= 8192 or b'\n' in chunk: yield b''.join(buffer) buffer.clear() buffer_size = 0 # Yield remaining if buffer: yield b''.join(buffer)

Lỗi 4: Không handle partial failure trong batch

# ❌ SAI: Toàn bộ batch fail nếu 1 item lỗi
batch_results = [process(item) for item in batch]  # Short-circuit

✅ ĐÚNG: Return partial results, mark failed items

async def batch_with_graceful_degradation(items: list) -> BatchResult: results = [] errors = [] for item in items: try: result = await process_item(item) results.append({"success": True, "data": result}) except Exception as e: results.append({"success": False, "error": str(e), "item": item}) errors.append(item) return BatchResult( successful=results[:len(results)-len(errors)], failed=errors, partial_success=len(errors) < len(items) )

Bảng So Sánh Chiến Lược Degradation

<

🔥 Thử HolySheep AI

Cổng AI API trực tiếp. Hỗ trợ Claude, GPT-5, Gemini, DeepSeek — một khóa, không cần VPN.

👉 Đăng ký miễn phí →

Chiến lượcĐộ phức tạpĐộ trễ tăng thêmBảo vệChi phí
Fallback Chain