ในโลกของการพัฒนา AI application ในปัจจุบัน การจัดการ request ที่มีปริมาณสูงเป็นสิ่งจำเป็นอย่างยิ่ง จากประสบการณ์ตรงของผมในการสร้างระบบที่รองรับผู้ใช้หลายหมื่นรายต่อวัน พบว่าการ implement rate limiting ที่ไม่เหมาะสมนำไปสู่ปัญหาร้ายแรงได้ ไม่ว่าจะเป็นค่าใช้จ่ายที่พุ่งสูงเกินควบคุม หรือ service ล่มจากการโจมตีแบบ DoS โดยไม่ตั้งใจ

บทความนี้จะพาคุณเข้าใจ architecture ของระบบ rate limiting ตั้งแต่พื้นฐานจนถึง advanced technique พร้อมโค้ด production-ready ที่สามารถนำไปใช้ได้จริง รวมถึงการ integrate กับ HolySheep AI ซึ่งมีความโดดเด่นเรื่อง latency เฉลี่ยต่ำกว่า 50ms และราคาประหยัดกว่า 85% เมื่อเทียบกับ provider อื่น

ทำไมต้องมี Rate Limiting?

ก่อนจะลงลึกใน technical detail มาทำความเข้าใจปัญหาที่ rate limiting แก้ไข

ปัญหาด้านความปลอดภัย

ปัญหาด้านประสิทธิภาพ

Rate Limiting Algorithm ที่นิยมใช้

1. Token Bucket Algorithm

เป็น algorithm ที่เหมาะกับ use case ที่ต้องการ allow burst traffic แต่ยังควบคุม average rate ได้ หลักการคือมี bucket ที่บรรจุ token จำนวนหนึ่ง แต่ละ request จะใช้ token 1 token และ token จะถูกเติมในอัตราคงที่

"""
Token Bucket Rate Limiter Implementation
Suitable for burst traffic handling
"""
import time
import threading
from dataclasses import dataclass, field
from typing import Dict, Optional
import asyncio


@dataclass
class TokenBucket:
    """Token Bucket implementation with thread-safe operations"""
    capacity: int
    refill_rate: float  # tokens per second
    tokens: float = field(init=False)
    last_refill: float = field(init=False)
    lock: threading.Lock = field(default_factory=threading.Lock)
    
    def __post_init__(self):
        self.tokens = float(self.capacity)
        self.last_refill = time.monotonic()
    
    def _refill(self) -> None:
        """Refill tokens based on elapsed time"""
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now
    
    def consume(self, tokens: int = 1) -> bool:
        """
        Attempt to consume tokens from bucket
        Returns True if successful, False if rate limited
        """
        with self.lock:
            self._refill()
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False
    
    def available_tokens(self) -> float:
        """Get current available tokens"""
        with self.lock:
            self._refill()
            return self.tokens


class RateLimiter:
    """
    Production-grade rate limiter supporting:
    - Per-client rate limiting
    - Multiple buckets per client (different endpoints)
    - Automatic cleanup of stale clients
    """
    
    def __init__(
        self,
        requests_per_second: float = 10,
        burst_size: int = 20,
        max_clients: int = 100000,
        cleanup_interval: int = 300
    ):
        self.buckets: Dict[str, TokenBucket] = {}
        self.rps = requests_per_second
        self.burst = burst_size
        self.max_clients = max_clients
        self.lock = threading.RLock()
        
        # Cleanup stale clients periodically
        self._cleanup_task = threading.Thread(
            target=self._cleanup_loop,
            args=(cleanup_interval,),
            daemon=True
        )
        self._cleanup_task.start()
    
    def _cleanup_loop(self, interval: int) -> None:
        """Remove stale buckets to prevent memory leak"""
        while True:
            time.sleep(interval)
            self._cleanup_stale()
    
    def _cleanup_stale(self) -> None:
        """Remove buckets with no activity in last hour"""
        cutoff = time.monotonic() - 3600
        with self.lock:
            stale = [
                k for k, v in self.buckets.items()
                if v.last_refill < cutoff
            ]
            for k in stale:
                del self.buckets[k]
            
            if len(stale) > 0:
                print(f"[RateLimiter] Cleaned up {len(stale)} stale clients")
    
    def check_rate_limit(self, client_id: str) -> tuple[bool, dict]:
        """
        Check if request should be allowed
        Returns (allowed, headers_dict)
        """
        with self.lock:
            if client_id not in self.buckets:
                if len(self.buckets) >= self.max_clients:
                    # Remove oldest client when at capacity
                    oldest = min(self.buckets.items(), key=lambda x: x[1].last_refill)
                    del self.buckets[oldest[0]]
                
                self.buckets[client_id] = TokenBucket(
                    capacity=self.burst,
                    refill_rate=self.rps
                )
            
            bucket = self.buckets[client_id]
            allowed = bucket.consume()
            
            return allowed, {
                'X-RateLimit-Limit': str(self.burst),
                'X-RateLimit-Remaining': str(int(bucket.available_tokens())),
                'X-RateLimit-Reset': str(int(time.time()) + 3600)
            }



