Giới thiệu về Rate Limiting trong Hệ thống AI Gateway

Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến khi triển khai hệ thống distributed rate limiter sử dụng Redis cho HolySheep AI Gateway — nơi chúng tôi xử lý hơn 2 triệu request mỗi ngày với độ trễ trung bình dưới 50ms.

Tại sao cần Rate Limiting?

Khi làm việc với các API AI như OpenAI, Anthropic, hay HolySheep, việc kiểm soát tần suất request là vô cùng quan trọng vì:

Kiến trúc Rate Limiter với Redis

1. Sliding Window Algorithm

Đây là thuật toán được đội ngũ HolySheep khuyên dùng vì độ chính xác cao và không có hiện tượng "burst" như fixed window.

import redis
import time
from typing import Tuple, Optional

class SlidingWindowRateLimiter:
    """
    Sliding Window Rate Limiter sử dụng Redis Sorted Set
    Độ chính xác cao, phù hợp cho multi-instance deployment
    """
    
    def __init__(self, redis_client: redis.Redis, 
                 key_prefix: str = "ratelimit",
                 window_size: int = 60):
        self.redis = redis_client
        self.key_prefix = key_prefix
        self.window_size = window_size  # Window size in seconds
    
    def _get_key(self, identifier: str, endpoint: str) -> str:
        """Tạo unique key cho mỗi user/endpoint"""
        return f"{self.key_prefix}:{identifier}:{endpoint}"
    
    def is_allowed(self, identifier: str, endpoint: str, 
                   max_requests: int) -> Tuple[bool, dict]:
        """
        Kiểm tra xem request có được phép thực thi không
        
        Args:
            identifier: User ID hoặc API key
            endpoint: Tên endpoint cần rate limit
            max_requests: Số request tối đa trong window
        
        Returns:
            Tuple[bool, dict]: (is_allowed, info_dict)
        """
        key = self._get_key(identifier, endpoint)
        now = time.time()
        window_start = now - self.window_size
        
        # Lua script để đảm bảo atomicity
        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])
        
        -- Xóa các request cũ ngoài window
        redis.call('ZREMRANGEBYSCORE', key, 0, window_start)
        
        -- Đếm số request hiện tại
        local current_count = redis.call('ZCARD', key)
        
        if current_count < max_requests then
            -- Thêm request mới
            redis.call('ZADD', key, now, now .. '-' .. math.random())
            redis.call('EXPIRE', key, window_size)
            return {1, max_requests - current_count - 1}
        else
            -- Lấy thời gian của request cũ nhất
            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_count, retry_after}
        end
        """
        
        result = self.redis.eval(
            lua_script, 1, key, now, window_start, 
            max_requests, self.window_size
        )
        
        is_allowed = bool(result[0])
        remaining = result[1]
        retry_after = result[2] if not is_allowed else 0
        
        return is_allowed, {
            "allowed": is_allowed,
            "remaining": remaining,
            "retry_after": retry_after,
            "limit": max_requests,
            "reset_at": int(now + self.window_size)
        }



=== Ví dụ sử dụng với HolySheep AI ===

redis_client = redis.Redis(host='localhost', port=6379, db=0)
rate_limiter = SlidingWindowRateLimiter(
    redis_client=redis_client,
    key_prefix="holysheep",
    window_size=60
)


Kiểm tra rate limit cho API key

api_key = "YOUR_HOLYSHEEP_API_KEY"
is_allowed, info = rate_limiter.is_allowed(
    identifier=api_key,
    endpoint="/v1/chat/completions",
    max_requests=100  # 100 requests mỗi phút
)

print(f"Allowed: {info['allowed']}")
print(f"Remaining: {info['remaining']}")
print(f"Reset at: {info['reset_at']}")

2. Token Bucket Algorithm - Phù hợp cho AI API

Token bucket đặc biệt hữu ích khi bạn muốn cho phép burst nhưng vẫn kiểm soát tổng consumption. HolySheep AI Gateway sử dụng thuật toán này cho tier-based rate limiting.

import redis
import time
from dataclasses import dataclass
from typing import Optional

@dataclass
class TokenBucketConfig:
    """Cấu hình cho mỗi tier của user"""
    capacity: int          # Số token tối đa
    refill_rate: float     # Token refill mỗi second
    name: str = ""         # Tên tier (free, pro, enterprise)

class TokenBucketRateLimiter:
    """
    Token Bucket Rate Limiter
    Cho phép burst (tiêu tốn nhiều token cùng lúc)
    Phù hợp cho use case AI với variable-length prompts
    """
    
    # Tier configurations của HolySheep
    TIERS = {
        "free": TokenBucketConfig(capacity=10, refill_rate=0.5, name="Free"),
        "pro": TokenBucketConfig(capacity=100, refill_rate=2.0, name="Pro"),
        "enterprise": TokenBucketConfig(capacity=1000, refill_rate=20.0, name="Enterprise"),
    }
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    def _get_bucket_key(self, user_id: str) -> str:
        return f"token_bucket:{user_id}"
    
    def consume(self, user_id: str, tokens: int = 1, 
                tier: str = "free") -> dict:
        """
        Consume tokens từ bucket
        
        Args:
            user_id: User identifier
            tokens: Số tokens cần consume
            tier: User tier (free/pro/enterprise)
        """
        config = self.TIERS.get(tier, self.TIERS["free"])
        key = self._get_bucket_key(user_id)
        
        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])
        
        -- Lấy current state
        local data = redis.call('HMGET', key, 'tokens', 'last_update')
        local current_tokens = tonumber(data[1]) or capacity
        local last_update = tonumber(data[2]) or now
        
        -- Refill tokens based on time passed
        local elapsed = now - last_update
        local refilled = elapsed * refill_rate
        current_tokens = math.min(capacity, current_tokens + refilled)
        
        -- Check if enough tokens
        local allowed = 0
        if current_tokens >= tokens then
            current_tokens = current_tokens - tokens
            allowed = 1
        end
        
        -- Update bucket state
        redis.call('HMSET', key, 'tokens', current_tokens, 'last_update', now)
        redis.call('EXPIRE', key, 3600)  -- Expire sau 1 hour
        
        return {
            allowed,
            math.floor(current_tokens),
            capacity,
            math.ceil((tokens - current_tokens) / refill_rate) if allowed == 0 and refill_rate > 0 then 0 else 0
        }
        """
        
        result = self.redis.eval(
            lua_script, 1, key,
            config.capacity,
            config.refill_rate,
            tokens,
            time.time()
        )
        
        return {
            "allowed": bool(result[0]),
            "remaining_tokens": result[1],
            "capacity": result[2],
            "retry_after": result[3],
            "tier": tier
        }



