AI API 生产环境 Checklist：上线前 20 项检查

บทนำ: บทเรียนจาก Production ที่ล่มไป 3 วัน

เมื่อเดือนที่แล้ว ผมเพิ่งผ่านประสบการณ์ที่ไม่อยากจดจำ — ระบบ AI API ที่พัฒนามา 3 เดือนล่มในวันแรกที่ปล่อย production เพราะข้อผิดพลาดเดียว:

ConnectionError: HTTPSConnectionPool(host='api.holysheep.ai', port=443): Max retries exceeded with url: /v1/chat/completions (Caused by NewConnectionError: '<urllib3.connection.HTTPSConnection object at 0x7f8a2c3d4b50>: Failed to establish a new connection: [Errno 110] Connection timed out'))

หลังจากแก้ปัญหาและทำ post-mortem ผมได้ compile "checklist 20 ข้อ" ที่ทุกครั้งก่อน deploy ต้องผ่านทุกข้อ ---

1. ตรวจสอบ API Key และ Environment Variables

# ✅ วิธีที่ถูกต้อง — ใช้ environment variable
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get("HOLYSHEEP_API_KEY"),  # ห้าม hardcode!
    base_url="https://api.holysheep.ai/v1"
)

ตรวจสอบว่า key ถูกตั้งค่าหรือไม่
if not os.environ.get("HOLYSHEEP_API_KEY"):
    raise ValueError("HOLYSHEEP_API_KEY environment variable is not set")

# ❌ วิธีที่ผิด — hardcode key โดยตรง (อันตรายมาก!)
API_KEY = "sk-holysheep-xxxxx"  # เปิดเผยบน GitHub = ถูกลักลอบใช้ทันที

✅ ใช้ .env file + python-dotenv
.env
HOLYSHEEP_API_KEY=sk-holysheep-xxxxx
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1

---

2. Timeout Configuration ที่เหมาะสม

import httpx
from openai import OpenAI

client = OpenAI(
    api_key=os.environ.get("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1",
    http_client=httpx.Client(
        timeout=httpx.Timeout(60.0, connect=10.0)  # total=60s, connect=10s
    )
)

หรือใช้ async สำหรับ high-concurrency
async_client = OpenAI(
    api_key=os.environ.get("HOLYSHEEP_API_KEY"),
    base_url="https://api.holysheep.ai/v1",
    http_client=httpx.AsyncClient(
        timeout=httpx.Timeout(60.0, connect=10.0)
    )
)

**ค่าแนะนำ:** - **Connect timeout**: 5-10 วินาที (ป้องกัน hang ตอน connection) - **Read timeout**: 60-120 วินาที (สำหรับ LLM response ที่อาจใช้เวลา) - **Total timeout**: 120 วินาทีสำหรับ streaming ---

3. Retry Logic ที่มี Exponential Backoff

import time
import httpx
from openai import APIError, RateLimitError, APITimeoutError

def call_with_retry(client, messages, max_retries=3):
    """เรียก API พร้อม retry logic แบบ exponential backoff"""
    
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-4.1",
                messages=messages,
                timeout=60.0
            )
            return response
            
        except RateLimitError:
            # HolySheep มี rate limit — รอก่อน retry
            wait_time = 2 ** attempt + 0.5  # 2.5s, 4.5s, 8.5s
            print(f"Rate limited, waiting {wait_time}s...")
            time.sleep(wait_time)
            
        except APITimeoutError:
            wait_time = 2 ** attempt
            print(f"Timeout, retrying in {wait_time}s...")
            time.sleep(wait_time)
            
        except APIError as e:
            if e.status_code in [500, 502, 503, 504]:
                # Server error — retry ได้
                wait_time = 2 ** attempt
                print(f"Server error {e.status_code}, retrying...")
                time.sleep(wait_time)
            else:
                # Client error — ไม่ควร retry
                raise
                
    raise Exception(f"Failed after {max_retries} retries")

---

