ในการพัฒนาระบบ AI Agent ในระดับ Production สิ่งที่ผมพบเจอบ่อยที่สุดคือ "ระบบล่มกลางดึก" ไม่ว่าจะเป็น API Timeout, Rate Limit, หรือ Context Length Exceeded วันนี้ผมจะแชร์ประสบการณ์ตรงในการออกแบบ Exception Recovery ที่ใช้งานได้จริงในโปรเจกต์ E-commerce, RAG System และงาน Freelance
ทำไมต้องมี Error Recovery Mechanism?
สมมติว่าคุณสร้าง AI Agent สำหรับ Customer Service ของร้านค้าออนไลน์ ระบบทำงานได้ดีใน Lab แต่พอขึ้น Production เจอปัญหาทันที:
- API ของ LLM เกิด Timeout 15% ของ Request
- Token Limit ถูกตีกลับเมื่อ User ส่งข้อความยาว
- Rate Limit ทำให้ Bot หยุดทำงานกลางคัน
ถ้าไม่มี Recovery Mechanism ระบบจะ Fail Silently และ User จะได้รับประสบการณ์ที่แย่ หรือร้ายกว่านั้นคือ Task สำคัญหายไปโดยไม่มี Trace
กรณีศึกษา: AI ลูกค้าสัมพันธ์สำหรับ E-commerce
ผมเคยพัฒนา AI Agent สำหรับร้านค้าออนไลน์ที่มี 50,000+ ผู้ใช้ต่อวัน ปัญหาหลักคือ:
- การตอบคำถามสินค้าต้องดึงข้อมูลจาก Database
- บางครั้ง Database Connection ล้มเหลว
- LLM API ตอบช้าเกิน 30 วินาที
สถาปัตยกรรม Retry with Exponential Backoff
import time
import requests
from typing import Optional, Dict, Any
class HolySheepAIClient:
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
self.max_retries = 3
self.timeout = 45
def chat_completion(
self,
messages: list,
model: str = "gpt-4.1",
retry_count: int = 0
) -> Dict[str, Any]:
"""AI Agent พร้อม Exponential Backoff Retry"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 2000
}
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=self.timeout
)
# Handle specific errors
if response.status_code == 429:
# Rate Limit - Retry with exponential backoff
if retry_count < self.max_retries:
wait_time = (2 ** retry_count) * 2 # 2s, 4s, 8s
print(f"Rate Limited. รอ {wait_time} วินาที...")
time.sleep(wait_time)
return self.chat_completion(
messages, model, retry_count + 1
)
else:
return {"error": "max_retries_exceeded", "code": 429}
elif response.status_code == 500:
# Server Error - Internal retry
if retry_count < self.max_retries:
time.sleep(2 ** retry_count)
return self.chat_completion(
messages, model, retry_count + 1
)
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
if retry_count < self.max_retries:
print(f"Timeout. ลองใหม่ครั้งที่ {retry_count + 1}")
return self.chat_completion(
messages, model, retry_count + 1
)
return {"error": "timeout_after_retries"}
except requests.exceptions.ConnectionError:
return {"error": "connection_failed", "recovery": "manual_intervention"}
จุดสำคัญคือการใช้ Exponential Backoff เพื่อไม่ให้กระทบ Rate Limit ของ API Provider ที่ HolySheep AI ผมใช้ GPT-4.1 ที่ $8/MTok ซึ่งถูกกว่าที่อื่น 85%+ ทำให้การ Retry หลายครั้งไม่ทำให้ต้นทุนพุ่ง
กรณีศึกษา: Enterprise RAG System
สำหรับระบบ RAG (Retrieval-Augmented Generation) ขององค์กร ผมต้องออกแบบ Recovery ที่ซับซ้อนกว่าเพราะมีหลายจุดที่พังได้:
from enum import Enum
from dataclasses import dataclass
from typing import Callable, Optional
import logging
class ErrorSeverity(Enum):
TRANSIENT = "transient" # Retry ได้ทันที
RECOVERABLE = "recoverable" # ต้องแก้ด้วย Fallback
CRITICAL = "critical" # ต้องมี Human in the Loop
@dataclass
class RecoveryResult:
success: bool
data: Optional[Any] = None
error: Optional[str] = None
severity: ErrorSeverity = ErrorSeverity.TRANSIENT
requires_human: bool = False
class RAGAgentWithRecovery:
def __init__(self, llm_client, vector_db, fallback_handler=None):
self.llm = llm_client
self.vector_db = vector_db
self.fallback = fallback_handler
self.logger = logging.getLogger(__name__)
def query_with_recovery(
self,
user_query: str,
context: dict
) -> RecoveryResult:
"""
RAG Query พร้อม Multi-layer Recovery
Layer 1: Retry → Layer 2: Fallback → Layer 3: Human
"""
# === Layer 1: Retrieval with Retry ===
try:
retrieved_docs = self._retrieve_with_retry(
user_query,
max_attempts=2
)
except Exception as e:
self.logger.error(f"Retrieval failed: {e}")
retrieved_docs = []
# === Layer 2: LLM Generation with Fallback ===
try:
response = self._generate_with_fallback(
retrieved_docs,
user_query,
context
)
return RecoveryResult(success=True, data=response)
except Exception as e:
self.logger.warning(f"Generation failed: {e}")
# Try fallback model
if self.fallback:
try:
fallback_response = self.fallback(
retrieved_docs,
user_query,
context
)
return RecoveryResult(
success=True,
data=fallback_response,
error=f"Used fallback: {str(e)}"
)
except Exception as fallback_error:
self.logger.error(f"Fallback failed: {fallback_error}")
# === Layer 3: Human Intervention Queue ===
return RecoveryResult(
success=False,
severity=ErrorSeverity.CRITICAL,
requires_human=True,
error=str(e)
)
def _retrieve_with_retry(self, query: str, max_attempts: int):
"""Vector Retrieval พร้อม Retry Logic"""
for attempt in range(max_attempts):
try:
return self.vector_db.similarity_search(
query,
k=5
)
except Exception as e:
if attempt == max_attempts - 1:
raise
time.sleep(1 * attempt) # Simple backoff
def _generate_with_fallback(
self,
docs: list,
query: str,
context: dict
):
"""Generation พร้อม Model Fallback Chain"""
# Try primary: GPT-4.1
primary_model = "gpt-4.1"
try:
return self.llm.chat_completion(
messages=[
{"role": "system", "content": f"Context: {context}"},
{"role": "user", "content": f"Based on docs: {docs}\n\n{query}"}
],
model=primary_model
)
except Exception as e:
self.logger.warning(f"Primary model failed, trying fallback...")
# Fallback: DeepSeek V3.2 ($0.42/MTok - ถูกมาก!)
return self.llm.chat_completion(
messages=[...],
model="deepseek-v3.2"
)
เมตริกการติดตาม (Monitoring Metrics)
สิ่งที่ผมเรียนรู้จากโปรเจกต์ RAG คือต้องมี Dashboard ติดตาม:
- Retry Rate: ควรน้อยกว่า 5%
- Human Intervention Rate: ควรน้อยกว่า 0.5%
- Average Latency: ต้องต่ำกว่า 2 วินาที
- Cost per Query: ติดตามเพื่อ Optimize
ที่ HolySheep AI ความหน่วง (Latency) น้อยกว่า 50ms ทำให้ End-to-end RAG Query เสร็จใน 1.5-2 วินาที รวมถึง Token ที่ถูกมาก เช่น Claude Sonnet 4.5 อยู่ที่ $15/MTok หรือ Gemini 2.5 Flash แค่ $2.50/MTok
กรณีศึกษา: โปรเจกต์นักพัฒนาอิสระ
สำหรับ Freelance Project ที่ผมทำ ปัญหาหลักคือ Budget จำกัด ผมเลยต้องออกแบบ Smart Retry ที่คำนึงถึงต้นทุน:
from dataclasses import dataclass
from typing import List, Tuple
@dataclass
class ModelCostConfig:
model_name: str
cost_per_mtok: float # USD
avg_latency_ms: float
reliability_score: float # 0-1
class CostAwareRetryEngine:
"""
Smart Retry ที่เลือก Model ตาม Budget และ Reliability
ออกแบบมาสำหรับ Freelance ที่ต้องประหยัด
"""
def __init__(self, api_key: str):
self.client = HolySheepAIClient(api_key)
# HolySheep AI Pricing (Updated 2026)
self.models = {
"gpt-4.1": ModelCostConfig(
model_name="gpt-4.1",
cost_per_mtok=8.0,
avg_latency_ms=800,
reliability_score=0.95
),
"claude-sonnet-4.5": ModelCostConfig(
model_name="claude-sonnet-4.5",
cost_per_mtok=15.0,
avg_latency_ms=1200,
reliability_score=0.92
),
"gemini-2.5-flash": ModelCostConfig(
model_name="gemini-2.5-flash",
cost_per_mtok=2.50,
avg_latency_ms=400,
reliability_score=0.88
),
"deepseek-v3.2": ModelCostConfig(
model_name="deepseek-v3.2",
cost_per_mtok=0.42,
avg_latency_ms=600,
reliability_score=0.85
),
}
self.fallback_chain = [
"deepseek-v3.2", # ถูกสุด ลองก่อน
"gemini-2.5-flash", # ถูก + เร็ว
"gpt-4.1", # แพงแต่เสถียร
"claude-sonnet-4.5", # สุดท้าย
]
def execute_with_budget_fallback(
self,
messages: list,
estimated_tokens: int,
max_budget_usd: float = 0.10 # Budget ต่อ Query
) -> Tuple[bool, str, float]:
"""
Execute พร้อม Budget-aware Fallback
Returns: (success, response, actual_cost)
"""
for model_name in self.fallback_chain:
model = self.models[model_name]
estimated_cost = (estimated_tokens / 1_000_000) * model.cost_per_mtok
# Skip ถ้าเกิน Budget
if estimated_cost > max_budget_usd:
continue
try:
start = time.time()
result = self.client.chat_completion(
messages=messages,
model=model_name
)
actual_cost = (estimated_tokens / 1_000_000) * model.cost_per_mtok
return True, result, actual_cost
except Exception as e:
print(f"Model {model_name} failed: {e}")
continue
# ถ้าทุก Model ล้มเหลว
return False, "All models failed", 0.0
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. Error: 429 Rate Limit Exceeded
อาการ: API ตอบกลับ 429 Too Many Requests หลังจากส่ง Request ติดต่อกันหลายครั้ง
สาเหตุ: เกิน Rate Limit ของ Provider โดยเฉพาะช่วง Peak Hour
วิธีแก้ไข:
# เพิ่ม Rate Limiter แบบ Token Bucket
import asyncio
from collections import defaultdict
class RateLimiter:
def __init__(self, requests_per_minute: int = 60):
self.requests_per_minute = requests_per_minute
self.tokens = defaultdict(int)
self.last_update = defaultdict(float)
self.lock = asyncio.Lock()
async def acquire(self, key: str):
async with self.lock:
now = asyncio.get_event_loop().time()
elapsed = now - self.last_update[key]
# Replenish tokens
self.tokens[key] = min(
self.requests_per_minute,
self.tokens[key] + elapsed * (self.requests_per_minute / 60)
)
if self.tokens[key] < 1:
wait_time = (1 - self.tokens[key]) * (60 / self.requests_per_minute)
await asyncio.sleep(wait_time)
self.tokens[key] -= 1
self.last_update[key] = asyncio.get_event_loop().time()
2. Error: context_length_exceeded
อาการ: "This model's maximum context length is X tokens" เกิดขึ้นเมื่อ Conversation ยาวเกิน Limit
สาเหตุ: ไม่มีการจัดการ Conversation History ที่ดี และไม่ Truncate ข้อความเก่า
วิธีแก้ไข:
def smart_truncate_messages(
messages: list,
max_tokens: int = 120000, # 留 buffer สำหรับ Response
model: str = "gpt-4.1"
) -> list:
"""ตัด Messages เก่าออกแต่เก็บ System Prompt"""
if not messages:
return messages
# นับ Token ของแต่ละ Message (approx: 4 chars = 1 token)
def estimate_tokens(msg: dict) -> int:
content = msg.get("content", "")
return len(content) // 4 + 50 # +50 สำหรับ overhead
total_tokens = sum(estimate_tokens(m) for m in messages)
if total_tokens <= max_tokens:
return messages
# เก็บ System Prompt และ Messages ล่าสุด
system_prompt = None
other_messages = []
for msg in messages:
if msg.get("role") == "system":
system_prompt = msg
else:
other_messages.append(msg)
# Truncate จากข้างบน (เก่าสุด)
truncated = []
running_tokens = 0
if system_prompt:
running_tokens += estimate_tokens(system_prompt)
truncated.append(system_prompt)
for msg in reversed(other_messages):
msg_tokens = estimate_tokens(msg)
if running_tokens + msg_tokens <= max_tokens:
truncated.insert(1 if system_prompt else 0, msg)
running_tokens += msg_tokens
else:
break
return truncated
3. Error: connection_timeout หรือ connection_reset
อาการ: Request ค้างนานเกินไปแล้ว Timeout หรือ Connection ถูก Reset กลางทาง
สาเ