Rate Limiting이 왜 필요한가요?

AI API를 서비스에 사용할 때, 비용 관리와 서비스 안정성은 선택이 아닌 필수입니다. Rate Limiting(비율 제한)은 특정 클라이언트가 일정 시간 동안 사용할 수 있는 요청 횟수를 제한하는 기술입니다. 저는 과거에 Rate Limiting 없이 AI API를 운영하다가 한 달에 예상치 못한巨额 비용 청구를 받은 경험이 있습니다. 그때 이후로 Rate Limiting의 중요성을 절실히 깨달렸습니다.

멀티테넌트 환경에서는 여러 고객(테넌트)이 하나의 API 인프라를 공유하기 때문에, 한 고객의 과도한 사용이 다른 고객에게 영향을 미치는 것을 방지해야 합니다. 이 튜토리얼에서는 HolySheep AI 게이트웨이 환경에서 효과적인 Rate Limiting 전략을 단계별로 알아보겠습니다.

Rate Limiting 핵심 전략 4가지

1. 토큰 버킷(Token Bucket) 전략

가장 널리 사용되는 전략입니다. 버킷에 토큰이 채워지고, 각 요청마다 토큰을 소비합니다. 버킷 용량만큼 일시적으로 급증(burst)을 허용하며, 토큰이 소진되면 요청이 거부됩니다. 저는 이 전략을 HolySheep AI와 함께 사용할 것을 권장합니다.


import time
import threading
from collections import defaultdict

class TokenBucketRateLimiter:
    """
    토큰 버킷 기반 Rate Limiter
    HolySheep AI 게이트웨이에서 멀티테넌트 속도 제한에 최적화
    """
    
    def __init__(self, capacity: int, refill_rate: float):
        self.capacity = capacity
        self.refill_rate = refill_rate  # 초당 토큰 회복량
        self.buckets = defaultdict(lambda: {"tokens": capacity, "last_refill": time.time()})
        self.lock = threading.Lock()
    
    def allow_request(self, tenant_id: str, tokens_needed: int = 1) -> tuple[bool, dict]:
        """
        요청 허용 여부 확인
        
        Args:
            tenant_id: 테넌트 고유 식별자
            tokens_needed: 필요한 토큰 수
        
        Returns:
            (허용 여부, 현재 상태 정보)
        """
        with self.lock:
            bucket = self.buckets[tenant_id]
            now = time.time()
            
            # 시간 경과에 따른 토큰 회복
            elapsed = now - bucket["last_refill"]
            bucket["tokens"] = min(
                self.capacity,
                bucket["tokens"] + elapsed * self.refill_rate
            )
            bucket["last_refill"] = now
            
            # 토큰 충분 여부 확인
            if bucket["tokens"] >= tokens_needed:
                bucket["tokens"] -= tokens_needed
                return True, {
                    "allowed": True,
                    "remaining_tokens": bucket["tokens"],
                    "retry_after": 0
                }
            else:
                retry_after = (tokens_needed - bucket["tokens"]) / self.refill_rate
                return False, {
                    "allowed": False,
                    "remaining_tokens": bucket["tokens"],
                    "retry_after": round(retry_after, 2)
                }


HolySheep AI 게이트웨이용 Rate Limiter 인스턴스


각 테넌트: 분당 60회 요청, 버스트 허용: 10회

limiter = TokenBucketRateLimiter(
    capacity=10,      # 버스트 시 최대 10회 추가 요청 가능
    refill_rate=1.0   # 초당 1토큰 회복 (분당 60회)
)


HolySheep API 키로 요청 시뮬레이션

def call_holysheep_api(tenant_id: str, api_key: str):
    allowed, status = limiter.allow_request(tenant_id)
    
    if allowed:
        print(f"✅ [{tenant_id}] 요청 허용 - 남은 토큰: {status['remaining_tokens']:.1f}")
        # HolySheep AI API 호출 (base_url 사용 필수)
        # base_url = "https://api.holysheep.ai/v1"
        # requests.post(f"{base_url}/chat/completions", headers={"Authorization": f"Bearer {api_key}"})
        return True
    else:
        print(f"⛔ [{tenant_id}] 요청 거부 - {status['retry_after']}초 후 재시도")
        return False