Async version for asyncio-based applications

class AsyncRateLimiter:
    """Async rate limiter for high-performance applications"""
    
    def __init__(
        self,
        requests_per_second: float = 10,
        burst_size: int = 20,
        max_clients: int = 100000
    ):
        self.buckets: Dict[str, TokenBucket] = {}
        self.rps = requests_per_second
        self.burst = burst_size
        self.max_clients = max_clients
        self.lock = asyncio.Lock()
    
    async def check_rate_limit(self, client_id: str) -> tuple[bool, dict]:
        async with self.lock:
            if client_id not in self.buckets:
                if len(self.buckets) >= self.max_clients:
                    oldest = min(self.buckets.items(), key=lambda x: x[1].last_refill)
                    del self.buckets[oldest[0]]
                
                self.buckets[client_id] = TokenBucket(
                    capacity=self.burst,
                    refill_rate=self.rps
                )
            
            bucket = self.buckets[client_id]
            allowed = bucket.consume()
            
            return allowed, {
                'X-RateLimit-Limit': str(self.burst),
                'X-RateLimit-Remaining': str(int(bucket.available_tokens())),
                'Retry-After': '1' if not allowed else '0'
            }

2. Sliding Window Counter

Algorithm นี้ให้ความแม่นยำมากกว่า fixed window โดยไม่มี boundary spike ปัญหาคือใช้ memory มากกว่าเพราะต้องเก็บ timestamp ของทุก request

"""
Sliding Window Rate Limiter
More accurate than fixed window, no boundary spike
"""
from collections import deque
from time import monotonic
import threading
from typing import Dict, Tuple


class SlidingWindowRateLimiter:
    """
    Sliding window rate limiter using Redis-like sorted sets approach
    Implemented in-memory for single-node deployment
    """
    
    def __init__(
        self,
        max_requests: int = 100,
        window_seconds: int = 60,
        max_clients: int = 50000
    ):
        self.max_requests = max_requests
        self.window_ms = window_seconds * 1000
        self.max_clients = max_clients
        self.requests: Dict[str, deque] = {}
        self.lock = threading.RLock()
        self._cleanup()
    
    def _cleanup(self) -> None:
        """Remove old entries periodically"""
        now = monotonic() * 1000
        cutoff = now - self.window_ms
        
        with self.lock:
            for client_id in list(self.requests.keys()):
                # Remove expired timestamps
                while self.requests[client_id] and self.requests[client_id][0] < cutoff:
                    self.requests[client_id].popleft()
                
                # Remove empty queues
                if not self.requests[client_id]:
                    del self.requests[client_id]
    
    def is_allowed(self, client_id: str) -> Tuple[bool, Dict[str, str]]:
        """
        Check if request is allowed under rate limit
        Returns (allowed, headers)
        """
        now = monotonic() * 1000
        cutoff = now - self.window_ms
        
        with self.lock:
            # Initialize or get existing window
            if client_id not in self.requests:
                if len(self.requests) >= self.max_clients:
                    # Remove client with fewest requests
                    min_client = min(
                        self.requests.items(),
                        key=lambda x: len(x[1])
                    )
                    del self.requests[min_client[0]]
                self.requests[client_id] = deque()
            
            window = self.requests[client_id]
            
            # Remove expired entries
            while window and window[0] < cutoff:
                window.popleft()
            
            # Check if under limit
            if len(window) < self.max_requests:
                window.append(now)
                remaining = self.max_requests - len(window)
                return True, {
                    'X-RateLimit-Limit': str(self.max_requests),
                    'X-RateLimit-Remaining': str(remaining),
                    'X-RateLimit-Window': str(self.window_ms),
                    'X-RateLimit-Reset': str(int((now + self.window_ms) / 1000))
                }
            
            # Rate limited
            oldest = window[0]
            retry_after_ms = oldest + self.window_ms - now
            return False, {
                'X-RateLimit-Limit': str(self.max_requests),
                'X-RateLimit-Remaining': '0',
                'X-RateLimit-Window': str(self.window_ms),
                'Retry-After': str(int(retry_after_ms / 1000) + 1)
            }



