AI API 게이트웨이 Rate Limiting 구현 완벽 가이드

슬라이딩 윈도우 카운터 구현

더 정밀한 Rate Limiting이 필요한 경우滑动 윈도우 카운터를 구현합니다. HolySheep AI의 다중 모델 환경에서 모델별 개별 제한에 적합합니다.

"""
HolySheep AI 슬라이딩 윈도우 Rate Limiter
Redis Sorted Set을 사용한 정확한 시간 기반 Rate Limiting
"""

import redis
import time
import json
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timedelta

class SlidingWindowRateLimiter:
    """
    HolySheep AI Multi-Model Rate Limiter
    모델별 독립적인 Rate Limit 관리
    """
    
    # 모델별 기본 RPM/TPM limits
    MODEL_LIMITS = {
        "gpt-4.1": {"rpm": 500, "tpm": 150000},
        "claude-sonnet-4": {"rpm": 400, "tpm": 120000},
        "gemini-2.5-flash": {"rpm": 1000, "tpm": 500000},
        "deepseek-v3": {"rpm": 600, "tpm": 200000}
    }
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis = redis.from_url(redis_url, decode_responses=True)
    
    def _get_window_key(self, api_key: str, model: str, window_type: str) -> str:
        """윈도우 키 생성"""
        return f"sw:{api_key}:{model}:{window_type}"
    
    def check_and_update(
        self,
        api_key: str,
        model: str,
        token_count: int = 1,
        window_seconds: int = 60
    ) -> Tuple[bool, Dict]:
        """
        슬라이딩 윈도우 Rate Limit 확인 및 업데이트
        
        Args:
            api_key: HolySheep API 키
            model: 모델명
            token_count: 소비된 토큰 수
            window_seconds: 윈도우 크기 (기본 60초)
        
        Returns:
            (allowed: bool, status: dict)
        """
        now = time.time()
        window_start = now - window_seconds
        
        rpm_key = self._get_window_key(api_key, model, "rpm")
        tpm_key = self._get_window_key(api_key, model, "tpm")
        
        limits = self.MODEL_LIMITS.get(model, {"rpm": 500, "tpm": 100000})
        
        # Lua 스크립트: 원자적 연산으로 슬라이딩 윈도우 업데이트
        lua_script = """
        local key = KEYS[1]
        local window_start = tonumber(ARGV[1])
        local now = tonumber(ARGV[2])
        local limit = tonumber(ARGV[3])
        local count = tonumber(ARGV[4])
        
        -- 윈도우 이전 레코드 제거
        redis.call('ZREMRANGEBYSCORE', key, '-inf', window_start)
        
        -- 현재 윈도우 카운트
        local current_count = redis.call('ZCARD', key)
        
        -- 허용 여부 확인
        local allowed = 0
        if current_count + count <= limit then
            allowed = 1
            -- 새 요청 추가
            redis.call('ZADD', key, now, now .. ':' .. math.random())
        end
        
        -- TTL 설정
        redis.call('EXPIRE', key, window_seconds * 2)
        
        return {allowed, current_count, limit - current_count}
        """
        
        # RPM 체크
        rpm_result = self.redis.eval(
            lua_script, 1, rpm_key, window_start, now, limits["rpm"], 1
        )
        rpm_allowed, rpm_used, rpm_remaining = rpm_result
        
        # TPM 체크
        tpm_result = self.redis.eval(
            lua_script, 1, tpm_key, window_start, now, limits["tpm"], token_count
        )
        tpm_allowed, tpm_used, tpm_remaining = tpm_result
        
        overall_allowed = bool(rpm_allowed) and bool(tpm_allowed)
        
        return overall_allowed, {
            "allowed": overall_allowed,
            "rpm": {"used": rpm_used, "remaining": rpm_remaining, "limit": limits["rpm"]},
            "tpm": {"used": tpm_used, "remaining": tpm_remaining, "limit": limits["tpm"]},
            "reset_in": window_seconds,
            "model": model
        }
    
    def get_usage_stats(self, api_key: str, model: str) -> Dict:
        """현재 사용량 통계 조회"""
        now = time.time()
        window_start = now - 60
        
        rpm_key = self._get_window_key(api_key, model, "rpm")
        tpm_key = self._get_window_key(api_key, model, "tpm")
        
        # 만료된 레코드 제거 후 카운트
        self.redis.zremrangebyscore(rpm_key, '-inf', window_start)
        self.redis.zremrangebyscore(tpm_key, '-inf', window_start)
        
        limits = self.MODEL_LIMITS.get(model, {"rpm": 500, "tpm": 100000})
        
        rpm_count = self.redis.zcard(rpm_key)
        tpm_count = self.redis.zcard(tpm_key)
        
        return {
            "model": model,
            "rpm": {"current": rpm_count, "limit": limits["rpm"], "utilization": f"{rpm_count/limits['rpm']*100:.1f}%"},
            "tpm": {"current": tpm_count, "limit": limits["tpm"], "utilization": f"{tpm_count/limits['tpm']*100:.1f}%"},
            "timestamp": datetime.now().isoformat()
        }