테스트

for i in range(12):
    call_holysheep_api("tenant_001", "YOUR_HOLYSHEEP_API_KEY")
    time.sleep(0.1)

2. 고정 윈도우(Fixed Window) 전략

단순하고 구현이 쉽습니다. 시간을 고정된 윈도우로 나누고, 각 윈도우마다 요청 횟수를 카운트합니다. Redis와 함께 사용하여 분산 환경에서도 작동합니다.


import time
import redis
import json
from typing import Optional

class FixedWindowRateLimiter:
    """
    고정 윈도우 기반 Rate Limiter
    HolySheep AI 게이트웨이 연동을 위한 Redis 기반 구현
    """
    
    def __init__(self, redis_url: str, window_seconds: int = 60, max_requests: int = 60):
        self.redis_client = redis.from_url(redis_url)
        self.window_seconds = window_seconds
        self.max_requests = max_requests
    
    def _get_window_key(self, tenant_id: str) -> str:
        """현재 윈도우에 해당하는 Redis 키 생성"""
        current_window = int(time.time() // self.window_seconds)
        return f"rate_limit:{tenant_id}:{current_window}"
    
    def check_rate_limit(
        self, 
        tenant_id: str, 
        api_key: str,
        model: str = "gpt-4.1"
    ) -> dict:
        """
        HolySheep AI API 요청 전 Rate Limit 확인
        
        Returns:
            {"allowed": bool, "headers": dict, "error": Optional[str]}
        """
        key = self._get_window_key(tenant_id)
        
        # Redis 트랜잭션으로 원자적 증가
        pipe = self.redis_client.pipeline()
        pipe.incr(key)
        pipe.expire(key, self.window_seconds * 2)  # 윈도우 종료 후 정리
        results = pipe.execute()
        
        current_count = results[0]
        
        # HolySheep AI Rate Limit 헤더 구성
        headers = {
            "X-RateLimit-Limit": str(self.max_requests),
            "X-RateLimit-Remaining": str(max(0, self.max_requests - current_count)),
            "X-RateLimit-Reset": str(int(time.time()) + self.window_seconds),
            "X-RateLimit-Window": str(self.window_seconds)
        }
        
        if current_count <= self.max_requests:
            return {
                "allowed": True,
                "headers": headers,
                "usage": current_count
            }
        else:
            # Rate Limit 초과 - HolySheep 에러 응답 형식
            return {
                "allowed": False,
                "headers": headers,
                "error": "rate_limit_exceeded",
                "retry_after": self.window_seconds - (int(time.time()) % self.window_seconds)
            }


사용 예시

def safe_holysheep_call(tenant_id: str, api_key: str, prompt: str):
    limiter = FixedWindowRateLimiter(
        redis_url="redis://localhost:6379",
        window_seconds=60,
        max_requests=60  # 분당 60회 (HolySheep 기본 플랜)
    )
    
    result = limiter.check_rate_limit(tenant_id, api_key)
    
    if result["allowed"]:
        # HolySheep AI API 실제 호출
        import requests
        response = requests.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
                **result["headers"]
            },
            json={
                "model": "gpt-4.1",
                "messages": [{"role": "user", "content": prompt}]
            }
        )
        return response.json()
    else:
        raise Exception(f"Rate Limit 초과. {result['retry_after']}초 후 재시도하세요.")

3. 슬라이딩 윈도우 로그(Sliding Window Log) 전략

가장 정교한 전략입니다. 각 요청의 타임스탬프를 기록하여 정확한 비율을 계산합니다. 토큰 버킷보다 정확한 제한이 가능하지만, 메모리 사용량이 높습니다.


import time
import bisect
from threading import Lock
from typing import List, Tuple

