AI Agent การจัดการ Timeout และการออกแบบ Graceful Degradation อย่างมืออาชีพ

ในโลกของการพัฒนา AI Agent เรื่อง Timeout และ Graceful Degradation เป็นสิ่งที่หลีกเลี่ยงไม่ได้เลย วันนี้ผมจะมาแชร์ประสบการณ์ตรงในการจัดการปัญหาเหล่านี้ด้วย HolySheep AI ที่ช่วยประหยัดค่าใช้จ่ายได้ถึง 85%+

สรุปคำตอบแบบรวบรัด

สำหรับคนที่ไม่มีเวลาอ่านทั้งหมด นี่คือสรุปวิธีแก้ปัญหา Timeout ใน AI Agent:

• ใช้ Exponential Backoff — รอนานขึ้นเรื่อยๆ เมื่อ retry หลายครั้ง
• ตั้งค่า Timeout ที่เหมาะสม — 30-60 วินาทีสำหรับงานทั่วไป
• ออกแบบ Fallback Strategy — เตรียมทางเลือกสำรองเสมอ
• ใช้ Circuit Breaker Pattern — ป้องกันระบบล่มเมื่อ API มีปัญหาต่อเนื่อง

ตารางเปรียบเทียบราคาและคุณสมบัติ API สำหรับ AI Agent

พื้นฐานการจัดการ Timeout ใน AI Agent

ก่อนจะไปถึงโค้ดขั้นสูง มาทำความเข้าใจพื้นฐานกันก่อน Timeout ใน AI Agent แบ่งเป็น 3 ประเภทหลัก:

• Connection Timeout — ระยะเวลารอเชื่อมต่อกับ Server
• Read Timeout — ระยะเวลารอรับ Response หลังส่ง Request แล้ว
• Total Timeout — ระยะเวลารวมทั้งหมด

โค้ดตัวอย่าง: AI Agent Timeout Handler พื้นฐาน

import requests
import time
from typing import Optional, Dict, Any, Callable
from dataclasses import dataclass
from enum import Enum

class TimeoutStrategy(Enum):
    IMMEDIATE_RETRY = "immediate_retry"
    EXPONENTIAL_BACKOFF = "exponential_backoff"
    GRACEFUL_DEGRADATION = "graceful_degradation"

@dataclass
class TimeoutConfig:
    connect_timeout: float = 10.0
    read_timeout: float = 30.0
    max_retries: int = 3
    strategy: TimeoutStrategy = TimeoutStrategy.EXPONENTIAL_BACKOFF
    fallback_enabled: bool = True

