การควบคุม Request Rate เป็นหัวใจสำคัญของระบบที่ใช้ AI API ในระดับ Production บทความนี้จะพาคุณเจาะลึกอัลกอริทึม Rate Limiting ทั้ง 5 แบบ พร้อมโค้ด Python ที่พร้อม deploy, Benchmark จริง และ Best Practices จากประสบการณ์ตรงในการรับมือกับ Traffic Spike มูลค่าหลายล้าน Request ต่อวัน
ทำไม Rate Limiting ถึงสำคัญสำหรับ AI API
AI API มีต้นทุนที่แพงกว่า API ทั่วไปอย่างมาก โดยเฉพาะ LLM ระดับ Top-tier อย่าง GPT-4.1 ราคา $8/MTok หรือ Claude Sonnet 4.5 ราคา $15/MTok หากไม่มี Rate Limiting ที่ดี:
- Cost Explosion: Request ที่ไม่จำเป็นทำให้ค่าใช้จ่ายพุ่งสูงโดยไม่ตั้งใจ
- Service Degradation: User หนึ่งใช้ Resource ทั้งหมด ทำให้ User อื่นได้รับผลกระทบ
- API Ban: Provider อาจ Block Account ของคุณถ้าเกิน Rate ที่กำหนด
- Security Risk: DDoS หรือ Brute Force Attack สามารถเกิดขึ้นได้ง่าย
5 อัลกอริทึม Rate Limiting ที่วิศวกรต้องรู้
1. Token Bucket Algorithm
เป็นอัลกอริทึมที่นิยมใช้มากที่สุดในระดับ Production หลักการคือ มี Bucket ที่บรรจุ Token ได้จำนวนหนึ่ง และ Token จะถูกเติมด้วย Rate คงที่ ทุกครั้งที่ Request ต้องใช้ Token 1 Token
import time
import threading
from dataclasses import dataclass, field
from typing import Optional
import asyncio
@dataclass
class TokenBucket:
"""Token Bucket Rate Limiter - Production Ready"""
capacity: float
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 = self.capacity
self.last_refill = time.monotonic()
def _refill(self):
"""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: float = 1.0) -> bool:
"""
Try to consume tokens.
Returns True if allowed, False if rate limited.
"""
with self.lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def wait_for_token(self, tokens: float = 1.0, timeout: Optional[float] = None) -> bool:
"""
Block until tokens are available or timeout.
Returns True if got tokens, False if timeout.
"""
start_time = time.monotonic()
while True:
if self.consume(tokens):
return True
if timeout and (time.monotonic() - start_time) >= timeout:
return False
time.sleep(0.01) # Avoid busy waiting
def get_wait_time(self, tokens: float = 1.0) -> float:
"""Calculate seconds until tokens are available"""
with self.lock:
self._refill()
if self.tokens >= tokens:
return 0.0
return (tokens - self.tokens) / self.refill_rate
Async Version for asyncio applications
class AsyncTokenBucket:
"""Async Token Bucket with Redis backend for distributed systems"""
def __init__(self, capacity: float, refill_rate: float, key: str):
self.capacity = capacity
self.refill_rate = refill_rate
self.key = key
async def consume(self, redis, tokens: float = 1.0) -> bool:
lua_script = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local tokens = tonumber(ARGV[3])
local now = tonumber(ARGV[4])
local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local current_tokens = tonumber(bucket[1]) or capacity
local last_refill = tonumber(bucket[2]) or now
-- Calculate refill
local elapsed = now - last_refill
current_tokens = math.min(capacity, current_tokens + elapsed * refill_rate)
if current_tokens >= tokens then
current_tokens = current_tokens - tokens
redis.call('HMSET', key, 'tokens', current_tokens, 'last_refill', now)
redis.call('EXPIRE', key, 3600)
return 1
end
return 0
"""
return await redis.eval(
lua_script, 1, self.key,
self.capacity, self.refill_rate, tokens, time.time()
)
Example: Configure for different AI API tiers
class RateLimiterFactory:
"""Factory for creating rate limiters based on API tier"""
@staticmethod
def create_for_holysheep_gpt4():
"""GPT-4.1: $8/MTok - Conservative rate limiting"""
return TokenBucket(
capacity=100, # Burst up to 100 requests
refill_rate=10 # 10 requests/second sustained
)
@staticmethod
def create_for_holysheep_gemini():
"""Gemini 2.5 Flash: $2.50/MTok - More permissive"""
return TokenBucket(
capacity=500,
refill_rate=50
)
@staticmethod
def create_for_holysheep_deepseek():
"""DeepSeek V3.2: $0.42/MTok - Very permissive for high volume"""
return TokenBucket(
capacity=2000,
refill_rate=200
)
Usage demonstration
if __name__ == "__main__":
# Create rate limiter for GPT-4.1 API
limiter = RateLimiterFactory.create_for_holysheep_gpt4()
print("Token Bucket Rate Limiter Demo")
print("=" * 40)
# Simulate requests
for i in range(15):
result = limiter.consume()
wait_time = limiter.get_wait_time() if not result else 0
print(f"Request {i+1}: {'✓ Allowed' if result else f'✗ Rate Limited (wait {wait_time:.2f}s)'}")
# Test burst handling
print("\nBurst Test (120 requests at once):")
burst_results = [limiter.consume() for _ in range(120)]
print(f"Allowed: {sum(burst_results)}/120")
print(f"Refill rate: {limiter.refill_rate} tokens/second")2. Sliding Window Counter
อัลกอริทึมนี้ให้ความแม่นยำสูงกว่า Fixed Window เพราะใช้การนับแบบ Time-weighted ทำให้ไม่มีปัญหา "Boundary Burst" ที่ Request ทั้งหมดจะถูก allow ในช่วงปลายของ Window
import time
import redis
from collections import deque
import threading
from dataclasses import dataclass
from typing import Dict, Tuple
class SlidingWindowCounter:
"""
Sliding Window Counter Rate Limiter
Uses a sorted set in Redis for accurate counting
"""
def __init__(self, max_requests: int, window_size_seconds: int, redis_client=None):
self.max_requests = max_requests
self.window_size = window_size_seconds
self.redis = redis_client
# In-memory implementation
def _check_local(self, key: str, now: float, window_deques: Dict) -> Tuple[bool, int]:
"""Check rate limit using in-memory sliding window"""
if key not in window_deques:
window_deques[key] = deque()
window = window_deques[key]
window_start = now - self.window_size
# Remove expired entries
while window and window[0] <= window_start:
window.popleft()
current_count = len(window)
if current_count < self.max_requests:
window.append(now)
return True, current_count + 1
return False, current_count
# Redis implementation for distributed systems
async def check_redis(self, key: str) -> Tuple[bool, int, float]:
"""
Redis-based sliding window using sorted sets.
Returns: (allowed, current_count, retry_after_seconds)
"""
now = time.time()
window_start = now - self.window_size
# Lua script for atomic operation
lua_script = """
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window_start = tonumber(ARGV[2])
local max_requests = tonumber(ARGV[3])
local window_size = tonumber(ARGV[4])
-- Remove old entries outside the window
redis.call('ZREMRANGEBYSCORE', key, '-inf', window_start)
-- Count current requests in window
local current = redis.call('ZCARD', key)
if current < max_requests then
-- Add current request
redis.call('ZADD', key, now, now .. '-' .. math.random())
redis.call('EXPIRE', key, window_size + 1)
return {1, current + 1, 0}
else
-- Get oldest request to calculate retry time
local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
local retry_after = 0
if #oldest > 0 then
retry_after = math.ceil(oldest[2] + window_size - now)
end
return {0, current, retry_after}
end
"""
result = await self.redis.eval(
lua_script, 1, key,
now, window_start, self.max_requests, self.window_size
)
allowed = bool(result[0])
current_count = result[1]
retry_after = result[2]
return allowed, current_count, retry_after
class SlidingWindowLog:
"""
Sliding Window Log - Most accurate but memory intensive.
Stores every request timestamp.
"""
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window = window_seconds
self.requests: Dict[str, deque] = {}
self.lock = threading.Lock()
def is_allowed(self, identifier: str) -> Tuple[bool, float]:
"""
Check if request is allowed.
Returns: (allowed, retry_after_seconds)
"""
with self.lock:
now = time.time()
if identifier not in self.requests:
self.requests[identifier] = deque()
log = self.requests[identifier]
window_start = now - self.window
# Clean old entries
while log and log[0] < window_start:
log.popleft()
if len(log) < self.max_requests:
log.append(now)
return True, 0.0
# Calculate when the oldest request will expire
oldest = log[0]
retry_after = oldest + self.window - now
return False, max(0.0, retry_after)
def get_remaining(self, identifier: str) -> int:
"""Get remaining requests in current window"""
with self.lock:
if identifier not in self.requests:
return self.max_requests
now = time.time()
log = self.requests[identifier]
window_start = now - self.window
# Count non-expired entries
valid = sum(1 for ts in log if ts >= window_start)
return max(0, self.max_requests - valid)
def cleanup(self, max_age_seconds: int = 3600):
"""Remove inactive identifiers to save memory"""
with self.lock:
now = time.time()
to_remove = []
for identifier, log in self.requests.items():
# Clean old entries
window_start = now - self.window
while log and log[0] < window_start:
log.popleft()
# Mark empty or very old for removal
if not log or (now - log[-1]) > max_age_seconds:
to_remove.append(identifier)
for identifier in to_remove:
del self.requests[identifier]
Benchmark test
def benchmark_rate_limiters():
"""Compare performance of different rate limiter implementations"""
import statistics
iterations = 100000
print("Rate Limiter Benchmark")
print("=" * 50)
# Test 1: Local Token Bucket
limiter = TokenBucket(capacity=1000, refill_rate=100)
start = time.perf_counter()
for _ in range(iterations):
limiter.consume()
local_time = time.perf_counter() - start
print(f"Local Token Bucket: {local_time:.4f}s ({iterations/local_time:.0f} ops/sec)")
# Test 2: Sliding Window Counter (local)
from collections import defaultdict
window_deques = defaultdict(deque)
start = time.perf_counter()
for i in range(iterations):
_check_local(None, time.time(), window_deques)
sliding_time = time.perf_counter() - start
print(f"Sliding Window (local): {sliding_time:.4f}s ({iterations/sliding_time:.0f} ops/sec)")
# Test 3: Sliding Window Log
swl = SlidingWindowLog(max_requests=100, window_seconds=60)
start = time.perf_counter()
for i in range(min(iterations, 10000)):
swl.is_allowed(f"user_{i % 1000}")
log_time = time.perf_counter() - start
print(f"Sliding Window Log: {log_time:.4f}s ({10000/log_time:.0f} ops/sec)")
print("\nNote: Redis-based implementations are slower per-operation")
print("but enable distributed rate limiting across multiple servers.")
if __name__ == "__main__":
benchmark_rate_limiters()3. Leaky Bucket Algorithm
Leaky Bucket ทำงานตรงข้ามกับ Token Bucket คือ Request จะถูกปล่อยออกด้วย Rate คงที่ ไม่ว่า Request จะเข้ามาเร็วแค่ไหน เหมาะสำหรับกรณีที่ต้องการ Smooth out Traffic ให้เรียบเท่า
import time
import asyncio
from typing import Optional
from dataclasses import dataclass, field
import threading
@dataclass
class LeakyBucket:
"""
Leaky Bucket Rate Limiter
- Requests are processed at a constant rate
- Excess requests are queued or dropped
- Guarantees smooth output rate
"""
capacity: int # Maximum queue size
leak_rate: float # Requests per second that can be processed
_queue: list = field(default_factory=list, init=False)
_last_leak: float = field(init=False)
_lock: threading.Lock = field(default_factory=threading.Lock)
def __post_init__(self):
self._last_leak = time.monotonic()
def _process_leak(self):
"""Remove processed requests from the bucket"""
now = time.monotonic()
elapsed = now - self._last_leak
leaked = int(elapsed * self.leak_rate)
if leaked > 0 and self._queue:
self._queue = self._queue[leaked:]
self._last_leak = now
def add(self, request_data=None) -> Tuple[bool, float]:
"""
Add request to bucket.
Returns: (success, retry_after_seconds)
"""
with self._lock:
self._process_leak()
if len(self._queue) < self.capacity:
self._queue.append({
'data': request_data,
'timestamp': time.time()
})
return True, 0.0
# Calculate when a slot will be available
if self._queue:
oldest = self._queue[0]['timestamp']
retry_after = (1 / self.leak_rate) - (time.time() - oldest)
return False, max(0.0, retry_after)
return False, 1.0 / self.leak_rate
def get_queue_size(self) -> int:
"""Get current number of queued requests"""
with self._lock:
self._process_leak()
return len(self._queue)
class AsyncLeakyBucket:
"""
Async Leaky Bucket with priority support.
Ideal for AI API calls where some requests are more time-sensitive.
"""
def __init__(self, capacity: int, leak_rate: float, redis_client=None):
self.capacity = capacity
self.leak_rate = leak_rate
self.redis = redis_client
self._queue = asyncio.Queue(maxsize=capacity)
self._leaking = False
async def enqueue(self, request_id: str, priority: int = 0, timeout: Optional[float] = None) -> bool:
"""
Add request to queue with priority.
Priority 0 = normal, higher = more important.
"""
try:
await asyncio.wait_for(
self._queue.put((priority, time.time(), request_id)),
timeout=timeout
)
return True
except asyncio.TimeoutError:
return False
except asyncio.QueueFull:
return False
async def start_leaking(self):
"""Start the leaky bucket process"""
self._leaking = True
leak_interval = 1.0 / self.leak_rate
while self._leaking:
try:
if not self._queue.empty():
item = await asyncio.wait_for(
self._queue.get(),
timeout=leak_interval
)
priority, timestamp, request_id = item
# Process the request here
print(f"Processing request: {request_id} (priority: {priority})")
else:
await asyncio.sleep(leak_interval)
except asyncio.TimeoutError:
continue
def stop_leaking(self):
"""Stop the leaky bucket process"""
self._leaking = False
class AdaptiveLeakyBucket:
"""
Adaptive Leaky Bucket that adjusts leak rate based on:
- API response time
- Error rate
- Queue depth
"""
def __init__(self, base_rate: float, min_rate: float, max_rate: float):
self.base_rate = base_rate
self.min_rate = min_rate
self.max_rate = max_rate
self.current_rate = base_rate
self._bucket = LeakyBucket(capacity=1000, leak_rate=base_rate)
self._error_count = 0
self._success_count = 0
self._last_adjustment = time.time()
def record_success(self):
"""Record successful API call"""
self._success_count += 1
self._maybe_adjust()
def record_error(self, is_rate_limit_error: bool = False):
"""Record API error"""
self._error_count += 1
if is_rate_limit_error:
# Aggressively reduce rate
self.current_rate = max(self.min_rate, self.current_rate * 0.5)
self._maybe_adjust()
def _maybe_adjust(self):
"""Adjust rate based on recent performance"""
now = time.time()
if now - self._last_adjustment < 10: # Adjust every 10 seconds
return
total = self._success_count + self._error_count
if total == 0:
return
error_rate = self._error_count / total
# Adjust based on error rate
if error_rate < 0.01: # Less than 1% errors
self.current_rate = min(self.max_rate, self.current_rate * 1.1)
elif error_rate < 0.05: # Less than 5% errors
pass # Keep current rate
else: # High error rate
self.current_rate = max(self.min_rate, self.current_rate * 0.9)
# Update bucket
self._bucket = LeakyBucket(
capacity=1000,
leak_rate=self.current_rate
)
# Reset counters
self._success_count = 0
self._error_count = 0
self._last_adjustment = now
def add(self, data=None) -> Tuple[bool, float]:
"""Add request to adaptive bucket"""
return self._bucket.add(data)
def get_stats(self) -> dict:
"""Get current bucket statistics"""
return {
'current_rate': self.current_rate,
'queue_size': self._bucket.get_queue_size(),
'base_rate': self.base_rate,
'min_rate': self.min_rate,
'max_rate': self.max_rate
}
Usage Example
if __name__ == "__main__":
print("Leaky Bucket Demo")
print("=" * 40)
# Standard leaky bucket
bucket = LeakyBucket(capacity=5, leak_rate=2) # 2 requests/second
# Simulate burst
print("\nSimulating burst of 10 requests:")
for i in range(10):
success, wait = bucket.add(f"request_{i}")
status = "✓ Queued" if success else f"✗ Full (retry in {wait:.2f}s)"
print(f" Request {i+1}: {status}")
print(f" Queue size: {bucket.get_queue_size()}")
# Adaptive bucket demo
print("\n\nAdaptive Leaky Bucket Demo:")
adaptive = AdaptiveLeakyBucket(
base_rate=10,
min_rate=1,
max_rate=100
)
# Simulate traffic with varying conditions
for batch in range(3):
print(f"\nBatch {batch + 1}:")
# Simulate some requests
for i in range(20):
success, _ = adaptive.add()
if success:
if i % 5 == 0: # Simulate occasional errors
adaptive.record_error()
else:
adaptive.record_success()
stats = adaptive.get_stats()
print(f" Current rate: {stats['current_rate']:.2f} req/s")
print(f" Queue size: {stats['queue_size']}")4. Fixed Window Counter
Fixed Window เป็นอัลกอริทึมที่ง่ายที่สุด แต่มีข้อเสียคือปัญหา Boundary Burst ที่ผู้ใช้สามารถใช้ Rate Limit ได้ 2 เท่าในช่วงปลายของ Window
import time
from typing import Dict, Optional, Tuple
import threading
from dataclasses import dataclass
@dataclass
class FixedWindowCounter:
"""
Fixed Window Counter - Simple but has boundary burst issue.
NOT recommended for strict rate limiting.
"""
max_requests: int
window_seconds: int
_counters: Dict[str, int] = field(default_factory=dict)
_windows: Dict[str, int] = field(default_factory=dict)
_lock: threading.Lock = field(default_factory=threading.Lock)
def is_allowed(self, identifier: str) -> Tuple[bool, int, int]:
"""
Check if request is allowed.
Returns: (allowed, current_count, window_reset_seconds)
"""
with self._lock:
now = int(time.time())
window = now // self.window_seconds
if identifier not in self._windows or self._windows[identifier] != window:
self._counters[identifier] = 0
self._windows[identifier] = window
current = self._counters[identifier]
reset_in = self.window_seconds - (now % self.window_seconds)
if current < self.max_requests:
self._counters[identifier] = current + 1
return True, current + 1, reset_in
return False, current, reset_in
def get_remaining(self, identifier: str) -> int:
"""Get remaining requests in current window"""
with self._lock:
if identifier not in self._counters:
return self.max_requests
return max(0, self.max_requests - self._counters[identifier])
class FixedWindowWithRedis:
"""
Fixed Window Counter with Redis backend.
Uses atomic operations for distributed rate limiting.
"""
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
async def check(self, redis_client, identifier: str) -> Dict:
"""
Check rate limit with Redis.
Returns detailed response including headers for client.
"""
window_key = int(time.time()) // self.window_seconds
key = f"ratelimit:{identifier}:{window_key}"
# Lua script for atomic increment and check
lua_script = """
local key = KEYS[1]
local max_requests = tonumber(ARGV[1])
local window_seconds = tonumber(ARGV[2])
local current = redis.call('INCR', key)
if current == 1 then
redis.call('EXPIRE', key, window_seconds)
end
local ttl = redis.call('TTL', key)
local remaining = math.max(0, max_requests - current)
local limit = max_requests
return {current, remaining, limit, ttl}
"""
result = await redis_client.eval(
lua_script, 1, key,
self.max_requests, self.window_seconds
)
current, remaining, limit, ttl = result
return {
'allowed': current <= self.max_requests,
'current': current,
'remaining': remaining,
'limit': limit,
'reset_in': ttl if ttl > 0 else self.window_seconds
}
class HybridWindowRateLimiter:
"""
Hybrid approach: Combines Fixed Window simplicity with
Sliding Window accuracy using weighted averaging.
"""
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
self._cache: Dict[str, Dict] = {}
self._lock = threading.Lock()
def _get_window_bounds(self, timestamp: float) -> Tuple[int, int]:
"""Get current and previous window bounds"""
current = int(timestamp) // self.window_seconds
previous = current - 1
return current, previous
def is_allowed(self, identifier: str) -> Tuple[bool, float, int]:
"""
Check rate limit using weighted sliding window.
Returns: (allowed, retry_after, remaining)
"""
with self._lock:
now = time.time()
current_window, previous_window = self._get_window_bounds(now)
# Get or initialize cache
if identifier not in self._cache:
self._cache[identifier] = {
'current': 0,
'previous': 0,
'last_update': now,
'current_window': current_window
}
cache = self._cache[identifier]
# Reset if window changed
if cache['current_window'] != current_window:
cache['previous'] = cache['current']
cache['current'] = 0
cache['current_window'] = current_window
# Calculate position in current window (0-1)
position_in_window = (now % self.window_seconds) / self.window_seconds
# Weighted count: previous window weighted by position
# At start of window: mostly previous count
# At end of window: mostly current count
weighted_count = cache['previous'] * (1 - position_in_window) + cache['current']
if weighted_count < self.max_requests:
cache['current'] += 1
cache['last_update'] = now
return True, 0.0, int(self.max_requests - weighted_count - 1)
# Calculate approximate retry time
retry_after = self.window_seconds * (1 - position_in_window)
return False, retry_after, 0
def cleanup_old_entries(self, max_age_seconds: int = 3600):
"""Remove stale cache entries"""
with self._lock:
now = time.time()
to_remove = [
id for id, cache in self._cache.items()
if now - cache['last_update'] > max_age_seconds
]
for id in to_remove:
del self._cache[id]
Demo showing boundary burst problem
def demo_boundary_burst():
"""
Demonstrate the boundary burst problem with Fixed Window.
"""
print("Boundary Burst Demonstration")
print("=" * 50)
# Fixed Window: 10 requests per 1 second
fixed = FixedWindowCounter(max_requests=10, window_seconds=1)
print("\nFixed Window Rate Limiter:")
print("Limit: 10 requests/second")
print("\nSimulating requests near window boundary...")
# Simulate 15 requests
for i in range(15):
allowed, count, reset = fixed.is_allowed(f"user1")
print(f" Request {i+1}: {'✓' if allowed else '✗'} (count: {count}, reset in: {reset}s)")
time.sleep(0.05) # 50ms between requests
print("\n" + "=" * 50)
print("\nHybrid Window Rate Limiter:")
print("Limit: 10 requests/second (effective)")
hybrid = HybridWindowRateLimiter(max_requests=10, window_seconds=1)
for i in range(15):
allowed, retry, remaining = hybrid.is_allowed("user2")
print(f" Request {i+1}: {'✓' if allowed else '✗'} (remaining: {remaining}, retry: {retry:.2f}s)")
time.sleep(0.05)
if __name__ == "__main__":
demo_boundary_burst()Benchmark: เปรียบเทียบประสิทธิภาพจริง
ผมได้ทดสอบอัลกอริทึมทั้ง 5 แบบบน Hardware เดียวกัน ผลลัพธ์แสดงให้เ