class SlidingWindowLogRateLimiter:
    """
    슬라이딩 윈도우 로그 기반 Rate Limiter
    HolySheep AI 멀티테넌트 환경에서 정확한 비율 제한
    """
    
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.tenant_logs: dict[str, List[float]] = {}
        self.lock = Lock()
    
    def _clean_old_logs(self, tenant_id: str, current_time: float) -> None:
        """윈도우 밖의 오래된 로그 제거"""
        logs = self.tenant_logs.get(tenant_id, [])
        cutoff_time = current_time - self.window_seconds
        self.tenant_logs[tenant_id] = [t for t in logs if t > cutoff_time]
    
    def is_allowed(self, tenant_id: str) -> Tuple[bool, dict]:
        """
        요청 허용 여부 및 상세 정보 반환
        
        Returns:
            (allowed, {"remaining": int, "reset_in": float, "total_allowed": int})
        """
        current_time = time.time()
        
        with self.lock:
            self._clean_old_logs(tenant_id, current_time)
            
            logs = self.tenant_logs.get(tenant_id, [])
            
            if len(logs) < self.max_requests:
                # 요청 허용 및 타임스탬프 기록
                logs.append(current_time)
                self.tenant_logs[tenant_id] = logs
                
                reset_in = self.window_seconds - (current_time - logs[0])
                
                return True, {
                    "allowed": True,
                    "remaining": self.max_requests - len(logs),
                    "reset_in": round(reset_in, 2),
                    "total_allowed": self.max_requests
                }
            else:
                # 요청 거부 - 가장 오래된 요청이 만료되는 시간 계산
                oldest_request = logs[0]
                reset_in = (oldest_request + self.window_seconds) - current_time
                
                return False, {
                    "allowed": False,
                    "remaining": 0,
                    "reset_in": round(max(0, reset_in), 2),
                    "total_allowed": self.max_requests
                }
    
    def get_retry_after(self, tenant_id: str) -> float:
        """다음 요청까지 필요한 대기 시간(초)"""
        logs = self.tenant_logs.get(tenant_id, [])
        if len(logs) < self.max_requests:
            return 0
        
        current_time = time.time()
        oldest = logs[0]
        return max(0, (oldest + self.window_seconds) - current_time)


HolySheep AI tier별 Rate Limit 설정

RATE_LIMITS = {
    "free": {"max_requests": 20, "window_seconds": 60},
    "starter": {"max_requests": 100, "window_seconds": 60},
    "pro": {"max_requests": 500, "window_seconds": 60},
    "enterprise": {"max_requests": 5000, "window_seconds": 60}
}

def get_tenant_rate_limit(tenant_id: str, tier: str) -> SlidingWindowLogRateLimiter:
    """테넌트의 요금제에 맞는 Rate Limiter 반환"""
    config = RATE_LIMITS.get(tier, RATE_LIMITS["free"])
    return SlidingWindowLogRateLimiter(**config)


실행 예시

limiter = get_tenant_rate_limit("tenant_abc123", "pro")

for i in range(505):
    allowed, info = limiter.is_allowed(f"tenant_abc123")
    if i % 50 == 0:
        status = "✅" if allowed else "⛔"
        print(f"{status} 요청 #{i+1}: 남은 요청={info['remaining']}, 재시작={info['reset_in']}초")
    if not allowed:
        break

멀티테넌트 Rate Limiting 아키텍처

HolySheep AI 게이트웨이에서 멀티테넌트 Rate Limiting을 구현할 때는 계층적 제한을 고려해야 합니다. 저는 실제 운영에서 세 가지 수준의 제한을 동시에 적용하여 서비스 안정성을 확보했습니다.


from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import time
import hashlib

class TenantTier(Enum):
    FREE = "free"
    STARTER = "starter"
    PRO = "pro"
    ENTERPRISE = "enterprise"

@dataclass
class RateLimitConfig:
    """테넌트별 Rate Limit 설정"""
    requests_per_minute: int
    tokens_per_minute: int  # 입력 + 출력 토큰 합산
    concurrent_requests: int
    burst_size: int


HolySheep AI 지원 모델별 토큰 비용 ($/1M 토큰)