class HolySheepAIAgent:
    """AI Agent พร้อมระบบ Timeout และ Graceful Degradation"""
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",
        config: Optional[TimeoutConfig] = None
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.config = config or TimeoutConfig()
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        })
    
    def chat_with_timeout(
        self,
        messages: list,
        model: str = "gpt-4.1",
        system_prompt: Optional[str] = None
    ) -> Dict[str, Any]:
        """ส่งข้อความไปยัง AI พร้อมระบบ Timeout"""
        
        full_messages = []
        if system_prompt:
            full_messages.append({"role": "system", "content": system_prompt})
        full_messages.extend(messages)
        
        payload = {
            "model": model,
            "messages": full_messages,
            "temperature": 0.7,
            "max_tokens": 2048
        }
        
        for attempt in range(self.config.max_retries):
            try:
                response = self.session.post(
                    f"{self.base_url}/chat/completions",
                    json=payload,
                    timeout=(
                        self.config.connect_timeout,
                        self.config.read_timeout
                    )
                )
                response.raise_for_status()
                return self._parse_response(response.json())
                
            except requests.Timeout as e:
                wait_time = self._calculate_wait_time(attempt)
                print(f"⏰ Timeout เกิดขึ้น (ครั้งที่ {attempt + 1}): รอ {wait_time}s")
                time.sleep(wait_time)
                
            except requests.RequestException as e:
                print(f"❌ Request ล้มเหลว: {e}")
                if attempt < self.config.max_retries - 1:
                    time.sleep(self._calculate_wait_time(attempt))
                continue
        
        # Graceful Degradation
        if self.config.fallback_enabled:
            return self._fallback_response()
        return {"error": "Max retries exceeded", "success": False}
    
    def _calculate_wait_time(self, attempt: int) -> float:
        """คำนวณเวลารอตาม Strategy"""
        if self.config.strategy == TimeoutStrategy.EXPONENTIAL_BACKOFF:
            return min(2 ** attempt * 1.0, 60.0)  # Max 60 วินาที
        elif self.config.strategy == TimeoutStrategy.IMMEDIATE_RETRY:
            return 0.5
        return 2.0
    
    def _parse_response(self, response: Dict) -> Dict[str, Any]:
        """แปลง Response ให้อยู่ในรูปแบบที่ใช้งานง่าย"""
        try:
            return {
                "success": True,
                "content": response["choices"][0]["message"]["content"],
                "model": response.get("model", "unknown"),
                "usage": response.get("usage", {}),
                "latency_ms": response.elapsed.total_seconds() * 1000
            }
        except (KeyError, IndexError) as e:
            return {"success": False, "error": f"Parse error: {e}"}
    
    def _fallback_response(self) -> Dict[str, Any]:
        """Response สำรองเมื่อระบบหลักมีปัญหา"""
        return {
            "success": True,
            "content": "ขออภัยครับ ระบบ AI ตอนนี้มีภาระงานสูง กรุณาลองใหม่อีกครั้งในอีกสักครู่",
            "fallback": True,
            "suggestion": "ลองใช้โมเดลที่เล็กกว่า เช่น gpt-4o-mini หรือ deepseek-v3.2"
        }

วิธีใช้งาน
if __name__ == "__main__":
    agent = HolySheepAIAgent(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        config=TimeoutConfig(
            connect_timeout=5.0,
            read_timeout=45.0,
            max_retries=3,
            strategy=TimeoutStrategy.EXPONENTIAL_BACKOFF
        )
    )
    
    result = agent.chat_with_timeout(
        messages=[{"role": "user", "content": "ทักทายฉันหน่อย"}],
        model="gpt-4.1"
    )
    print(result)

โค้ดตัวอย่าง: Circuit Breaker Pattern สำหรับ AI Agent

import time
import threading
from collections import deque
from typing import Optional, Callable, Any
from dataclasses import dataclass, field
from enum import Enum
import requests

class CircuitState(Enum):
    CLOSED = "closed"       # ทำงานปกติ
    OPEN = "open"           # หยุดทำงานชั่วคราว
    HALF_OPEN = "half_open" # ทดสอบว่าระบบกลับมาหรือยัง

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5      # จำนวนครั้งที่ล้มเหลวก่อนเปิด Circuit
    success_threshold: int = 2      # จำนวนครั้งที่ต้องสำเร็จก่อนปิด Circuit
    timeout_duration: float = 60.0  # วินาทีที่จะรอก่อนลองใหม่
    half_open_max_calls: int = 3     # จำนวนครั้งที่จะลองในโหมด Half-Open

class CircuitBreaker:
    """Circuit Breaker Pattern สำหรับป้องกันระบบล่ม"""
    
    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 = threading.RLock()
    
    @property
    def is_available(self) -> bool:
        with self._lock:
            if self.state == CircuitState.CLOSED:
                return True
            elif self.state == CircuitState.OPEN:
                if self._should_attempt_reset:
                    self.state = CircuitState.HALF_OPEN
                    self.half_open_calls = 0
                    return True
                return False
            else:  # HALF_OPEN
                return self.half_open_calls < self.config.half_open_max_calls
    
    @property
    def _should_attempt_reset(self) -> bool:
        if self.last_failure_time is None:
            return False
        return (time.time() - self.last_failure_time) >= self.config.timeout_duration
    
    def record_success(self):
        with self._lock:
            self.failure_count = 0
            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                self.half_open_calls += 1
                if self.success_count >= self.config.success_threshold:
                    self.state = CircuitState.CLOSED
                    self.success_count = 0
                    print("🔄 Circuit Breaker: CLOSED (Recovery successful)")
            elif self.state == CircuitState.CLOSED:
                self.half_open_calls = 0
    
    def record_failure(self):
        with self._lock:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.OPEN
                self.half_open_calls = 0
                print("🔴 Circuit Breaker: OPEN (Failed during recovery)")
            elif (self.failure_count >= self.config.failure_threshold and 
                  self.state == CircuitState.CLOSED):
                self.state = CircuitState.OPEN
                print("🔴 Circuit Breaker: OPEN (Too many failures)")

