토큰 기반 레이트 리미팅, 토큰 버킷과 슬라이딩 윈도우 알고리즘, 사용자별 한도 설정, 비용 캡, Redis 기반 구현을 학습합니다.
전통적인 API에서는 RPS(Requests Per Second)로 트래픽을 제어합니다. 하지만 AI API에서 하나의 요청이 소비하는 리소스는 토큰 수에 따라 수십 배 차이가 납니다.
| 시나리오 | 토큰 수 | 비용 | GPU 시간 |
|---|---|---|---|
| 짧은 질문 응답 | 100 토큰 | $0.001 | 200ms |
| 긴 문서 요약 | 10,000 토큰 | $0.10 | 5s |
| 코드 생성 | 50,000 토큰 | $0.50 | 30s |
RPS만으로 제어하면, 소수의 대형 요청이 GPU 리소스를 독점하거나 비용을 폭발시킬 수 있습니다. Gartner는 2026년까지 AI/LLM 도구로 인한 API 수요가 30% 이상 증가할 것으로 예측하며, 토큰 기반 레이트 리미팅이 표준이 되었습니다.
현대 AI API의 레이트 리미팅은 이중 구조를 채택합니다. 요청 수(RPM, Requests Per Minute)와 토큰 수(TPM, Tokens Per Minute)를 동시에 제한하여, DDoS 방지와 리소스 보호를 모두 달성합니다.
토큰 버킷(Token Bucket)은 가장 널리 사용되는 레이트 리미팅 알고리즘입니다. 일정 속도로 토큰이 버킷에 채워지고, 요청은 토큰을 소비합니다.
import time
import asyncio
from dataclasses import dataclass, field
@dataclass
class TokenBucket:
"""토큰 버킷 레이트 리미터"""
capacity: int # 버킷 최대 용량
refill_rate: float # 초당 충전되는 토큰 수
tokens: float = field(init=False)
last_refill: float = field(init=False)
def __post_init__(self):
self.tokens = float(self.capacity)
self.last_refill = time.monotonic()
def _refill(self) -> None:
now = time.monotonic()
elapsed = now - self.last_refill
self.tokens = min(
self.capacity,
self.tokens + elapsed * self.refill_rate,
)
self.last_refill = now
def try_consume(self, amount: int) -> tuple[bool, dict]:
"""토큰 소비를 시도합니다."""
self._refill()
if self.tokens >= amount:
self.tokens -= amount
return True, {
"remaining": int(self.tokens),
"limit": self.capacity,
"reset_seconds": 0,
}
# 토큰이 부족한 경우: 충전까지 남은 시간 계산
deficit = amount - self.tokens
wait_seconds = deficit / self.refill_rate
return False, {
"remaining": 0,
"limit": self.capacity,
"reset_seconds": round(wait_seconds, 1),
}
# 사용 예시
bucket = TokenBucket(
capacity=100_000, # 최대 10만 토큰
refill_rate=1666.67, # 분당 10만 토큰 (초당 ~1667)
)
# 5000 토큰 요청
allowed, info = bucket.try_consume(5000)
# allowed=True, info={"remaining": 95000, "limit": 100000, ...}슬라이딩 윈도우(Sliding Window)는 고정 윈도우의 경계 문제를 해결합니다. 시간 윈도우가 요청 시점을 기준으로 이동하므로 더 정확한 속도 제어가 가능합니다.
from collections import deque
from dataclasses import dataclass, field
@dataclass
class SlidingWindowCounter:
"""슬라이딩 윈도우 카운터"""
window_seconds: int # 윈도우 크기 (초)
max_tokens: int # 윈도우 내 최대 토큰
entries: deque = field(default_factory=deque)
current_total: int = 0
def _cleanup(self, now: float) -> None:
"""윈도우 밖의 오래된 엔트리 제거"""
cutoff = now - self.window_seconds
while self.entries and self.entries[0][0] < cutoff:
_, tokens = self.entries.popleft()
self.current_total -= tokens
def try_consume(self, tokens: int) -> tuple[bool, dict]:
now = time.monotonic()
self._cleanup(now)
if self.current_total + tokens <= self.max_tokens:
self.entries.append((now, tokens))
self.current_total += tokens
return True, {
"remaining": self.max_tokens - self.current_total,
"limit": self.max_tokens,
"window_seconds": self.window_seconds,
}
# 가장 오래된 엔트리가 만료될 때까지 대기 시간
if self.entries:
oldest_time = self.entries[0][0]
wait = oldest_time + self.window_seconds - now
else:
wait = 0
return False, {
"remaining": 0,
"limit": self.max_tokens,
"reset_seconds": round(wait, 1),
}| 알고리즘 | 장점 | 단점 | 적합한 상황 |
|---|---|---|---|
| 토큰 버킷 | 버스트 허용, 구현 간단 | 메모리 효율적이나 정밀도 낮음 | 일반적인 API 제한 |
| 슬라이딩 윈도우 | 정확한 속도 제어 | 메모리 사용량 높음 | 정밀한 토큰 제어 |
| 고정 윈도우 | 가장 간단 | 경계에서 2배 버스트 | 대략적 제한 |
| 누출 버킷 | 일정한 출력 속도 | 버스트 불허 | 균일한 처리 필요 |
AI API의 레이트 리미팅은 단일 차원이 아닌, 여러 축을 동시에 제어합니다.
from dataclasses import dataclass
@dataclass
class RateLimitConfig:
"""다차원 레이트 리밋 설정"""
# 요청 수 제한
requests_per_minute: int
requests_per_day: int
# 토큰 제한
tokens_per_minute: int
tokens_per_day: int
# 비용 제한
cost_per_day_usd: float
cost_per_month_usd: float
# 티어별 한도 설정
TIER_LIMITS = {
"free": RateLimitConfig(
requests_per_minute=20,
requests_per_day=1_000,
tokens_per_minute=10_000,
tokens_per_day=100_000,
cost_per_day_usd=1.0,
cost_per_month_usd=5.0,
),
"pro": RateLimitConfig(
requests_per_minute=100,
requests_per_day=10_000,
tokens_per_minute=100_000,
tokens_per_day=1_000_000,
cost_per_day_usd=50.0,
cost_per_month_usd=500.0,
),
"enterprise": RateLimitConfig(
requests_per_minute=1_000,
requests_per_day=100_000,
tokens_per_minute=1_000_000,
tokens_per_day=10_000_000,
cost_per_day_usd=500.0,
cost_per_month_usd=10_000.0,
),
}
# 모델별 추가 제한
MODEL_LIMITS = {
"claude-4": {
"max_tokens_per_request": 8192,
"max_context_length": 200_000,
"concurrent_requests": 5,
},
"claude-4-vision": {
"max_tokens_per_request": 4096,
"max_context_length": 128_000,
"concurrent_requests": 3,
"max_images_per_request": 20,
},
}class MultiDimensionalRateLimiter:
"""여러 차원의 레이트 리밋을 동시에 검사합니다."""
def __init__(self, config: RateLimitConfig):
self.limiters = {
"rpm": SlidingWindowCounter(60, config.requests_per_minute),
"rpd": SlidingWindowCounter(86400, config.requests_per_day),
"tpm": TokenBucket(
config.tokens_per_minute,
config.tokens_per_minute / 60,
),
"tpd": SlidingWindowCounter(86400, config.tokens_per_day),
}
self.cost_cap = CostCap(
daily=config.cost_per_day_usd,
monthly=config.cost_per_month_usd,
)
async def check(
self,
user_id: str,
estimated_tokens: int,
estimated_cost: float,
) -> RateLimitResult:
"""모든 차원에서 요청 허용 여부를 검사합니다."""
# 요청 수 검사
rpm_ok, rpm_info = self.limiters["rpm"].try_consume(1)
if not rpm_ok:
return RateLimitResult(
allowed=False,
reason="requests_per_minute_exceeded",
retry_after=rpm_info["reset_seconds"],
headers=self._build_headers("rpm", rpm_info),
)
# 토큰 수 검사
tpm_ok, tpm_info = self.limiters["tpm"].try_consume(estimated_tokens)
if not tpm_ok:
return RateLimitResult(
allowed=False,
reason="tokens_per_minute_exceeded",
retry_after=tpm_info["reset_seconds"],
headers=self._build_headers("tpm", tpm_info),
)
# 비용 검사
cost_ok = await self.cost_cap.check(user_id, estimated_cost)
if not cost_ok:
return RateLimitResult(
allowed=False,
reason="budget_exceeded",
retry_after=None, # 비용 한도는 시간으로 해결 안 됨
headers={"X-Budget-Exceeded": "true"},
)
return RateLimitResult(
allowed=True,
headers=self._build_headers("tpm", tpm_info),
)
def _build_headers(
self, dimension: str, info: dict
) -> dict[str, str]:
return {
f"X-RateLimit-Limit-{dimension.upper()}": str(info["limit"]),
f"X-RateLimit-Remaining-{dimension.upper()}": str(info["remaining"]),
}비용 제한은 레이트 리미팅의 상위 개념으로, 사용자나 조직이 예상치 못한 비용 폭발을 방지합니다.
from decimal import Decimal
class CostTracker:
"""사용자별 비용 추적"""
def __init__(self, redis_client):
self.redis = redis_client
async def record_cost(
self,
user_id: str,
org_id: str,
model: str,
prompt_tokens: int,
completion_tokens: int,
) -> CostRecord:
"""비용을 기록하고 한도를 확인합니다."""
pricing = MODEL_PRICING[model]
prompt_cost = Decimal(prompt_tokens) * pricing.input_per_token
completion_cost = Decimal(completion_tokens) * pricing.output_per_token
total_cost = prompt_cost + completion_cost
# 일별, 월별 비용 누적 (Redis)
pipe = self.redis.pipeline()
daily_key = f"cost:{user_id}:daily:{date.today().isoformat()}"
monthly_key = f"cost:{user_id}:monthly:{date.today().strftime('%Y-%m')}"
org_key = f"cost:{org_id}:monthly:{date.today().strftime('%Y-%m')}"
pipe.incrbyfloat(daily_key, float(total_cost))
pipe.expire(daily_key, 86400 * 2) # 2일 후 만료
pipe.incrbyfloat(monthly_key, float(total_cost))
pipe.expire(monthly_key, 86400 * 35) # 35일 후 만료
pipe.incrbyfloat(org_key, float(total_cost))
pipe.expire(org_key, 86400 * 35)
results = await pipe.execute()
return CostRecord(
user_id=user_id,
daily_total=float(results[0]),
monthly_total=float(results[2]),
org_monthly_total=float(results[4]),
request_cost=float(total_cost),
)
async def check_budget(
self,
user_id: str,
org_id: str,
estimated_cost: float,
) -> BudgetCheckResult:
"""예산 한도 내인지 확인합니다."""
limits = await self.get_limits(user_id, org_id)
daily = float(
await self.redis.get(
f"cost:{user_id}:daily:{date.today().isoformat()}"
) or 0
)
monthly = float(
await self.redis.get(
f"cost:{user_id}:monthly:{date.today().strftime('%Y-%m')}"
) or 0
)
if daily + estimated_cost > limits.daily_usd:
return BudgetCheckResult(
allowed=False,
reason="daily_budget_exceeded",
daily_used=daily,
daily_limit=limits.daily_usd,
)
if monthly + estimated_cost > limits.monthly_usd:
return BudgetCheckResult(
allowed=False,
reason="monthly_budget_exceeded",
monthly_used=monthly,
monthly_limit=limits.monthly_usd,
)
return BudgetCheckResult(allowed=True)비용 캡에서 "예상 비용"은 입력 토큰만으로 계산할 수 있지만, 출력 토큰은 생성이 완료될 때까지 정확히 알 수 없습니다. 따라서 max_tokens 파라미터를 기준으로 최대 비용을 미리 예약(reservation)하고, 실제 사용량으로 정산하는 2단계 방식이 안전합니다.
다중 서버 환경에서 레이트 리미팅을 정확하게 적용하려면 공유 상태 저장소가 필요합니다.
import redis.asyncio as redis
class RedisTokenBucket:
"""Redis 기반 분산 토큰 버킷"""
# Lua 스크립트: 원자적 토큰 소비
CONSUME_SCRIPT = """
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 bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1]) or capacity
local last_refill = tonumber(bucket[2]) or now
-- 토큰 충전
local elapsed = now - last_refill
tokens = math.min(capacity, tokens + elapsed * refill_rate)
local allowed = 0
local remaining = 0
if tokens >= requested then
tokens = tokens - requested
allowed = 1
end
remaining = math.floor(tokens)
-- 상태 저장
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
redis.call('EXPIRE', key, 3600)
return {allowed, remaining}
"""
def __init__(
self,
redis_client: redis.Redis,
capacity: int,
refill_rate: float,
):
self.redis = redis_client
self.capacity = capacity
self.refill_rate = refill_rate
self.script = self.redis.register_script(self.CONSUME_SCRIPT)
async def try_consume(
self,
key: str,
tokens: int,
) -> tuple[bool, dict]:
now = time.time()
result = await self.script(
keys=[f"ratelimit:{key}"],
args=[self.capacity, self.refill_rate, tokens, now],
)
allowed = bool(result[0])
remaining = int(result[1])
return allowed, {
"remaining": remaining,
"limit": self.capacity,
}
# FastAPI 미들웨어로 적용
class RateLimitMiddleware:
def __init__(self, app, redis_url: str):
self.app = app
self.redis = redis.from_url(redis_url)
self.limiters = {}
async def __call__(self, request: Request, call_next):
user_id = request.state.user_id
tier = request.state.tier
config = TIER_LIMITS[tier]
limiter = self._get_limiter(tier)
# 요청 수 검사
allowed, info = await limiter.try_consume(
f"rpm:{user_id}",
tokens=1,
)
if not allowed:
return JSONResponse(
status_code=429,
content={
"error": {
"type": "rate_limit_exceeded",
"message": "요청 한도를 초과했습니다",
}
},
headers={
"Retry-After": str(info.get("reset_seconds", 60)),
"X-RateLimit-Limit-RPM": str(config.requests_per_minute),
"X-RateLimit-Remaining-RPM": str(info["remaining"]),
},
)
response = await call_next(request)
# 응답 헤더에 레이트 리밋 정보 추가
response.headers["X-RateLimit-Limit-RPM"] = str(
config.requests_per_minute
)
response.headers["X-RateLimit-Remaining-RPM"] = str(
info["remaining"]
)
return response429 응답은 클라이언트가 자동으로 재시도할 수 있도록 충분한 정보를 제공해야 합니다.
{
"error": {
"type": "rate_limit_exceeded",
"message": "분당 토큰 한도(100,000 TPM)를 초과했습니다. 15초 후 다시 시도해주세요.",
"code": "tokens_per_minute_exceeded",
"details": {
"limit_type": "tokens_per_minute",
"limit": 100000,
"used": 98500,
"requested": 5000,
"retry_after_seconds": 15
}
}
}HTTP/1.1 429 Too Many Requests
Retry-After: 15
X-RateLimit-Limit-RPM: 100
X-RateLimit-Remaining-RPM: 0
X-RateLimit-Reset-RPM: 2026-03-30T12:01:00Z
X-RateLimit-Limit-TPM: 100000
X-RateLimit-Remaining-TPM: 0
X-RateLimit-Reset-TPM: 2026-03-30T12:00:15ZRetry-After 헤더에 정확한 대기 시간을 제공하면 클라이언트의 지수 백오프(exponential backoff) 전략이 더 효율적으로 동작합니다. 불필요한 재시도를 줄여 서버 부하와 클라이언트 비용을 모두 절감할 수 있습니다.
이 장에서는 AI API 비용 제어의 핵심인 레이트 리미팅을 다루었습니다. 전통적 RPS 기반에서 토큰 기반 레이트 리미팅으로의 전환 이유, 토큰 버킷과 슬라이딩 윈도우 알고리즘, 다차원 한도 설계(요청 수 + 토큰 수 + 비용), 비용 캡 메커니즘, 그리고 Redis 기반 분산 구현을 살펴보았습니다.
효과적인 레이트 리미팅은 서비스 안정성과 공정한 리소스 분배를 보장하면서, 클라이언트에게 투명한 한도 정보와 재시도 가이드를 제공하는 것입니다.
9장에서는 API 스펙으로부터 SDK를 자동 생성하고, 개발자 경험(DX)을 최적화하는 방법을 다룹니다. Stainless, Speakeasy, openapi-generator를 활용한 타입 안전 SDK 생성과, API 문서화, 인터랙티브 플레이그라운드를 살펴봅니다.
이 글이 도움이 되셨나요?
관련 주제 더 보기
OpenAPI 스펙에서 타입 안전 SDK를 자동 생성하고, API 문서화, 인터랙티브 플레이그라운드로 개발자 경험을 최적화하는 방법을 학습합니다.
URL 경로, 헤더, 쿼리 파라미터 버전 관리 전략과 AI 서비스에서의 모델 버전 분리, 프롬프트 버전 관리, 폐기 정책을 학습합니다.
LLM 게이트웨이를 활용한 멀티 프로바이더 라우팅, 모델 폴백, 인증/인가, 캐싱, 관측 가능성 등 프로덕션 API 인프라를 학습합니다.