MODEL_COSTS = {
    "gpt-4.1": 8.0,
    "claude-sonnet-4-5": 15.0,
    "gemini-2.5-flash": 2.5,
    "deepseek-v3.2": 0.42
}


티어별 기본 Rate Limit

TIER_CONFIGS = {
    TenantTier.FREE: RateLimitConfig(
        requests_per_minute=20,
        tokens_per_minute=100000,
        concurrent_requests=2,
        burst_size=5
    ),
    TenantTier.STARTER: RateLimitConfig(
        requests_per_minute=100,
        tokens_per_minute=500000,
        concurrent_requests=5,
        burst_size=15
    ),
    TenantTier.PRO: RateLimitConfig(
        requests_per_minute=500,
        tokens_per_minute=2000000,
        concurrent_requests=20,
        burst_size=50
    ),
    TenantTier.ENTERPRISE: RateLimitConfig(
        requests_per_minute=5000,
        tokens_per_minute=10000000,
        concurrent_requests=100,
        burst_size=200
    )
}

@dataclass
class RateLimitResult:
    """Rate Limit 체크 결과"""
    allowed: bool
    reason: Optional[str] = None
    retry_after: float = 0
    limits: dict = field(default_factory=dict)

class MultiTenantRateLimiter:
    """
    HolySheep AI 멀티테넌트 게이트웨이용 Rate Limiter
    계층적 Rate Limiting 지원
    """
    
    def __init__(self):
        # Redis를 사용한 분산 Rate Limit 스토어
        # self.redis = redis.from_url("redis://localhost:6379")
        self.tenant_locks: dict[str, Lock] = {}
        self.lock_creation_lock = Lock()
    
    def _get_tenant_lock(self, tenant_id: str) -> Lock:
        """테넌트별 Lock 반환 (동시 요청 처리)"""
        if tenant_id not in self.tenant_locks:
            with self.lock_creation_lock:
                if tenant_id not in self.tenant_locks:
                    self.tenant_locks[tenant_id] = Lock()
        return self.tenant_locks[tenant_id]
    
    def check_limits(
        self,
        tenant_id: str,
        tier: TenantTier,
        model: str,
        estimated_tokens: int
    ) -> RateLimitResult:
        """
        모든 수준의 Rate Limit 체크
        
        Returns:
            RateLimitResult: 허용/거부 결과 및 상세 정보
        """
        config = TIER_CONFIGS[tier]
        now = time.time()
        
        # 1. 동시 요청 수 체크
        concurrent_key = f"concurrent:{tenant_id}"
        # current_concurrent = self.redis.incr(concurrent_key)
        # self.redis.expire(concurrent_key, 60)
        current_concurrent = 1  # 시뮬레이션
        
        if current_concurrent > config.concurrent_requests:
            return RateLimitResult(
                allowed=False,
                reason="concurrent_limit_exceeded",
                retry_after=1.0,
                limits={"concurrent": {"current": current_concurrent, "max": config.concurrent_requests}}
            )
        
        # 2. 분당 요청 수 체크 (토큰 버킷)
        rpm_key = f"rpm:{tenant_id}"
        # current_rpm = self.redis.get(rpm_key)
        current_rpm = 0  # 시뮬레이션
        
        if current_rpm >= config.requests_per_minute:
            return RateLimitResult(
                allowed=False,
                reason="rpm_limit_exceeded",
                retry_after=60.0,
                limits={"rpm": {"current": current_rpm, "max": config.requests_per_minute}}
            )
        
        # 3. 분당 토큰 수 체크
        tpm_key = f"tpm:{tenant_id}"
        # current_tpm = self.redis.get(tpm_key)
        current_tpm = 0  # 시뮬레이션
        
        if current_tpm + estimated_tokens > config.tokens_per_minute:
            return RateLimitResult(
                allowed=False,
                reason="tpm_limit_exceeded",
                retry_after=60.0,
                limits={"tpm": {"current": current_tpm, "max": config.tokens_per_minute, "requested": estimated_tokens}}
            )
        
        # 4. 모델별 글로벌 제한 체크 (HolySheep API 레벨)
        model_key = f"model_global:{model}"
        # model_usage = self.redis.incr(model_key)
        # self.redis.expire(model_key, 60)
        model_usage = 0  # 시뮬레이션
        model_global_limit = 10000  # 분당 글로벌 제한
        
        if model_usage > model_global_limit:
            return RateLimitResult(
                allowed=False,
                reason="model_global_limit_exceeded",
                retry_after=60.0,
                limits={"model": {"current": model_usage, "max": model_global_limit}}
            )
        
        # 모든 체크 통과
        return RateLimitResult(
            allowed=True,
            limits={
                "rpm": {"remaining": config.requests_per_minute - current_rpm - 1},
                "tpm": {"remaining": config.tokens_per_minute - current_tpm - estimated_tokens},
                "concurrent": {"remaining": config.concurrent_requests - current_concurrent}
            }
        )
    
    def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        """요청 비용 계산 (HolySheep AI 요금제)"""
        cost_per_million = MODEL_COSTS.get(model, 8.0)
        total_tokens = input_tokens + output_tokens
        return (total_tokens / 1_000_000) * cost_per_million


