ทำไมต้องมีระบบจำกัดคำขอ?

เคยไหมที่กำลังใช้ AI API อยู่ดีๆ ก็โดนบล็อกกะทันหัน? หรือเจอข้อผิดพลาด "429 Too Many Requests"? นี่คือสัญญาณว่าคุณส่งคำขอมากเกินกว่าที่ระบบกำหนด วันนี้ผมจะสอนคุณสร้างระบบจำกัดคำขอ (Rate Limiter) ด้วยตัวเอง ใช้ได้กับทุก API ไม่ว่าจะเป็น HolySheep AI หรือบริการอื่นๆ

สองอัลกอริทึมหลักที่ต้องรู้จัก

1. Token Bucket (ถังโทเค็น) — ยืดหยุ่นแต่ซับซ้อนกว่า

ลองนึกภาพถังใส่เหรียญ เหรียญจะถูกเติมเข้าไปเรื่อยๆ ตามจังหวะเวลาที่กำหนด เมื่อคุณอยากส่งคำขอ ต้องหยิบเหรียญจากถัง ถ้าถังว่างก็รอ นี่คือหลักการของ Token Bucket

ข้อดี:

ข้อเสีย:

2. Leaky Bucket (ถังรั่ว) — เรียบง่ายแต่เข้มงวด

ลองนึกภาพถังที่ก้นรั่ว คำขอเข้ามาเป็นหยดน้ำ ไหลออกทีละหยดด้วยความเร็วคงที่ ไม่ว่าจะส่งมากแค่ไหน ระบบจะปล่อยออกทีละน้อย ถ้าถังเต็ม คำขอใหม่จะถูกปฏิเสธทันที

ข้อดี:

ข้อเสีย:

สร้าง Token Bucket ด้วย Python กันเลย

เรามาเริ่มสร้างระบบ Token Bucket กันแบบง่ายๆ ก่อน สมมติคุณใช้ HolySheep AI ซึ่งมีราคาถูกมาก (GPT-4.1 เพียง $8 ต่อล้าน token) และเวลาตอบสนองน้อยกว่า 50ms รองรับ WeChat และ Alipay ด้วย

โครงสร้างพื้นฐานของ Token Bucket

import time
import threading
from dataclasses import dataclass, field
from typing import Optional
import requests

@dataclass
class TokenBucket:
    """ระบบ Token Bucket สำหรับจำกัดอัตราคำขอ"""
    capacity: float  # ขนาดถังสูงสุด (จำนวน token สูงสุดที่เก็บได้)
    refill_rate: float  # อัตราการเติม token ต่อวินาที
    
    tokens: float = field(init=False)  # token ปัจจุบันในถัง
    last_update: float = field(init=False)  # เวลาล่าสุดที่อัปเดต
    
    def __post_init__(self):
        self.tokens = self.capacity
        self.last_update = time.time()
    
    def _refill(self):
        """เติม token ตามเวลาที่ผ่านไป"""
        now = time.time()
        elapsed = now - self.last_update
        # เพิ่ม token ตามเวลาที่ผ่าน
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_update = now
    
    def consume(self, tokens: float = 1.0) -> bool:
        """
        พยายามใช้ token
        คืนค่า True ถ้าสำเร็จ, False ถ้าไม่มี token เพียงพอ
        """
        self._refill()
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False
    
    def wait_for_token(self, tokens: float = 1.0):
        """รอจนกว่าจะมี token เพียงพอ"""
        while not self.consume(tokens):
            time.sleep(0.01)  # รอสั้นๆ แล้วลองใหม่


ตัวอย่างการใช้งาน

bucket = TokenBucket(capacity=10, refill_rate=5)  # ถังใหญ่สุด 10, เติมวินาทีละ 5

print(f"เริ่มต้น: {bucket.tokens:.2f} tokens")
time.sleep(2)
bucket._refill()
print(f"หลังรอ 2 วินาที: {bucket.tokens:.2f} tokens")


ทดสอบการใช้ token

if bucket.consume(5):
    print("ใช้ไป 5 tokens สำเร็จ!")
else:
    print("ไม่มี token เพียงพอ")

สร้าง Leaky Bucket ด้วย Python

ต่อไปมาลองสร้าง Leaky Bucket กัน ระบบนี้จะควบคุมให้คำขอออกด้วยอัตราคงที่เสมอ

import time
import threading
from collections import deque
import requests