HolySheep AI 다중 모델 라우팅과 통합
class HolySheepAPIClient:
    """
    HolySheep AI API Client with built-in Rate Limiting
    """
    
    def __init__(self, api_key: str, redis_url: str = "redis://localhost:6379"):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.limiter = SlidingWindowRateLimiter(redis_url)
        self.session = None
    
    async def chat_completions(
        self,
        model: str,
        messages: List[Dict],
        max_tokens: int = 1000,
        **kwargs
    ):
        """Rate Limited chat completions API"""
        import aiohttp
        
        # 토큰 예상치 계산
        estimated_tokens = sum(
            len(str(m.get("content", ""))) // 4 for m in messages
        ) + max_tokens
        
        # Rate limit 확인
        allowed, status = self.limiter.check_and_update(
            self.api_key, model, estimated_tokens
        )
        
        if not allowed:
            raise RateLimitError(
                f"Rate limit exceeded for {model}",
                status=status
            )
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": messages,
            "max_tokens": max_tokens,
            **kwargs
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload
            ) as response:
                result = await response.json()
                return result


class RateLimitError(Exception):
    """Rate Limit 초과 에러"""
    def __init__(self, message: str, status: dict):
        super().__init__(message)
        self.status = status

# 잘못된 접근: 단순 재시도
response = requests.post(url, json=data)
if response.status_code == 429:
    time.sleep(5)  # 비효율적 고정 대기
    response = requests.post(url, json=data)

올바른 접근: 지수 백오프와 Retry-After 헤더 활용
def robust_request_with_retry(url, headers, payload, max_retries=5):
    """HolySheep AI API용 견고한 재시도 로직"""
    import time
    import requests
    
    for attempt in range(max_retries):
        response = requests.post(url, headers=headers, json=payload)
        
        if response.status_code == 200:
            return response.json()
        
        if response.status_code == 429:
            # Retry-After 헤더에서 대기 시간 추출
            retry_after = int(response.headers.get('Retry-After', 1))
            
            # 지수 백오프 적용
            wait_time = min(retry_after, (2 ** attempt) + random.uniform(0, 1))
            print(f"Rate limited. Waiting {wait_time:.2f}s (attempt {attempt + 1}/{max_retries})")
            time.sleep(wait_time)
        else:
            response.raise_for_status()
    
    raise Exception(f"Failed after {max_retries} retries")

# 문제: 동시성이 높을 때 Redis 연산 충돌
async def bad_concurrent_handler():
    # Race condition 발생 가능
    current = await redis.get("count")
    new = int(current) + 1
    await redis.set("count", new)  # 다른 요청이 먼저 실행될 수 있음

해결: Redis Lua 스크립트로 원자적 연산
async def correct_concurrent_handler():
    lua_script = """
    local current = tonumber(redis.call('GET', KEYS[1]) or 0)
    local limit = tonumber(ARGV[1])
    
    if current >= limit then
        return 0  -- 제한 초과
    end
    
    redis.call('INCR', KEYS[1])
    redis.call('EXPIRE', KEYS[1], 60)
    return 1  -- 성공
    """
    
    result = await redis.eval(lua_script, 1, "ratelimit:counter", 100)
    if not result:
        raise RateLimitExceededError()
    return True

# 문제: 만료되지 않는 키 누적
HolySheep AI 다중 모델 환경에서 수백만 개의 만료된 키가 쌓임

해결 1:定期적인 키 정리
async def cleanup_expired_keys(redis_client, pattern="sw:*", max_age_seconds=120):
    """만료된 슬라이딩 윈도우 키 정리"""
    cursor = 0
    deleted = 0
    
    while True:
        cursor, keys = await redis_client.scan(cursor, match=pattern, count=1000)
        
        for key in keys:
            # 스코어(시간戳)로 만료 여부 확인
            oldest = await redis_client.zrange(key, 0, 0, withscores=True)
            if oldest:
                timestamp = oldest[0][1]
                if time.time() - timestamp > max_age_seconds:
                    await redis_client.delete(key)
                    deleted += 1
        
        if cursor == 0:
            break
    
    return deleted

해결 2: 자동 TTL 설정
async def safe_key_creation(redis_client, key, value, ttl_seconds=120):
    """항상 TTL이 설정된 안전한 키 생성"""
    pipe = redis_client.pipeline()
    pipe.set(key, value)
    pipe.expire(key, ttl_seconds)
    await pipe.execute()

# 문제: HolySheep AI는 OpenAI 호환 형식 사용
잘못된 응답 처리
if "choices" in response and response["choices"][0]["message"]["content"]:
    # 정상 처리
    
올바른 접근: 에러 응답 포함 처리
def handle_holysheep_response(response):
    """HolySheep AI 응답 완전한 처리"""
    if "error" in response:
        error = response["error"]
        if error.get("type") == "rate_limit_exceeded":
            # Rate limit 정보 파싱
            return {
                "type": "rate_limited",
                "retry_after": error.get("param", 60),
                "model": error.get("model", "unknown")
            }
        raise HolySheepAPIError(error)
    
    # 정상 응답 처리
    return {
        "content": response["choices"][0]["message"]["content"],
        "usage": response.get("usage", {}),
        "model": response.get("model")
    }