Production-ready middleware for FastAPI

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware


class RateLimitMiddleware(BaseHTTPMiddleware):
    """FastAPI middleware for rate limiting"""
    
    def __init__(
        self,
        app,
        limiter: SlidingWindowRateLimiter,
        key_func=lambda req: req.client.host
    ):
        super().__init__(app)
        self.limiter = limiter
        self.key_func = key_func
    
    async def dispatch(self, request: Request, call_next):
        client_id = self.key_func(request)
        allowed, headers = self.limiter.is_allowed(client_id)
        
        if not allowed:
            return JSONResponse(
                status_code=429,
                content={
                    'error': 'Too Many Requests',
                    'message': 'Rate limit exceeded. Please try again later.',
                    'retry_after': headers.get('Retry-After', '60')
                },
                headers=headers
            )
        
        response = await call_next(request)
        
        # Add rate limit headers to successful responses
        for key, value in headers.items():
            response.headers[key] = value
        
        return response



Example usage with FastAPI

app = FastAPI()
limiter = SlidingWindowRateLimiter(
    max_requests=100,  # 100 requests
    window_seconds=60, # per minute
    max_clients=10000
)

app.add_middleware(RateLimitMiddleware, limiter=limiter)

การออกแบบ Distributed Rate Limiter

สำหรับระบบที่มีหลาย server instance การใช้ in-memory rate limiter จะไม่เพียงพอ เพราะแต่ละ instance จะมี counter แยกกัน ทำให้ผู้ใช้สามารถ bypass rate limit ได้โดยการกระจาย request ไปยังหลาย server

"""
Distributed Rate Limiter using Redis
Supports multiple rate limit strategies
"""
import redis
import time
import json
from typing import Tuple, Dict, Optional
from enum import Enum


class RateLimitStrategy(Enum):
    FIXED_WINDOW = "fixed"
    SLIDING_WINDOW = "sliding"
    TOKEN_BUCKET = "token_bucket"
    LEAK_BUCKET = "leak_bucket"