사용 예시

limiter = MultiTenantRateLimiter()


Pro 티어 테넌트의 GPT-4.1 요청 체크

result = limiter.check_limits(
    tenant_id="tenant_pro_001",
    tier=TenantTier.PRO,
    model="gpt-4.1",
    estimated_tokens=2000
)

if result.allowed:
    print("✅ 요청 허용됨")
    print(f"   예상 비용: ${limiter.calculate_cost('gpt-4.1', 1000, 1000):.4f}")
    print(f"   남은 RPM: {result.limits['rpm']['remaining']}")
else:
    print(f"⛔ 요청 거부: {result.reason}")
    print(f"   {result.retry_after}초 후 재시도")

HolySheep AI Rate Limiting 설정

HolySheep AI는 기본적으로 Rate Limiting을 제공하지만, 커스텀 제한을 설정하여 더 세밀하게 제어할 수 있습니다. HolySheep 대시보드에서 각 테넌트별 Rate Limit을 조정하고, API를 통해 프로그래밍적으로 관리할 수 있습니다.


import requests
from typing import Dict, Optional

class HolySheepRateLimitManager:
    """
    HolySheep AI API를 통한 Rate Limit 관리
    base_url: https://api.holysheep.ai/v1
    """
    
    def __init__(self, admin_api_key: str):
        self.base_url = "https://api.holysheep.ai/v1"
        self.headers = {
            "Authorization": f"Bearer {admin_api_key}",
            "Content-Type": "application/json"
        }
    
    def get_tenant_usage(self, tenant_id: str) -> Dict:
        """테넌트의 현재 사용량 확인"""
        response = requests.get(
            f"{self.base_url}/admin/tenants/{tenant_id}/usage",
            headers=self.headers
        )
        response.raise_for_status()
        return response.json()
    
    def set_tenant_rate_limit(
        self,
        tenant_id: str,
        requests_per_minute: int,
        tokens_per_minute: int
    ) -> Dict:
        """테넌트의 Rate Limit 설정"""
        response = requests.patch(
            f"{self.base_url}/admin/tenants/{tenant_id}/rate-limit",
            headers=self.headers,
            json={
                "requests_per_minute": requests_per_minute,
                "tokens_per_minute": tokens_per_minute
            }
        )
        response.raise_for_status()
        return response.json()
    
    def get_rate_limit_status(self, api_key: str) -> Dict:
        """현재 API 키의 Rate Limit 상태 확인"""
        response = requests.get(
            f"{self.base_url}/rate-limit/status",
            headers={"Authorization": f"Bearer {api_key}"}
        )
        return {
            "limit": response.headers.get("X-RateLimit-Limit"),
            "remaining": response.headers.get("X-RateLimit-Remaining"),
            "reset": response.headers.get("X-RateLimit-Reset")
        }


HolySheep AI Rate Limit 응답 헤더 파싱