class ResilientAIAgent:
    """AI Agent พร้อม Circuit Breaker และ Multi-Provider Fallback"""
    
    def __init__(self, primary_api_key: str, fallback_api_key: str = None):
        self.providers = {
            "primary": {
                "api_key": primary_api_key,
                "base_url": "https://api.holysheep.ai/v1",
                "models": ["gpt-4.1", "gpt-4o", "deepseek-v3.2"]
            },
            "fallback": {
                "api_key": fallback_api_key or primary_api_key,
                "base_url": "https://api.holysheep.ai/v1",
                "models": ["deepseek-v3.2", "gpt-4o-mini"]
            }
        }
        
        self.circuit_breakers = {
            name: CircuitBreaker(CircuitBreakerConfig())
            for name in self.providers
        }
        self.current_provider = "primary"
        self.request_history: deque = deque(maxlen=100)
    
    def send_message(
        self,
        message: str,
        model: str = "gpt-4.1",
        use_fallback: bool = False
    ) -> dict:
        """ส่งข้อความพร้อมระบบป้องกันความผิดพลาดหลายชั้น"""
        
        provider_name = "fallback" if use_fallback else self.current_provider
        provider = self.providers[provider_name]
        circuit = self.circuit_breakers[provider_name]
        
        # ตรวจสอบ Circuit Breaker
        if not circuit.is_available:
            if not use_fallback:
                print("⚠️ Provider หลักไม่พร้อมใช้งาน สลับไป Fallback")
                return self.send_message(message, model, use_fallback=True)
            return self._create_error_response("ทุก Provider ไม่พร้อมใช้งาน")
        
        try:
            start_time = time.time()
            response = self._make_request(provider, model, message)
            latency = (time.time() - start_time) * 1000
            
            circuit.record_success()
            self._log_request(provider_name, model, latency, True)
            
            return {
                "success": True,
                "content": response,
                "provider": provider_name,
                "latency_ms": round(latency, 2)
            }
            
        except requests.Timeout:
            circuit.record_failure()
            self._log_request(provider_name, model, 0, False)
            
            if not use_fallback:
                return self.send_message(message, model, use_fallback=True)
            return self._create_error_response("Timeout เกินขีดจำกัด")
            
        except requests.RequestException as e:
            circuit.record_failure()
            self._log_request(provider_name, model, 0, False)
            
            if not use_fallback:
                return self.send_message(message, model, use_fallback=True)
            return self._create_error_response(f"Request ล้มเหลว: {str(e)}")
    
    def _make_request(self, provider: dict, model: str, message: str) -> str:
        """ทำ Request ไปยัง Provider"""
        headers = {
            "Authorization": f"Bearer {provider['api_key']}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model if model in provider["models"] else provider["models"][0],
            "messages": [{"role": "user", "content": message}],
            "temperature": 0.7
        }
        
        response = requests.post(
            f"{provider['base_url']}/chat/completions",
            headers=headers,
            json=payload,
            timeout=30.0
        )
        response.raise_for_status()
        return response.json()["choices"][0]["message"]["content"]
    
    def _log_request(self, provider: str, model: str, latency: float, success: bool):
        """บันทึกประวัติการใช้งาน"""
        self.request_history.append({
            "provider": provider,
            "model": model,
            "latency": latency,
            "success": success,
            "timestamp": time.time()
        })
    
    def _create_error_response(self, message: str) -> dict:
        """สร้าง Response สำหรับ Error"""
        return {
            "success": False,
            "error": message,
            "suggestion": "กรุณาลองใหม่ในอีกสักครู่ หรือติดต่อผู้ดูแลระบบ"
        }
    
    def get_circuit_status(self) -> dict:
        """ดูสถานะ Circuit Breaker ของทุก Provider"""
        return {
            name: {
                "state": cb.state.value,
                "failures": cb.failure_count,
                "available": cb.is_available
            }
            for name, cb in self.circuit_breakers.items()
        }