4. Rate Limit Handling

import time
from collections import defaultdict
from threading import Lock

class RateLimiter:
    """Token bucket rate limiter สำหรับ HolySheep API"""
    
    def __init__(self, requests_per_minute=60):
        self.requests_per_minute = requests_per_minute
        self.requests = []
        self.lock = Lock()
    
    def acquire(self):
        with self.lock:
            now = time.time()
            # ลบ request ที่เก่ากว่า 1 นาที
            self.requests = [t for t in self.requests if now - t < 60]
            
            if len(self.requests) >= self.requests_per_minute:
                sleep_time = 60 - (now - self.requests[0])
                print(f"Rate limit reached, sleeping {sleep_time:.1f}s")
                time.sleep(sleep_time)
                return self.acquire()  # retry
            
            self.requests.append(now)

ใช้งาน
limiter = RateLimiter(requests_per_minute=60)

def send_request():
    limiter.acquire()  # รอจนกว่าจะพร้อม
    return client.chat.completions.create(
        model="gpt-4.1",
        messages=[{"role": "user", "content": "Hello"}]
    )

---

5. Input Validation ก่อนส่งไป API

from pydantic import BaseModel, Field, validator
from typing import Optional

class ChatRequest(BaseModel):
    model: str = Field(default="gpt-4.1")
    messages: list = Field(min_length=1)
    temperature: float = Field(default=0.7, ge=0, le=2)
    max_tokens: Optional[int] = Field(default=2048, ge=1, le=128000)
    
    @validator('messages')
    def validate_messages(cls, v):
        for msg in v:
            if not isinstance(msg.get('role'), str):
                raise ValueError("Each message must have a 'role' string")
            if not isinstance(msg.get('content'), str):
                raise ValueError("Each message must have 'content' string")
            if msg['role'] not in ['system', 'user', 'assistant']:
                raise ValueError(f"Invalid role: {msg['role']}")
        return v
    
    @validator('model')
    def validate_model(cls, v):
        allowed_models = ['gpt-4.1', 'claude-sonnet-4.5', 'gemini-2.5-flash', 'deepseek-v3.2']
        if v not in allowed_models:
            raise ValueError(f"Model must be one of: {allowed_models}")
        return v

def create_chat_completion(request: ChatRequest):
    validated = ChatRequest(**request)
    return client.chat.completions.create(**validated.model_dump())

---

6. Output Validation และ Error Handling

from pydantic import BaseModel
from typing import Optional, List

class Message(BaseModel):
    role: str
    content: str

class Usage(BaseModel):
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int

class ChatResponse(BaseModel):
    id: str
    model: str
    choices: List[Message]
    usage: Usage
    created: int

def safe_parse_response(response):
    """Parse response อย่างปลอดภัยพร้อม fallback"""
    
    try:
        if response is None:
            return {"error": "Empty response from API"}
        
        if hasattr(response, 'choices') and len(response.choices) > 0:
            return {
                "content": response.choices[0].message.content,
                "model": response.model,
                "usage": {
                    "prompt_tokens": response.usage.prompt_tokens,
                    "completion_tokens": response.usage.completion_tokens,
                    "total_tokens": response.usage.total_tokens
                }
            }
        
        return {"error": "Unexpected response format", "raw": str(response)}
        
    except Exception as e:
        return {"error": f"Parse error: {str(e)}"}

---

7. Monitoring และ Logging

import time
import logging
from functools import wraps

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def monitor_api_call(func):
    """Decorator สำหรับ monitor latency และ errors"""
    
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        request_id = f"req_{int(time.time() * 1000)}"
        
        logger.info(f"[{request_id}] Starting API call")
        
        try:
            result = func(*args, **kwargs)
            latency = (time.time() - start_time) * 1000
            
            logger.info(f"[{request_id}] Success: {latency:.2f}ms")
            
            # เก็บ metrics
            metrics.append({
                "timestamp": time.time(),
                "latency_ms": latency,
                "success": True,
                "request_id": request_id
            })
            
            return result
            
        except Exception as e:
            latency = (time.time() - start_time) * 1000
            logger.error(f"[{request_id}] Failed: {str(e)} ({latency:.2f}ms)")
            
            metrics.append({
                "timestamp": time.time(),
                "latency_ms": latency,
                "success": False,
                "error": str(e),
                "request_id": request_id
            })
            
            raise
    
    return wrapper