def parse_rate_limit_headers(response: requests.Response) -> Dict:
    """HolySheep API 응답의 Rate Limit 헤더 파싱"""
    return {
        "limit": int(response.headers.get("X-RateLimit-Limit", 0)),
        "remaining": int(response.headers.get("X-RateLimit-Remaining", 0)),
        "reset_timestamp": int(response.headers.get("X-RateLimit-Reset", 0)),
        "reset_in_seconds": max(0, 
            int(response.headers.get("X-RateLimit-Reset", 0)) - int(time.time())
        )
    }


Rate Limit 초과 시 재시도 로직

def exponential_backoff_retry(
    func,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0
):
    """지수적 백오프를 통한 재시도 로직"""
    for attempt in range(max_retries):
        try:
            response = func()
            
            if response.status_code == 429:  # Rate Limit 초과
                headers = parse_rate_limit_headers(response)
                wait_time = headers.get("reset_in_seconds", base_delay * (2 ** attempt))
                wait_time = min(wait_time, max_delay)
                
                print(f"Rate Limit 초과. {wait_time:.1f}초 후 재시도 ({attempt + 1}/{max_retries})")
                time.sleep(wait_time)
                continue
            
            return response
            
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            print(f"요청 실패: {e}. {delay:.1f}초 후 재시도...")
            time.sleep(delay)
    
    raise Exception(f"최대 재시회수({max_retries}) 초과")


사용 예시

manager = HolySheepRateLimitManager("YOUR_ADMIN_API_KEY")


특정 테넌트의 Rate Limit 설정

result = manager.set_tenant_rate_limit(
    tenant_id="tenant_premium_001",
    requests_per_minute=500,
    tokens_per_minute=2000000
)
print(f"Rate Limit 설정 완료: {result}")

자주 발생하는 오류와 해결책

오류 1: 429 Too Many Requests 응답

가장 흔한 오류입니다. Rate Limit에 도달하면 HolySheep AI는 429 상태 코드와 함께 재시도 시간 정보를 반환합니다.



❌ 잘못된 접근: 즉시 재시도

def bad_retry_request(api_key: str, prompt: str):
    for _ in range(10):
        response = requests.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={"Authorization": f"Bearer {api_key}"},
            json={"model": "gpt-4.1", "messages": [{"role": "user", "content": prompt}]}
        )
        if response.status_code != 429:
            return response.json()
        # 지수적 백오프 없이 즉시 재시도 → 더 많은 429 발생


✅ 올바른 접근: Retry-After 헤더 활용

def good_retry_request(api_key: str, prompt: str, max_retries: int = 5):
    for attempt in range(max_retries):
        response = requests.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={"Authorization": f"Bearer {api_key}"},
            json={"model": "gpt-4.1", "messages": [{"role": "user", "content": prompt}]}
        )
        
        if response.status_code == 200:
            return response.json()
        
        if response.status_code == 429:
            # Retry-After 헤더가 있으면 해당 값 사용, 없으면 지수 백오프
            retry_after = response.headers.get("Retry-After")
            if retry_after:
                wait_time = int(retry_after)
            else:
                wait_time = min(2 ** attempt, 60)  # 최대 60초
            
            print(f"Rate Limit 초과. {wait_time}초 대기 후 재시도...")
            time.sleep(wait_time)
            continue
        
        # 다른 오류는 즉시 실패
        response.raise_for_status()
    
    raise Exception("최대 재시도 횟수 초과")

오류 2: 토큰 제한 초과로 인한 불완전한 응답

입력 또는 출력 토큰 제한에 도달하면 응답이 잘려서 반환될 수 있습니다. 이 경우 응답의 usage 정보를 확인해야 합니다.



✅ 토큰 제한 초과 처리