class RedisRateLimiter:
    """
    Production distributed rate limiter using Redis
    Supports multiple algorithms and Lua scripting for atomicity
    """
    
    # Lua script for fixed window - atomic increment and check
    FIXED_WINDOW_SCRIPT = """
    local key = KEYS[1]
    local limit = tonumber(ARGV[1])
    local window = tonumber(ARGV[2])
    local now = tonumber(ARGV[3])
    local window_start = math.floor(now / window) * window
    
    local current = redis.call('GET', key)
    if current == false then
        current = 0
    else
        current = tonumber(current)
    end
    
    if current >= limit then
        return {0, limit, current, window_start + window - now}
    end
    
    current = redis.call('INCR', key)
    if current == 1 then
        redis.call('EXPIRE', key, window)
    end
    
    return {1, limit, current, 0}
    """
    
    # Lua script for sliding window log
    SLIDING_WINDOW_SCRIPT = """
    local key = KEYS[1]
    local limit = tonumber(ARGV[1])
    local window = tonumber(ARGV[2])
    local now = tonumber(ARGV[3])
    local cutoff = now - window
    
    -- Remove old entries
    redis.call('ZREMRANGEBYSCORE', key, '-inf', cutoff)
    
    -- Count current requests
    local current = redis.call('ZCARD', key)
    
    if current >= limit then
        -- Get oldest entry to calculate retry time
        local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
        local retry_after = 0
        if oldest and #oldest >= 2 then
            retry_after = tonumber(oldest[2]) + window - now
        end
        return {0, limit, current, retry_after}
    end
    
    -- Add new request
    redis.call('ZADD', key, now, now .. '-' .. math.random())
    redis.call('EXPIRE', key, window)
    
    return {1, limit, current + 1, 0}
    """
    
    # Lua script for token bucket
    TOKEN_BUCKET_SCRIPT = """
    local key = KEYS[1]
    local capacity = tonumber(ARGV[1])
    local refill_rate = tonumber(ARGV[2])
    local tokens_requested = tonumber(ARGV[3])
    local now = tonumber(ARGV[4])
    
    local bucket = redis.call('HMGET', key, 'tokens', 'last_update')
    local tokens = tonumber(bucket[1])
    local last_update = tonumber(bucket[2])
    
    if tokens == nil then
        tokens = capacity
        last_update = now
    end
    
    -- Calculate token refill
    local elapsed = now - last_update
    tokens = math.min(capacity, tokens + elapsed * refill_rate)
    
    if tokens >= tokens_requested then
        tokens = tokens - tokens_requested
        redis.call('HMSET', key, 'tokens', tokens, 'last_update', now)
        redis.call('EXPIRE', key, 3600)
        return {1, capacity, tokens, 0}
    end
    
    -- Not enough tokens
    redis.call('HMSET', key, 'tokens', tokens, 'last_update', now)
    redis.call('EXPIRE', key, 3600)
    local retry_after = math.ceil((tokens_requested - tokens) / refill_rate)
    return {0, capacity, tokens, retry_after}
    """
    
    def __init__(
        self,
        redis_url: str = "redis://localhost:6379/0",
        strategy: RateLimitStrategy = RateLimitStrategy.SLIDING_WINDOW,
        default_limit: int = 100,
        default_window: int = 60,
        key_prefix: str = "ratelimit:"
    ):
        self.redis = redis.from_url(redis_url)
        self.strategy = strategy
        self.default_limit = default_limit
        self.default_window = default_window
        self.key_prefix = key_prefix
        
        # Register Lua scripts
        self._fixed_script = self.redis.register_script(self.FIXED_WINDOW_SCRIPT)
        self._sliding_script = self.redis.register_script(self.SLIDING_WINDOW_SCRIPT)
        self._token_bucket_script = self.redis.register_script(self.TOKEN_BUCKET_SCRIPT)
    
    def _get_key(self, identifier: str, endpoint: str = "default") -> str:
        """Generate Redis key for rate limiting"""
        return f"{self.key_prefix}{endpoint}:{identifier}"
    
    def check_rate_limit(
        self,
        client_id: str,
        limit: Optional[int] = None,
        window: Optional[int] = None,
        endpoint: str = "default"
    ) -> Tuple[bool, Dict[str, str]]:
        """
        Check rate limit for client
        Returns (allowed, headers_dict)
        """
        limit = limit or self.default_limit
        window = window or self.default_window
        key = self._get_key(client_id, endpoint)
        now = time.time()
        
        if self.strategy == RateLimitStrategy.FIXED_WINDOW:
            result = self._fixed_script(
                keys=[key],
                args=[limit, window, now]
            )
        elif self.strategy == RateLimitStrategy.SLIDING_WINDOW:
            result = self._sliding_script(
                keys=[key],
                args=[limit, window * 1000, now * 1000]
            )
        elif self.strategy == RateLimitStrategy.TOKEN_BUCKET:
            # Token bucket needs refill_rate in tokens per second
            refill_rate = limit / window
            result = self._token_bucket_script(
                keys=[key],
                args=[limit, refill_rate, 1, now]
            )
        else:
            raise ValueError(f"Unknown strategy: {self.strategy}")
        
        allowed = bool(result[0])
        retry_after = int(result[3]) if result[3] > 0 else 0
        
        headers = {
            'X-RateLimit-Limit': str(int(result[1])),
            'X-RateLimit-Remaining': str(max(0, int(result[1]) - int(result[2]))),
            'X-RateLimit-Window': str(window),
        }
        
        if not allowed:
            headers['Retry-After'] = str(retry_after)
        
        return allowed, headers
    
    def get_remaining(
        self,
        client_id: str,
        endpoint: str = "default"
    ) -> int:
        """Get remaining requests for client"""
        key = self._get_key(client_id, endpoint)
        
        if self.strategy == RateLimitStrategy.FIXED_WINDOW:
            current = self.redis.get(key)
            if current is None:
                return self.default_limit
            return max(0, self.default_limit - int(current))
        
        elif self.strategy == RateLimitStrategy.SLIDING_WINDOW:
            window = self.default_window * 1000
            cutoff = (time.time() * 1000) - window
            self.redis.zremrangebyscore(key, '-inf', cutoff)
            current = self.redis.zcard(key)
            return max(0, self.default_limit - current)
        
        return 0
    
    def reset(self, client_id: str, endpoint: str = "default") -> bool:
        """Reset rate limit for client"""
        key = self._get_key(client_id, endpoint)
        return bool(self.redis.delete(key))