metrics = []

@monitor_api_call
def call_holysheep(messages):
    return client.chat.completions.create(
        model="gpt-4.1",
        messages=messages
    )

ดู metrics
def get_metrics_summary():
    if not metrics:
        return "No metrics yet"
    
    successful = [m for m in metrics if m.get('success')]
    failed = [m for m in metrics if not m.get('success')]
    latencies = [m['latency_ms'] for m in successful]
    
    return {
        "total_requests": len(metrics),
        "success_rate": len(successful) / len(metrics) * 100,
        "avg_latency_ms": sum(latencies) / len(latencies) if latencies else 0,
        "p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)] if latencies else 0
    }

---

8. Circuit Breaker Pattern

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # ปกติ — ทำงานได้
    OPEN = "open"          # ปิด — ปฏิเสธ request ทันที
    HALF_OPEN = "half_open"  # กึ่งเปิด — ลองทดสอบ

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.failures = 0
        self.state = CircuitState.CLOSED
        self.last_failure_time = None
    
    def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker is OPEN - too many failures")
        
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
    
    def _on_success(self):
        self.failures = 0
        self.state = CircuitState.CLOSED
    
    def _on_failure(self):
        self.failures += 1
        self.last_failure_time = time.time()
        
        if self.failures >= self.failure_threshold:
            self.state = CircuitState.OPEN
            print(f"Circuit breaker OPENED after {self.failures} failures")

ใช้งาน
breaker = CircuitBreaker(failure_threshold=5, timeout=60)

def safe_api_call(messages):
    return breaker.call(client.chat.completions.create,
        model="gpt-4.1",
        messages=messages
    )

---

9. Caching Strategy

import hashlib
import json
import time
from typing import Optional
import redis

class APICache:
    """Cache LLM responses ด้วย Redis สำหรับ identical requests"""
    
    def __init__(self, redis_client, ttl=3600):
        self.cache = redis_client
        self.ttl = ttl
    
    def _make_key(self, messages, model, temperature):
        content = json.dumps({"messages": messages, "model": model, "temperature": temperature})
        return f"llm_cache:{hashlib.sha256(content.encode()).hexdigest()}"
    
    def get_or_fetch(self, messages, model="gpt-4.1", temperature=0.7):
        cache_key = self._make_key(messages, model, temperature)
        
        # ลองดึงจาก cache
        cached = self.cache.get(cache_key)
        if cached:
            print("Cache HIT")
            return json.loads(cached)
        
        print("Cache MISS - calling API")
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temperature
        )
        
        # เก็บใน cache
        result = response.choices[0].message.content
        self.cache.setex(cache_key, self.ttl, json.dumps(result))
        
        return result

ใช้งาน
try:
    cache = APICache(redis.Redis(host='localhost', port=6379), ttl=3600)
    
    # Request แรก - เรียก API
    result1 = cache.get_or_fetch([{"role": "user", "content": "Hello"}])
    
    # Request ที่สองด้วย message เดิม - ใช้ cache
    result2 = cache.get_or_fetch([{"role": "user", "content": "Hello"}])
except:
    print("Redis not available, skipping cache")

---

10. Health Check Endpoint

from fastapi import FastAPI, HTTPException
import httpx

app = FastAPI()

HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