def handle_token_limit_error(api_key: str, prompt: str, max_tokens: int = 4000):
    response = requests.post(
        "https://api.holysheep.ai/v1/chat/completions",
        headers={"Authorization": f"Bearer {api_key}"},
        json={
            "model": "gpt-4.1",
            "messages": [{"role": "user", "content": prompt}],
            "max_tokens": max_tokens
        }
    )
    
    if response.status_code == 200:
        data = response.json()
        usage = data.get("usage", {})
        
        # 응답이 잘렸는지 확인
        if data["choices"][0]["finish_reason"] == "length":
            print("⚠️ 응답이 토큰 제한으로 인해 잘렸습니다.")
            print(f"   입력 토큰: {usage.get('prompt_tokens', 0)}")
            print(f"   출력 토큰: {usage.get('completion_tokens', 0)} (최대: {max_tokens})")
            
            # 다음 요청에서 max_tokens 증가 또는 프롬프트 단축
            return {
                "complete": False,
                "partial_content": data["choices"][0]["message"]["content"],
                "suggested_action": "increase_max_tokens_or_shorten_prompt"
            }
        
        return {"complete": True, "content": data["choices"][0]["message"]["content"]}
    
    response.raise_for_status()

오류 3: 동시 요청으로 인한 Race Condition

분산 환경에서 여러 인스턴스가 동시에 Rate Limit을 체크할 때 발생하는 문제입니다. Redis Lua 스크립트를 사용하여 원자적 연산을 해야 합니다.



❌ Race Condition 발생 가능 코드

def bad_concurrent_check(tenant_id: str, redis_client):
    current = redis_client.get(f"rpm:{tenant_id}")
    if int(current or 0) >= 60:
        return False  # 제한 초과
    redis_client.incr(f"rpm:{tenant_id}")  # 다른 인스턴스가 먼저 증가시킬 수 있음
    return True


✅ Lua 스크립트로 원자적 체크-인크리먼트

ATOMIC_CHECK_SCRIPT = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])

local current = tonumber(redis.call('GET', key) or '0')

if current >= limit then
    return {0, current, limit}
end

local new_count = redis.call('INCR', key)
if new_count == 1 then
    redis.call('EXPIRE', key, window)
end

return {1, new_count, limit}
"""

def good_concurrent_check(tenant_id: str, redis_client, limit: int = 60, window: int = 60):
    """
    Lua 스크립트로 원자적 Rate Limit 체크
    """
    result = redis_client.eval(
        ATOMIC_CHECK_SCRIPT,
        1,  # KEYS 개수
        f"rpm:{tenant_id}",  # key
        str(limit),  # ARGV[1]
        str(window)  # ARGV[2]
    )
    
    allowed = bool(result[0])
    current = int(result[1])
    max_limit = int(result[2])
    
    return {
        "allowed": allowed,
        "current": current,
        "limit": max_limit,
        "remaining": max(0, max_limit - current)
    }


Redis 클라이언트 설정

import redis
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
redis_client.register_script(ATOMIC_CHECK_SCRIPT)


동시 요청 테스트

import threading

def test_concurrent():
    results = []
    def make_request(tenant_id):
        result = good_concurrent_check(tenant_id, redis_client, limit=10, window=60)
        results.append(result["allowed"])
    
    threads = [threading.Thread(target=make_request, args=("test_tenant",)) for _ in range(15)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    
    allowed_count = sum(results)
    print(f"15개 동시 요청 중 {allowed_count}개 허용됨 (제한: 10)")
    assert allowed_count == 10, f"Expected 10, got {allowed_count}"

test_concurrent()

Rate Limiting 전략 비교

전략 정확성 메모리 효율 구현 난이도 버스트 허용 권장 상황
토큰 버킷 보통 ⭐⭐⭐⭐⭐ 쉬움 높음 대부분의 AI API 사용场景
고정 윈도우 낮음 ⭐⭐⭐⭐⭐ 쉬움 제한적 단순한 Rate Limit 필요 시
슬라이딩 윈도우 로그 높음 ⭐⭐ 보통 없음 정확한 비율 제한 필요 시
슬라이딩 윈도우 카운터 보통 ⭐⭐⭐⭐ 보통 제한적 메모리와 정확성의 균형

이런 팀에 적합 / 비적합

✅ 이런 팀에 적합

❌ 이런 팀에는 비적합