=== Sử dụng thực tế ===

redis_client = redis.Redis(host='localhost', port=6379, db=0)
limiter = TokenBucketRateLimiter(redis_client)


Check cho user Pro tier

result = limiter.consume(
    user_id="user_12345",
    tokens=5,  # Prompt dài 5 tokens
    tier="pro"
)

if result["allowed"]:
    print(f"✓ Request được phép. Còn lại: {result['remaining_tokens']} tokens")
else:
    print(f"✗ Rate limit exceeded. Thử lại sau {result['retry_after']}s")

3. Redis Cluster cho High Availability

Để đảm bảo 99.99% uptime, bạn nên sử dụng Redis Cluster với cấu hình sharding.

# docker-compose.yml cho Redis Cluster HA
version: '3.8'

services:
  redis-primary:
    image: redis:7.2-alpine
    command: redis-server --appendonly yes
    volumes:
      - redis_primary_data:/data
    ports:
      - "6379:6379"
    networks:
      - rate_limiter_network
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

  redis-replica:
    image: redis:7.2-alpine
    command: redis-server --replicaof redis-primary 6379 --appendonly yes
    volumes:
      - redis_replica_data:/data
    ports:
      - "6380:6379"
    networks:
      - rate_limiter_network
    depends_on:
      - redis-primary
    healthcheck:
      test: ["CMD", "redis-cli", "-h", "redis-primary", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

  sentinel:
    image: redis:7.2-alpine
    command: |
      redis-sentinel /usr/local/etc/redis/sentinel.conf
    volumes:
      - ./sentinel.conf:/usr/local/etc/redis/sentinel.conf
    ports:
      - "26379:26379"
    networks:
      - rate_limiter_network
    depends_on:
      - redis-primary
      - redis-replica

volumes:
  redis_primary_data:
  redis_replica_data:

networks:
  rate_limiter_network:
    driver: bridge
# Python client với Sentinel support
from redis.sentinel import Sentinel

class HighAvailabilityRateLimiter:
    """Rate Limiter với Redis Sentinel cho failover tự động"""
    
    def __init__(self, sentinel_hosts: list, master_name: str = "mymaster"):
        self.sentinel = Sentinel(
            sentinel_hosts,
            socket_timeout=0.1,
            sentinel_kwargs={'password': 'sentinel_password'}
        )
        self.master_name = master_name
    
    @property
    def master(self):
        """Lấy master connection với automatic failover"""
        return self.sentinel.master_for(
            self.master_name, 
            socket_timeout=0.5,
            redis_class=redis.Redis
        )
    
    @property
    def slave(self):
        """Lấy slave cho read operations (không dùng cho rate limiting)"""
        return self.sentinel.slave_for(
            self.master_name,
            socket_timeout=0.5,
            redis_class=redis.Redis
        )
    
    def check_rate_limit(self, user_id: str, limit: int) -> dict:
        """
        Rate limit check với automatic master/slave switching
        """
        try:
            # Luôn dùng master cho write operations
            client = self.master
            return self._check_and_increment(client, user_id, limit)
        except redis.exceptions.ConnectionError:
            # Fallback: Dùng local counter nếu Redis hoàn toàn down
            return self._local_fallback(user_id, limit)
    
    def _check_and_increment(self, client, user_id: str, limit: int) -> dict:
        key = f"ratelimit:{user_id}"
        pipe = client.pipeline()
        pipe.incr(key)
        pipe.ttl(key)
        results = pipe.execute()
        
        current = results[0]
        ttl = results[1]
        
        # Set expiry nếu key mới tạo
        if ttl == -1:
            client.expire(key, 60)
        
        return {
            "allowed": current <= limit,
            "remaining": max(0, limit - current),
            "current": current,
            "reset_in": ttl if ttl > 0 else 60
        }
    
    def _local_fallback(self, user_id: str, limit: int) -> dict:
        """
        Local fallback khi Redis hoàn toàn unavailable
        ⚠️ Chỉ dùng trong trường hợp emergency
        """
        import threading
        local_counter = {}
        lock = threading.Lock()
        
        with lock:
            current = local_counter.get(user_id, 0) + 1
            local_counter[user_id] = current
        
        return {
            "allowed": current <= limit,
            "remaining": max(0, limit - current),
            "current": current,
            "reset_in": 60,
            "fallback": True  # Cảnh báo: không sync across instances
        }



=== Khởi tạo với Sentinel ===

sentinel_hosts = [
    ('redis-sentinel-1', 26379),
    ('redis-sentinel-2', 26379),
    ('redis-sentinel-3', 26379),
]

ha_limiter = HighAvailabilityRateLimiter(
    sentinel_hosts=sentinel_hosts,
    master_name="mymaster"
)

Tích hợp với HolySheep AI Gateway

Đây là cách tôi đã tích hợp rate limiter vào production system gọi HolySheep AI. Với HolySheep AI, bạn được hưởng mức giá cực kỳ cạnh tranh:

Tất cả thanh toán hỗ trợ WeChat/Alipay, độ trễ trung bình dưới 50ms.

import aiohttp
import asyncio
import redis.asyncio as aioredis
from typing import Optional, Dict, Any
from dataclasses import dataclass

@dataclass
class HolySheepConfig:
    """Cấu hình HolySheep API"""
    base_url: str = "https://api.holysheep.ai/v1"
    api_key: str = "YOUR_HOLYSHEEP_API_KEY"
    default_model: str = "gpt-4.1"
    max_retries: int = 3
    timeout: int = 60

class HolySheepRateLimitedClient:
    """
    HolySheep AI Client với built-in rate limiting
    Sử dụng sliding window + token bucket hybrid approach
    """
    
    def __init__(self, config: HolySheepConfig, 
                 redis_url: str = "redis://localhost:6379"):
        self.config = config
        self.redis = aioredis.from_url(redis_url, decode_responses=True)
        self.rate_limit_key = "holysheep:ratelimit"
        self.requests_per_minute = 60  # Giới hạn request/phút
        self.tokens_per_minute = 100000  # Giới hạn tokens/phút
    
    async def _check_rate_limit(self, user_id: str) -> tuple[bool, dict]:
        """
        Kiểm tra rate limit với sliding window
        Trả về (is_allowed, metadata)
        """
        key = f"{self.rate_limit_key}:{user_id}"
        now = await asyncio.to_thread(time.time)
        window_start = now - 60
        
        # Lua script để atomic check và increment
        script = """
        local key = KEYS[1]
        local now = tonumber(ARGV[1])
        local window_start = tonumber(ARGV[2])
        local limit = tonumber(ARGV[3])
        
        -- Remove old entries
        redis.call('ZREMRANGEBYSCORE', key, 0, window_start)
        
        -- Count current requests
        local current = redis.call('ZCARD', key)
        
        if current < limit then
            redis.call('ZADD', key, now, now .. '-' .. math.random())
            redis.call('EXPIRE', key, 65)
            return {1, limit - current - 1, 0}
        else
            local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
            local retry_after = 0
            if #oldest >= 2 then
                retry_after = math.ceil(oldest[2] + 60 - now)
            end
            return {0, 0, retry_after}
        end
        """
        
        result = await self.redis.eval(
            script, 1, key, now, window_start, self.requests_per_minute
        )
        
        return bool(result[0]), {
            "remaining": result[1],
            "retry_after": result[2],
            "limit": self.requests_per_minute
        }
    
    async def chat_completions(self, messages: list, 
                               model: Optional[str] = None,
                               user_id: str = "anonymous",
                               **kwargs) -> Dict[Any, Any]:
        """
        Gọi HolySheep Chat Completions API với rate limit
        """
        # 1. Check rate limit
        allowed, rate_info = await self._check_rate_limit(user_id)
        
        if not allowed:
            raise RateLimitError(
                f"Rate limit exceeded. Retry after {rate_info['retry_after']}s",
                retry_after=rate_info['retry_after']
            )
        
        # 2. Call HolySheep API
        url = f"{self.config.base_url}/chat/completions"
        headers = {
            "Authorization": f"Bearer {self.config.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": model or self.config.default_model,
            "messages": messages,
            **kwargs
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                url, json=payload, headers=headers,
                timeout=aiohttp.ClientTimeout(total=self.config.timeout)
            ) as response:
                if response.status == 429:
                    retry_after = int(response.headers.get('Retry-After', 60))
                    raise RateLimitError(
                        "HolySheep API rate limit exceeded",
                        retry_after=retry_after
                    )
                
                if response.status != 200:
                    error_text = await response.text()
                    raise APIError(f"API Error: {response.status} - {error_text}")
                
                return await response.json()


class RateLimitError(Exception):
    """Custom exception cho rate limit"""
    def __init__(self, message: str, retry_after: int = 60):
        super().__init__(message)
        self.retry_after = retry_after

class APIError(Exception):
    """Custom exception cho API errors"""
    pass



=== Sử dụng trong production ===

async def main():
    config = HolySheepConfig(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        default_model="deepseek-v3.2"  # Model rẻ nhất, $0.42/1M tokens
    )
    
    client = HolySheepRateLimitedClient(
        config=config,
        redis_url="redis://localhost:6379"
    )
    
    try:
        response = await client.chat_completions(
            messages=[
                {"role": "system", "content": "Bạn là trợ lý AI"},
                {"role": "user", "content": "Giải thích về rate limiting"}
            ],
            user_id="user_pro_123",
            temperature=0.7,
            max_tokens=1000
        )
        print(f"Response: {response['choices'][0]['message']['content']}")
        
    except RateLimitError as e:
        print(f"Rate limited! Retry after {e.retry_after}s")
        await asyncio.sleep(e.retry_after)
    
    except APIError as e:
        print(f"API Error: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Monitoring và Alerting

Để đảm bảo hệ thống hoạt động ổn định, bạn cần monitoring kỹ lưỡng. Đây là setup Prometheus metrics mà đội ngũ HolySheep khuyên dùng:

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time


Define metrics

RATE_LIMIT_REQUESTS_TOTAL = Counter(
    'rate_limit_requests_total',
    'Total number of rate-limited requests',
    ['endpoint', 'status']  # allowed, rejected
)

RATE_LIMIT_LATENCY = Histogram(
    'rate_limit_check_latency_seconds',
    'Latency of rate limit checks',
    buckets=[0.001, 0.005, 0.01, 0.025, 0.05, 0.1]
)

ACTIVE_REQUESTS = Gauge(
    'rate_limit_active_requests',
    'Number of currently active requests'
)

TOKEN_USAGE = Counter(
    'api_token_usage_total',
    'Total tokens used',
    ['model', 'user_tier']
)

class MonitoredRateLimiter:
    """Rate Limiter với Prometheus metrics"""
    
    def __init__(self, base_limiter: SlidingWindowRateLimiter):
        self.base_limiter = base_limiter
    
    def check(self, identifier: str, endpoint: str, 
              max_requests: int) -> dict:
        start_time = time.time()
        ACTIVE_REQUESTS.inc()
        
        try:
            result = self.base_limiter.is_allowed(
                identifier, endpoint, max_requests
            )
            
            status = "allowed" if result[0] else "rejected"
            RATE_LIMIT_REQUESTS_TOTAL.labels(
                endpoint=endpoint,
                status=status
            ).inc()
            
            return result
            
        finally:
            ACTIVE_REQUESTS.dec()
            RATE_LIMIT_LATENCY.observe(time.time() - start_time)



Start Prometheus metrics server

start_http_server(9090)
print("Prometheus metrics available at :9090")

Lỗi thường gặp và cách khắc phục

1. Lỗi: "Connection refused" khi kết nối Redis

Nguyên nhân: Redis server không chạy hoặc firewall block port.

# Cách khắc phục:
import redis
from redis.exceptions import ConnectionError, TimeoutError

def safe_redis_connection(host='localhost', port=6379, db=0):
    """
    Khởi tạo Redis connection với retry logic và timeout
    """
    max_retries = 3
    retry_delay = 1  # second
    
    for attempt in range(max_retries):
        try:
            client = redis.Redis(
                host=host,
                port=port,
                db=db,
                socket_timeout=5,        # Timeout 5 giây
                socket_connect_timeout=5,
                retry_on_timeout=True,   # Retry khi timeout
                health_check_interval=30 # Health check mỗi 30s
            )
            
            # Test connection
            client.ping()
            print(f"✓ Redis connected successfully at {host}:{port}")
            return client
            
        except (ConnectionError, TimeoutError) as e:
            print(f"✗ Attempt {attempt + 1}/{max_retries} failed: {e}")
            
            if attempt < max_retries - 1:
                import time
                time.sleep(retry_delay)
                retry_delay *= 2  # Exponential backoff
            else:
                print("✗ All retry attempts failed. Using in-memory fallback.")
                return create_inmemory_fallback()

def create_inmemory_fallback():
    """
    In-memory rate limiter fallback khi Redis hoàn toàn down
    ⚠️ CHỈ DÙNG TRONG EMERGENCY - không sync across instances!
    """
    import threading
    import time
    from collections import defaultdict
    
    class InMemoryRateLimiter:
        def __init__(self):
            self._lock = threading.Lock()
            self._requests = defaultdict(list)
        
        def is_allowed(self, identifier: str, window: int = 60) -> bool:
            now = time.time()
            window_start = now - window
            
            with self._lock:
                # Remove old requests
                self._requests[identifier] = [
                    t for t in self._requests[identifier]
                    if t > window_start
                ]
                
                if len(self._requests[identifier]) < 60:  # 60 req/min default
                    self._requests[identifier].append(now)
                    return True
                return False
    
    return InMemoryRateLimiter()

2. Lỗi: Rate limit không chính xác do Redis Cluster Rebalancing

Nguyên nhân: Khi Redis Cluster resharding, key có thể di chuyển sang node khác, gây race condition.

# Cách khắc phục: Sử dụng Redlock algorithm hoặc sticky sessions
import redis
import hashlib

class ClusterAwareRateLimiter:
    """
    Rate Limiter tương thích với Redis Cluster
    Sử dụng consistent hashing để đảm bảo key consistency
    """
    
    def __init__(self, cluster_nodes: list):
        self.nodes = [
            redis.Redis(host=node['host'], port=node['port'])
            for node in cluster_nodes
        ]
        self.node_count = len(self.nodes)
    
    def _get_node(self, key: str) -> redis.Redis:
        """Consistent hashing để chọn node"""
        hash_value = int(hashlib.md5(key.encode()).hexdigest(), 16)
        node_index = hash_value % self.node_count
        return self.nodes[node_index]
    
    def check_rate_limit(self, user_id: str, limit: int) -> dict:
        key = f"ratelimit:{user_id}"
        node = self._get_node(key)
        
        pipe = node.pipeline()
        pipe.incr(key)
        pipe.expire(key, 60)
        results = pipe.execute()
        
        current = results[0]
        is_allowed = current <= limit
        
        return {
            "allowed": is_allowed,
            "remaining": max(0, limit - current),
            "current": current,
            "node_id": self.nodes.index(node)
        }
    
    def check_with_sliding_window(self, user_id: str, 
                                  endpoint: str, limit: int) -> dict:
        """
        Sliding window implementation tương thích cluster
        Sử dụng MULTI/EXEC để đảm bảo atomicity
        """
        key = f"ratelimit:sw:{user_id}:{endpoint}"
        node = self._get_node(key)
        now = time.time()
        window_start = now - 60
        
        # Lua script vẫn chạy trên single node
        script = """
        local key = KEYS[1]
        local now = tonumber(ARGV[1])
        local window_start = tonumber(ARGV[2])
        local limit = tonumber(ARGV[3])
        
        redis.call('ZREMRANGEBYSCORE', key, 0, window_start)
        local count = redis.call('ZCARD', key)
        
        if count < limit then
            redis.call('ZADD', key, now, now)
            redis.call('EXPIRE', key, 65)
            return {1, limit - count - 1}
        else
            return {0, 0}
        end
        """
        
        result = node.eval(script, 1, key, now, window_start, limit)
        
        return {
            "allowed": bool(result[0]),
            "remaining": result[1],
            "limit": limit
        }

3. Lỗi: Token count không chính xác với multi-byte characters

Nguyên nhân: Khi sử dụng HolySheep API với tiếng Việt/Trung Quốc/Nhật Bản, token counting thường bị sai vì đếm characters thay vì tokens thực sự.

# Cách khắc phục: Sử dụng tiktoken hoặc gọi API để lấy exact token count
import tiktoken
import re

class TokenCounter:
    """
    Token counter hỗ trợ đa ngôn ngữ
    Sử dụng cl100k_base encoding (GPT-4 compatible)
    """
    
    def __init__(self, encoding_name: str = "cl100k_base"):
        self.encoding = tiktoken.get_encoding(encoding_name)
    
    def count_tokens(self, text: str) -> int:
        """Đếm tokens chính xác cho text"""
        # Encode và decode để split đúng
        tokens = self.encoding.encode(text)
        return len(tokens)
    
    def count_messages_tokens(self, messages: list) -> int:
        """
        Đếm tokens cho chat format
        Format: [role, content, ...] với overhead
        """
        num_tokens = 0
        
        for message in messages:
            # Base overhead cho mỗi message
            num_tokens += 4
            
            # Đếm role
            num_tokens += len(self.encoding.encode(message.get('role', '')))
            
            # Đếm content
            if 'content' in message:
                num_tokens += len(self.encoding.encode(message['content']))
            
            # Đếm name field (nếu có)
            if 'name' in message:
                num_tokens += len(self.encoding.encode(message['name']))
        
        # Overhead cho message array
        num_tokens += 3
        
        return num_tokens
    
    def estimate_cost(self, tokens: int, model: str) -> float:
        """
        Ước tính chi phí dựa trên HolySheep pricing
        Pricing 2026 (USD/1M tokens)
        """
        pricing = {
            "gpt-4.1": 8.0,
            "claude-sonnet-4.5": 15.0,
            "gemini-2.5-flash": 2.50,
            "deepseek-v3.2": 0.42,  # Rẻ nhất!
        }
        
        price_per_million = pricing.get(model, 8.0)
        cost = (tokens / 1_000_000) * price_per_million
        
        return round(cost, 6)



=== Sử dụng trong rate limiter ===

counter = TokenCounter()


Đếm tokens cho request

vietnamese_text = "Xin chào, tôi muốn tìm hiểu về API rate limiting"
tokens = counter.count_tokens(vietnamese_text)
print(f"Tokens: {tokens}")


Tính chi phí

messages = [
    {"role": "system", "content": "Bạn là trợ lý AI hữu ích"},
    {"role": "user", "content": vietnamese_text}
]
total_tokens = counter.count_messages_tokens(messages)
cost = counter.estimate_cost(total_tokens, "deepseek-v3.2")
print(f"Total tokens: {total_tokens}, Estimated cost: ${cost}")

Kết luận

Qua bài viết này, tôi đã chia sẻ những best practices thực chiến về việc triển khai distributed rate limiter với Redis: