AI API를 활용한 대규모 서비스를 운영하면서 가장 많이 마주치는 문제 중 하나가 바로 Rate Limiting입니다. 저는 HolySheep AI에서 수백 개의 개발팀과 협업하며rate limit 초과로 인한 서비스 장애를 가장 효과적으로 해결하는 방법을 정리했습니다. 이 튜토리얼에서는 분산 환경에서 안정적으로 동작하는 Token Bucket 알고리즘의 원리부터 실제 구현 코드까지 단계별로 설명드리겠습니다.
핵심 결론: 왜 Token Bucket인가?
- burst traffic 처리: 단기간에 몰리는 요청을 버킷의 토큰으로 흡수하여 서비스 안정성 확보
- 平滑한 트래픽 분배:-leaky bucket과 달리 요청 간격을 균등하게 유지하면서도 burst 허용
- 분산 환경 호환: Redis 기반 Lua 스크립트로 원자적 연산 보장, 여러 인스턴스 간 동기화 가능
- HolySheep AI 통합: 단일 API 키로 다중 모델 관리 시 필수적인 클라이언트 사이드 rate limiting 구현 가능
AI API 서비스 비교: Rate Limiting 설계의 전제 조건
Rate limiter를 설계하기 전에, 내가 연동하는 AI API 서비스들의 제약 조건을 정확히 이해해야 합니다. HolySheep AI와 주요 경쟁 서비스를 다양한维度로 비교해보겠습니다.
| 서비스 | 기본 Rate Limit | GPT-4.1 가격 | 지연 시간 (P50) | 결제 방식 | 지원 모델 | 적합한 팀 |
|---|---|---|---|---|---|---|
| HolySheep AI | RPM: 500 / TPM: 1M | $8.00/MTok | ~180ms | 로컬 결제 (신용카드 불필요) | GPT-4.1, Claude 3.5, Gemini 2.5, DeepSeek V3.2 | 스타트업, 해외결제为难한 팀 |
| OpenAI 공식 | RPM: 500 / TPM: 1M | $8.00/MTok | ~150ms | 신용카드 필수 | GPT-4.1, o1, o3 | 미국 기반 기업 |
| Anthropic 공식 | RPM: 1000 / TPM: 2M | $15.00/MTok | ~200ms | 신용카드 필수 | Claude 3.5 Sonnet, Opus 4 | 대화형 AI 특화 팀 |
| Google Vertex AI | RPM: 600 / TPM: 2M | $7.00/MTok | ~220ms | 신용카드 + 청구서 | Gemini 2.5, Claude via Partner | GCP 생태계 사용자 |
| AWS Bedrock | RPM: 1000 / TPM: 동적 | $6.50/MTok | ~250ms | AWS 결제수단 | Claude 3.5, Titan, Llama 3.3 | AWS 인프라 활용 팀 |
저의 경험상, HolySheep AI를 선택하는 가장 큰 장점은 로컬 결제 지원입니다. 저는 과거 해외 신용카드 발급에 数개월이 걸린 경험이 있는데, HolySheep AI는 가입 후 즉시 API 키를 발급받아 분산 rate limiter를 테스트할 수 있었습니다. 또한 단일 엔드포인트로 여러 모델을 호출할 수 있어서, 각 서비스별 rate limit을 별도로 관리해야 하는 번거로움이 없습니다.
Token Bucket 알고리즘 원리
Token Bucket은Rate limiting 분야에서 가장 널리 사용되는 알고리즘입니다. 핵심 개념은 다음과 같습니다:
- 버킷 용량 (bucket capacity): 최대로 저장 가능한 토큰 수. burst traffic의 크기를 결정합니다.
- 토큰 충천률 (refill rate): 시간당 생성되는 토큰 수. 평균 처리량을 결정합니다.
- 토큰 소비 (consume): 각 요청마다 토큰을 소비합니다. 토큰이 없으면 요청을 거부합니다.
분산 환경에서는 이 연산을 원자적(atomic)으로 수행해야 합니다. Redis의 Lua 스크립트는 이를 보장하는 가장 확실한 방법입니다.
분산 Token Bucket 구현: Redis + Lua
실제 프로덕션 환경에서 사용하는 Token Bucket 구현입니다. HolySheep AI API와 함께 사용하기 위해 최적화되어 있습니다.
-- token_bucket.lua
-- Redis Lua 스크립트: 원자적 Token Bucket 연산
-- KEYS[1]: Rate limiter 키 (예: "ratelimit:user:123")
-- ARGV[1]: 버킷 용량 (capacity)
-- ARGV[2]: 토큰 충천률 (refill_rate per second)
-- ARGV[3]: 요청에 필요한 토큰 수 (기본 1)
-- ARGV[4]: 현재 타임스탬프 (밀리초)
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local requested = tonumber(ARGV[3])
local now = tonumber(ARGV[4])
-- Redis 해시에서 상태 가져오기
local data = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(data[1])
local last_refill = tonumber(data[2])
-- 초기 상태 처리 (첫 요청인 경우)
if tokens == nil then
tokens = capacity
last_refill = now
end
-- 토큰 충천 계산
local elapsed = (now - last_refill) / 1000.0 -- 초 단위
local refill_amount = elapsed * refill_rate
tokens = math.min(capacity, tokens + refill_amount)
last_refill = now
-- 토큰 소비 시도
local allowed = 0
local remaining = tokens
if tokens >= requested then
tokens = tokens - requested
allowed = 1
remaining = tokens
end
-- 상태 저장 (TTL 설정으로 만료된 키 자동 정리)
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', last_refill)
redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) + 10)
-- 결과 반환: allowed, remaining_tokens, retry_after_ms
if allowed == 1 then
return {allowed, remaining, 0}
else
local retry_after = ((requested - tokens) / refill_rate) * 1000
return {allowed, 0, math.ceil(retry_after)}
end
위 Lua 스크립트를 Redis에 로드하고, Python 클라이언트에서 호출하는方法是 다음과 같습니다:
# token_bucket_client.py
import redis
import time
import httpx
from typing import Tuple, Optional
class DistributedTokenBucket:
"""
분산 환경용 Token Bucket Rate Limiter
HolySheep AI API 연동 최적화 버전
"""
def __init__(
self,
redis_url: str = "redis://localhost:6379",
capacity: int = 100,
refill_rate: float = 10.0,
api_key: str = "YOUR_HOLYSHEEP_API_KEY",
base_url: str = "https://api.holysheep.ai/v1"
):
self.redis = redis.from_url(redis_url)
self.capacity = capacity
self.refill_rate = refill_rate
self.api_key = api_key
self.base_url = base_url
# Lua 스크립트 로드 (SHA로 캐싱)
self._script = self.redis.register_script(self._get_lua_script())
def _get_lua_script(self) -> str:
return """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local requested = tonumber(ARGV[3])
local now = tonumber(ARGV[4])
local data = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(data[1])
local last_refill = tonumber(data[2])
if tokens == nil then
tokens = capacity
last_refill = now
end
local elapsed = (now - last_refill) / 1000.0
local refill_amount = elapsed * refill_rate
tokens = math.min(capacity, tokens + refill_amount)
last_refill = now
local allowed = 0
local remaining = tokens
if tokens >= requested then
tokens = tokens - requested
allowed = 1
remaining = tokens
end
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', last_refill)
redis.call('EXPIRE', key, math.ceil(capacity / refill_rate) + 10)
if allowed == 1 then
return {allowed, remaining, 0}
else
local retry_after = ((requested - tokens) / refill_rate) * 1000
return {allowed, 0, math.ceil(retry_after)}
end
"""
def acquire(
self,
key: str,
tokens: int = 1,
block: bool = False,
timeout: float = 30.0
) -> Tuple[bool, float, Optional[float]]:
"""
토큰 획득 시도
Returns:
(allowed: bool, remaining: float, retry_after: Optional[float])
"""
redis_key = f"ratelimit:{key}"
start_time = time.time()
while True:
now = time.time() * 1000 # 밀리초
result = self._script(
keys=[redis_key],
args=[self.capacity, self.refill_rate, tokens, now]
)
allowed = bool(result[0])
remaining = float(result[1])
retry_after = float(result[2]) / 1000 if result[2] > 0 else None
if allowed or not block:
return allowed, remaining, retry_after
# 블로킹 모드: 대기 후 재시도
wait_time = min(retry_after, timeout - (time.time() - start_time))
if wait_time <= 0:
return False, 0, None
time.sleep(min(wait_time, 0.1)) # 최대 100ms 대기
def call_with_rate_limit(
self,
model: str,
messages: list,
max_tokens: int = 1000
) -> dict:
"""
HolySheep AI API 호출 + Rate Limiting 통합
"""
# 1단계: Rate Limit 확인
allowed, remaining, retry_after = self.acquire(f"user:default", tokens=1)
if not allowed:
raise RateLimitError(f"Rate limit exceeded. Retry after {retry_after:.2f}s")
# 2단계: API 호출
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens
}
with httpx.Client(timeout=60.0) as client:
response = client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
if response.status_code == 429:
# HolySheep AI의 rate limit 응답 처리
retry_after_header = response.headers.get('retry-after', '5')
raise RateLimitError(
f"API rate limit exceeded. Wait {retry_after_header}s"
)
response.raise_for_status()
return response.json()
class RateLimitError(Exception):
"""Rate Limit 초과 예외"""
pass
사용 예제
if __name__ == "__main__":
limiter = DistributedTokenBucket(
redis_url="redis://localhost:6379",
capacity=50, # 버킷 용량: burst 50개 요청 허용
refill_rate=10.0 # 초당 10개 토큰 충천
)
# HolySheep AI로 다중 모델 호출
messages = [{"role": "user", "content": "안녕하세요!"}]
for model in ["gpt-4.1", "claude-3-5-sonnet", "gemini-2.5-flash"]:
try:
result = limiter.call_with_rate_limit(model, messages)
print(f"{model}: {result['choices'][0]['message']['content'][:50]}...")
except RateLimitError as e:
print(f"{model}: {e}")
HolySheep AI 다중 모델 Rate Limiter
실제 프로덕션에서는 모델마다異なる rate limit이 적용됩니다. HolySheep AI에서 제공하는 여러 모델의 제한을 개별적으로 관리하는 고급 구현을 소개합니다.
# multi_model_rate_limiter.py
from collections import defaultdict
from dataclasses import dataclass
from typing import Dict, Optional
import time
@dataclass
class ModelRateLimit:
"""모델별 Rate Limit 설정"""
rpm: int # Requests Per Minute
tpm: int # Tokens Per Minute
tps: float # Tokens Per Second (平滑 제어용)
burst_capacity: int
class HolySheepMultiModelRateLimiter:
"""
HolySheep AI 다중 모델 Rate Limiter
모델별 개별 버킷 관리 + 전체 요청聚合 limit
"""
# HolySheep AI 공식 Rate Limits (2024 기준)
DEFAULT_LIMITS = {
"gpt-4.1": ModelRateLimit(rpm=500, tpm=1000000, tps=16666, burst_capacity=100),
"gpt-4.1-mini": ModelRateLimit(rpm=500, tpm=1000000, tps=16666, burst_capacity=100),
"claude-3-5-sonnet": ModelRateLimit(rpm=1000, tpm=2000000, tps=33333, burst_capacity=200),
"claude-3-5-haiku": ModelRateLimit(rpm=1000, tpm=2000000, tps=33333, burst_capacity=200),
"gemini-2.5-flash": ModelRateLimit(rpm=600, tpm=2000000, tps=33333, burst_capacity=150),
"deepseek-v3": ModelRateLimit(rpm=2000, tpm=10000000, tps=166666, burst_capacity=500),
}
def __init__(self, redis_client):
self.redis = redis_client
self.limits = self.DEFAULT_LIMITS.copy()
async def check_limit(
self,
user_id: str,
model: str,
token_count: int
) -> Dict[str, any]:
"""
Rate Limit 확인 및 토큰 소비
Returns:
{
"allowed": bool,
"model_bucket_remaining": int,
"global_remaining": int,
"retry_after_ms": Optional[float],
"limit_type": Optional[str] # "rpm" | "tpm" | "tokens"
}
"""
model_limit = self.limits.get(model)
if not model_limit:
# 알 수 없는 모델은 허용 (설정 추가 필요)
return {"allowed": True, "model_bucket_remaining": -1}
now_ms = time.time() * 1000
# 1. RPM 체크 (모델별 요청 수)
rpm_key = f"ratelimit:rpm:{user_id}:{model}"
rpm_remaining = self._check_rpm_bucket(rpm_key, model_limit.rpm, 60)
if rpm_remaining <= 0:
return {
"allowed": False,
"model_bucket_remaining": 0,
"global_remaining": 0,
"retry_after_ms": 60000,
"limit_type": "rpm"
}
# 2. TPM 체크 (전체 토큰 사용량)
tpm_key = f"ratelimit:tpm:{user_id}"
tpm_remaining = self._check_tpm_bucket(tpm_key, model_limit.tpm, 60, token_count)
if tpm_remaining <= 0:
return {
"allowed": False,
"model_bucket_remaining": rpm_remaining,
"global_remaining": 0,
"retry_after_ms": 60000,
"limit_type": "tpm"
}
# 3. 버스트 토큰 Bucket (平滑 제어를 위한 실시간 토큰)
bucket_key = f"ratelimit:bucket:{user_id}:{model}"
bucket_allowed, bucket_remaining = self._check_token_bucket(
bucket_key,
model_limit.burst_capacity,
model_limit.tps,
now_ms,
1 # 요청 1개 = 토큰 1개
)
if not bucket_allowed:
retry_ms = ((1 - bucket_remaining) / model_limit.tps) * 1000
return {
"allowed": False,
"model_bucket_remaining": rpm_remaining,
"global_remaining": tpm_remaining,
"retry_after_ms": max(retry_ms, 100),
"limit_type": "tokens"
}
return {
"allowed": True,
"model_bucket_remaining": rpm_remaining,
"global_remaining": tpm_remaining - token_count,
"retry_after_ms": None,
"limit_type": None
}
def _check_rpm_bucket(self, key: str, limit: int, window: int) -> int:
"""Sliding Window Counter로 RPM 체크"""
now = int(time.time())
window_start = now - window
pipe = self.redis.pipeline()
pipe.zremrangebyscore(key, 0, window_start)
pipe.zadd(key, {str(now): now})
pipe.zcard(key)
pipe.expire(key, window + 1)
results = pipe.execute()
current_count = results[2]
return max(0, limit - current_count)
def _check_tpm_bucket(self, key: str, limit: int, window: int, tokens: int) -> int:
"""Sliding Window Counter로 TPM 체크"""
now = int(time.time())
window_start = now - window
pipe = self.redis.pipeline()
pipe.zremrangebyscore(key, 0, window_start)
pipe.zadd(key, {f"{now}:{tokens}": now})
pipe.zrangebyscore(key, window_start, now)
pipe.expire(key, window + 1)
results = pipe.execute()
# 해당 윈도우 내 토큰 합계 계산
total_tokens = sum(
int(item.split(":")[1])
for item in results[2]
)
return max(0, limit - total_tokens)
def _check_token_bucket(
self,
key: str,
capacity: int,
refill_rate: float,
now_ms: float,
requested: int
) -> tuple:
"""Token Bucket 알고리즘 (Lua 스크립트 사용 권장)"""
script = """
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local requested = tonumber(ARGV[4])
local data = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(data[1])
local last_refill = tonumber(data[2])
if tokens == nil then
tokens = capacity
last_refill = now
end
local elapsed = (now - last_refill) / 1000.0