class LeakyBucket:
    """
    ระบบ Leaky Bucket สำหรับจำกัดอัตราคำขอ
    คำขอจะถูกปล่อยออกด้วยอัตราคงที่
    """
    def __init__(self, rate: float, capacity: int):
        """
        rate: จำนวนคำขอที่ปล่อยได้ต่อวินาที
        capacity: ขนาดถังสูงสุด
        """
        self.rate = rate  # อัตราการรั่ว (คำขอ/วินาที)
        self.capacity = capacity  # ความจุถัง
        self.bucket = deque()  # เก็บ timestamp ของคำขอที่รอ
        self.lock = threading.Lock()
        self.last_leak = time.time()
    
    def _leak(self):
        """ปล่อยคำขอเก่าออกตามเวลา"""
        now = time.time()
        elapsed = now - self.last_leak
        # คำนวณจำนวนที่รั่วออกได้
        leak_count = elapsed * self.rate
        
        while self.bucket and leak_count >= 1:
            self.bucket.popleft()
            leak_count -= 1
        
        self.last_leak = now
    
    def add_request(self, request_id: str = None) -> bool:
        """
        เพิ่มคำขอเข้าระบบ
        คืนค่า True ถ้าสำเร็จ, False ถ้าถังเต็ม
        """
        with self.lock:
            self._leak()
            
            if len(self.bucket) < self.capacity:
                # ยังมีที่ว่าง เพิ่มคำขอเข้าไป
                self.bucket.append({
                    'id': request_id,
                    'time': time.time()
                })
                return True
            else:
                # ถังเต็ม ต้องรอ
                return False
    
    def process_request(self, func, *args, **kwargs):
        """
        ประมวลผลคำขอพร้อมรอถ้าจำเป็น
        """
        # รอจนกว่าจะเพิ่มคำขอสำเร็จ
        while not self.add_request():
            time.sleep(0.01)
        
        # ประมวลผล
        return func(*args, **kwargs)
    
    def time_until_next_slot(self) -> float:
        """คำนวณเวลาที่ต้องรอจนช่องว่างถัดไป"""
        with self.lock:
            self._leak()
            if len(self.bucket) < self.capacity:
                return 0.0
            # คำนวณเวลาจนกว่าจะมีที่ว่าง
            return (self.capacity - len(self.bucket) + 1) / self.rate


ทดสอบ Leaky Bucket

bucket = LeakyBucket(rate=2, capacity=5)  # ปล่อยได้ 2 ต่อวินาที, รองรับ 5 คำขอ

print("เริ่มทดสอบ Leaky Bucket...")
for i in range(8):
    if bucket.add_request(f"req_{i}"):
        print(f"คำขอ {i} ถูกเพิ่มสำเร็จ (มี {len(bucket.bucket)} คำรอ)")
    else:
        print(f"คำขอ {i} ถูกปฏิเสธ - ถังเต็ม")
    time.sleep(0.1)

รวม Token Bucket กับการเรียก HolySheep AI

ต่อไปมาดูการนำ Token Bucket ไปใช้จริงกับ HolySheep AI ซึ่งมีราคาที่คุ้มค่ามาก อัตราแลกเปลี่ยน ¥1=$1 ประหยัดได้ถึง 85% เมื่อเทียบกับบริการอื่น

import time
import threading
from typing import Optional
import requests

class HolySheepRateLimiter:
    """ตัวจำกัดอัตราสำหรับ HolySheep AI API"""
    
    def __init__(self, requests_per_second: float = 10, burst_limit: int = 20):
        # ใช้ Token Bucket
        self.capacity = burst_limit  # รองรับ burst ได้สูงสุด
        self.refill_rate = requests_per_second  # เติมวินาทีละเท่าไหร่
        self.tokens = float(self.capacity)
        self.last_update = time.time()
        self.lock = threading.Lock()
    
    def _refill(self):
        now = time.time()
        elapsed = now - self.last_update
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_update = now
    
    def acquire(self) -> bool:
        """ขอ token เพื่อส่งคำขอ"""
        with self.lock:
            self._refill()
            if self.tokens >= 1:
                self.tokens -= 1
                return True
            return False
    
    def wait_and_call(self, endpoint: str, messages: list, 
                      api_key: str = "YOUR_HOLYSHEEP_API_KEY") -> dict:
        """
        รอจนกว่าจะมี token แล้วเรียก API
        """
        while not self.acquire():
            time.sleep(0.05)  # รอสักหน่อย
        
        # เรียก HolySheep AI
        url = "https://api.holysheep.ai/v1/chat/completions"
        headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": "gpt-4.1",
            "messages": messages
        }
        
        response = requests.post(url, headers=headers, json=payload)
        return response.json()


ตัวอย่างการใช้งาน

limiter = HolySheepRateLimiter(requests_per_second=5, burst_limit=15)


ส่งคำขอ 20 คำขอติดต่อกัน

messages = [{"role": "user", "content": "ทักทายโลก"}]

print("เริ่มส่งคำขอ 20 ครั้ง...")
start = time.time()

for i in range(20):
    try:
        # สมมติว่าเรียก API จริง
        print(f"คำขอที่ {i+1}: รอ token...", end=" ")
        # result = limiter.wait_and_call("/chat/completions", messages)
        print("สำเรัจ!")
    except Exception as e:
        print(f"ผิดพลาด: {e}")

elapsed = time.time() - start
print(f"\nเวลาทั้งหมด: {elapsed:.2f} วินาที")
print(f"เฉลี่ย: {20/elapsed:.1f

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