วิธีใช้งาน
if __name__ == "__main__":
    agent = ResilientAIAgent(
        primary_api_key="YOUR_HOLYSHEEP_API_KEY",
        fallback_api_key="YOUR_HOLYSHEEP_API_KEY"
    )
    
    # ส่งข้อความ
    result = agent.send_message("อธิบายเรื่อง Circuit Breaker Pattern")
    print(result)
    
    # ดูสถานะ Circuit Breaker
    print("\n📊 Circuit Status:", agent.get_circuit_status())

โค้ดตัวอย่าง: Async AI Agent พร้อมระบบ Queue และ Retry

import asyncio
import aiohttp
import time
from typing import Optional, List, Dict, Any
from dataclasses import dataclass
import json

@dataclass
class AIRequest:
    request_id: str
    messages: List[Dict[str, str]]
    model: str
    priority: int = 1  # 1=สูงสุด, 5=ต่ำสุด
    max_retries: int = 3
    timeout: float = 45.0
    created_at: float = None
    
    def __post_init__(self):
        if self.created_at is None:
            self.created_at = time.time()

@dataclass
class AIResponse:
    request_id: str
    success: bool
    content: Optional[str] = None
    error: Optional[str] = None
    latency_ms: float = 0.0
    retries: int = 0
    provider: str = "unknown"

class AsyncAIAgentPool:
    """ระบบ AI Agent แบบ Async พร้อม Queue และ Load Balancing"""
    
    def __init__(
        self,
        api_keys: List[str],
        base_url: str = "https://api.holysheep.ai/v1",
        max_concurrent: int = 10,
        models: Dict[str, float] = None
    ):
        self.base_url = base_url
        self.max_concurrent = max_concurrent
        self.models = models or {
            "gpt-4.1": 0.008,
            "gpt-4o-mini": 0.0015,
            "deepseek-v3.2": 0.00042,
            "claude-sonnet-4.5": 0.015
        }
        
        # ใช้ Round-Robin สำหรับหลาย API Keys
        self.api_keys = api_keys
        self.current_key_index = 0
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.session: Optional[aiohttp.ClientSession] = None
        
        # สถิติ
        self.stats = {
            "total_requests": 0,
            "successful_requests": 0,
            "failed_requests": 0,
            "total_cost": 0.0,
            "avg_latency": 0.0
        }
    
    async def __aenter__(self):
        timeout = aiohttp.ClientTimeout(
            total=60.0,
            connect=10.0,
            sock_read=45.0
        )
        self.session = aiohttp.ClientSession(timeout=timeout)
        return self
    
    async def __aexit__(self, *args):
        if self.session:
            await self.session.close()
    
    def _get_next_api_key(self) -> str:
        """Round-Robin สำหรับ API Keys"""
        key = self.api_keys[self.current_key_index]
        self.current_key_index = (self.current_key_index + 1) % len(self.api_keys)
        return key
    
    async def send_request(self, request: AIRequest) -> AIResponse:
        """ส่ง Request พร้อม Retry Logic"""
        self.stats["total_requests"] += 1
        
        async with self.semaphore:
            for attempt in range(request.max_retries):
                try:
                    start_time = time.time()
                    content = await self._make_async_request(
                        api_key=self._get_next_api_key(),
                        model=request.model,
                        messages=request.messages
                    )
                    latency = (time.time() - start_time) * 1000
                    
                    # คำนวณค่าใช้จ่าย (ประมาณการ)
                    estimated_tokens = len(str(request.messages)) // 4
                    cost = self.models.get
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
Qdrant 向量检索 API 集成教程：完整迁移指南与性能优化实战
Claude Opus 4.7: การทดสอบเชิงปฏิบัติของ API การแยกข้อมูลเชิง
AI客服质检系统完整教程：从零开始搭建智能客服质量检查平台