Example: Tiered rate limiting for different plans

class TieredRateLimiter:
    """Rate limiter with different limits per subscription tier"""
    
    TIERS = {
        'free': {'requests': 60, 'window': 60},      # 60 req/min
        'basic': {'requests': 600, 'window': 60},    # 600 req/min
        'pro': {'requests': 6000, 'window': 60},     # 6000 req/min
        'enterprise': {'requests': 60000, 'window': 60},  # 60000 req/min
    }
    
    def __init__(self, redis_url: str):
        self.redis_limiter = RedisRateLimiter(
            redis_url=redis_url,
            strategy=RateLimitStrategy.SLIDING_WINDOW
        )
    
    def check(self, user_id: str, tier: str, endpoint: str = "api") -> Tuple[bool, Dict]:
        """Check rate limit based on user tier"""
        tier_config = self.TIERS.get(tier, self.TIERS['free'])
        
        return self.redis_limiter.check_rate_limit(
            client_id=user_id,
            limit=tier_config['requests'],
            window=tier_config['window'],
            endpoint=endpoint
        )

การ Integrate กับ AI Provider

เมื่อเข้าใจ algorithm แล้ว มาดูว่าจะนำไปใช้กับ AI API request อย่างไร โดยเฉพาะการ integrate กับ HolySheep AI ที่ให้บริการ API ราคาประหยัดมาก

"""
AI API Rate Limiter with Cost Control
Integrates with HolySheep AI API for production use
"""
import asyncio
import aiohttp
import time
from typing import Optional, Dict, List, Tuple
from dataclasses import dataclass
from enum import Enum
import json


class Model(Enum):
    """Supported AI models with pricing (per 1M tokens)"""
    GPT4_1 = {"id": "gpt-4.1", "input": 8.0, "output": 8.0}
    CLAUDE_SONNET_45 = {"id": "claude-sonnet-4.5", "input": 15.0, "output": 15.0}
    GEMINI_25_FLASH = {"id": "gemini-2.5-flash", "input": 2.50, "output": 2.50}
    DEEPSEEK_V32 = {"id": "deepseek-v3.2", "input": 0.42, "output": 0.42}


@dataclass
class CostBudget:
    """Cost control configuration"""
    daily_limit: float = 100.0      # Max $100 per day
    monthly_limit: float = 2000.0   # Max $2000 per month
    per_request_max: float = 0.50   # Max $0.50 per request