@app.get("/health")
async def health_check():
    """Health check endpoint สำหรับ load balancer"""
    
    checks = {
        "api_reachable": False,
        "api_responsive": False,
        "latency_ms": None
    }
    
    try:
        # ตรวจสอบว่า HolySheep API accessible
        async with httpx.AsyncClient() as client:
            start = time.time()
            response = await client.get(
                f"{HOLYSHEEP_BASE_URL}/models",
                headers={"Authorization": f"Bearer {os.environ.get('HOLYSHEEP_API_KEY')}"},
                timeout=5.0
            )
            checks["latency_ms"] = (time.time() - start) * 1000
            checks["api_reachable"] = True
            checks["api_responsive"] = response.status_code == 200
            
    except Exception as e:
        checks["error"] = str(e)
    
    # ถ้า latency > 1000ms ถือว่า unhealthy
    if checks["latency_ms"] and checks["latency_ms"] > 1000:
        raise HTTPException(status_code=503, detail="API latency too high")
    
    if not checks["api_responsive"]:
        raise HTTPException(status_code=503, detail="API not responding")
    
    return {"status": "healthy", "checks": checks}

@app.get("/ready")
async def readiness_check():
    """Readiness check — พร้อมรับ traffic หรือยัง"""
    return {"status": "ready"}

---

11-20. Checklist ฉบับย่อ

| # | รายการตรวจสอบ | ความสำคัญ | |---|---------------|-----------| | 11 | SSL/TLS certificate validation เปิดอยู่ | 🔴 บังคับ | | 12 | Request size limit (max_tokens) | 🔴 บังคับ | | 13 | Content filtering/sanitization | 🔴 บังคับ | | 14 | Cost monitoring & budget alert | 🟡 สำคัญ | | 15 | Graceful degradation เมื่อ API down | 🟡 สำคัญ | | 16 | Request/response logging (mask sensitive) | 🟡 สำคัญ | | 17 | Load testing ก่อน deploy | 🟡 สำคัญ | | 18 | Backup API key rotation plan | 🟢 แนะนำ | | 19 | Documentation สำหรับ dev team | 🟢 แนะนำ | | 20 | Runbook สำหรับ incident | 🟢 แนะนำ | ---

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. 401 Unauthorized — API Key ไม่ถูกต้อง

# ❌ ข้อผิดพลาด
openai.AuthenticationError: Incorrect API key provided

✅ แก้ไข: ตรวจสอบ format และการตั้งค่า environment
import os

1. ตรวจสอบว่า key ถูก set
api_key = os.environ.get("HOLYSHEEP
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
การผสานรวม AI ในสถาปัตยกรรมไมโครเซอร์วิส: แนวปฏิบัติที่ดีที่
RAG การประเมินผลด้วย Context Precision และ Answer Relevance 
AI Unit Test Generation: เครื่องมือสร้าง Test Case อัตโนมัติ

บทนำ: บทเรียนจาก Production ที่ล่มไป 3 วัน

1. ตรวจสอบ API Key และ Environment Variables

ตรวจสอบว่า key ถูกตั้งค่าหรือไม่

✅ ใช้ .env file + python-dotenv

.env

HOLYSHEEP_API_KEY=sk-holysheep-xxxxx

HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1

2. Timeout Configuration ที่เหมาะสม

หรือใช้ async สำหรับ high-concurrency

3. Retry Logic ที่มี Exponential Backoff

4. Rate Limit Handling

ใช้งาน

5. Input Validation ก่อนส่งไป API

6. Output Validation และ Error Handling

7. Monitoring และ Logging

ดู metrics

8. Circuit Breaker Pattern

ใช้งาน

9. Caching Strategy

ใช้งาน

10. Health Check Endpoint

11-20. Checklist ฉบับย่อ

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. 401 Unauthorized — API Key ไม่ถูกต้อง

openai.AuthenticationError: Incorrect API key provided

✅ แก้ไข: ตรวจสอบ format และการตั้งค่า environment

1. ตรวจสอบว่า key ถูก set

แหล่งข้อมูลที่เกี่ยวข้อง

บทความที่เกี่ยวข้อง

🔥 ลอง HolySheep AI

`HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1`