class AIRequestLimiter:
    """
    Intelligent rate limiter for AI API calls
    - Per-user rate limiting
    - Cost budgeting and tracking
    - Automatic fallback between providers
    - Retry with exponential backoff
    """
    
    BASE_URL = "https://api.holysheep.ai/v1"  # HolySheep AI endpoint
    
    def __init__(
        self,
        api_key: str,
        redis_url: str = "redis://localhost:6379/0",
        cost_budget: Optional[CostBudget] = None
    ):
        from redis_rate_limiter import RedisRateLimiter, RateLimitStrategy
        
        self.api_key = api_key
        self.cost_budget = cost_budget or CostBudget()
        
        # Redis-backed rate limiter
        self.rate_limiter = RedisRateLimiter(
            redis_url=redis_url,
            strategy=RateLimitStrategy.SLIDING_WINDOW,
            default_limit=1000,
            default_window=60
        )
        
        # Cost tracking Redis keys
        self.redis = self.rate_limiter.redis
        self._cost_prefix = "ai_cost:"
        
        # Semaphore for concurrent request control
        self.semaphore = asyncio.Semaphore(100)
    
    def _estimate_cost(
        self,
        model: str,
        input_tokens: int,
        output_tokens: int
    ) -> float:
        """Estimate request cost based on token usage"""
        model_info = None
        for m in Model:
            if m.value["id"] == model:
                model_info = m.value
                break
        
        if not model_info:
            return 0.0
        
        input_cost = (input_tokens / 1_000_000) * model_info["input"]
        output_cost = (output_tokens / 1_000_000) * model_info["output"]
        return input_cost + output_cost
    
    def _get_cost_period(self) -> Tuple[str, int]:
        """Get current cost period (daily/monthly)"""
        now = time.time()
        day_start = now - (now % 86400)
        month_start = now - (now % (86400 * 30))
        return day_start, month_start
    
    def _update_cost_tracking(
        self,
        user_id: str,
        cost: float
    ) -> bool:
        """
        Update cost tracking for user
        Returns False if budget exceeded
        """
        day_start, month_start = self._get_cost_period()
        
        # Check daily budget
        daily_key = f"{self._cost_prefix}daily:{user_id}:{int(day_start)}"
        monthly_key = f"{self._cost_prefix}monthly:{user_id}:{int(month_start)}"
        
        # Get current costs
        current_daily = float(self.redis.get(daily_key) or 0)
        current_monthly = float(self.redis.get(monthly_key) or 0)
        
        # Check budgets
        if current_daily + cost > self.cost_budget.daily_limit:
            return False
        if current_monthly + cost > self.cost_budget.monthly_limit:
            return False
        if cost > self.cost_budget.per_request_max:
            return False
        
        # Update costs atomically
        pipe = self.redis.pipeline()
        pipe.incrbyfloat(daily_key, cost)
        pipe.expire(daily_key, 86400 * 2)
        pipe.incrbyfloat(monthly_key, cost)
        pipe.expire(monthly_key, 86400 * 60)
        pipe.execute()
        
        return True
    
    async def chat_completion(
        self,
        user_id: str,
        messages: List[Dict],
        model: str = "deepseek-v3.2",
        max_tokens: int = 2048,
        temperature: float = 0.7
    ) -> Tuple[Optional[Dict], Dict]:
        """
        Make rate-limited AI API request
        
        Returns:
            (response_data, metadata) or (None, error_metadata)
        """
        # Estimate cost before making request
        estimated_input = sum(
            len(str(m.get('content', ''))) // 4
            for m in messages
        )
        estimated_output = max_tokens
        estimated_cost = self._estimate_cost(
            model, estimated_input, estimated_output
        )
        
        # Check cost budget
        if not self._update_cost_tracking(user_id, estimated_cost):
            return None, {
                'error': 'cost_exceeded',
                'message': 'Monthly or daily cost budget exceeded',
                'retry_after': 86400
            }
        
        # Check rate limit
        allowed, headers = self.rate_limiter.check_rate_limit(
            client_id=user_id,
            limit=1000,

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