저는去年某 글로벌 이커머스 플랫폼에서 AI 고객 서비스 시스템을 구축한 경험이 있습니다.那年促销期间，系统突然面临每秒3,000건 이상의 AI 쿼리 폭증 상황.단 5분 만에 API 호출 비용이 $500을 초과하며, 시스템은 과부하 상태에 빠졌습니다.이 경험을 통해 AI API限流設計의 중요성을 뼈저리게 깨달았습니다.
본 튜토리얼에서는 令牌桶(Token Bucket)과 漏桶(Leaky Bucket) 알고리즘을 실제 AI API 환경에서 구현하는 방법을 상세히 다룹니다.특히 HolySheep AI 게이트웨이를 활용한 비용 최적화 전략과 함께 실제 프로덕션에서 검증된 코드를 제공합니다.
왜 AI API限流가 중요한가?
AI API限流는 단순히 "호출 횟수를 제한하는 것"이 아닙니다.다음과 같은 핵심 가치를 제공합니다:
- 비용 보호:突发请求による無制限な課金を防止.예를 들어 Gemini 2.5 Flash는 $2.50/MTok이지만, 무제한 호출 시 하루 만에 수백 달러가 발생할 수 있습니다.
- 시스템 안정성:上游API의 과부하를 방지하고 일관된 응답 시간을 유지합니다.
- 公平한 리소스分配:다중 테넌트 환경에서 각 사용자에게公平한 리소스를 제공합니다.
- 버스트 트래픽 처리:突発的な流量のピーク를 허용하면서도平均使用量を控制합니다.
令牌桶(Token Bucket) 알고리즘
원리
令牌桶 알고리즘은 다음과 같은 메커니즘으로 동작합니다:
- 桶(버킷)에令牌(토큰)이 일정한 속도로 채워집니다.
- 각 요청은 토큰 1개를 소비합니다.
- 桶이 비어 있으면 요청은 대기하거나 거부됩니다.
- 桶이 가득 차면 추가 토큰은溢れ出します(저장 불가).
핵심 장점: 버스트 트래픽을 허용합니다.예를 들어 최대 100토큰 용량의 버킷이 있으면,平时はゆっくりと補充されながら,必要时には最大 100개의 동시 요청을 처리할 수 있습니다.
令牌桶 구현
import time
import threading
from typing import Optional
from dataclasses import dataclass
@dataclass
class TokenBucket:
"""令牌桶限流器 - AI API용"""
capacity: int # 最大令牌数
refill_rate: float # 초당 토큰 충전 속도
tokens: float # 현재 토큰 수
last_refill_time: float
def __init__(self, capacity: int, refill_rate: float):
self.capacity = capacity
self.refill_rate = refill_rate
self.tokens = float(capacity)
self.last_refill_time = time.time()
self._lock = threading.Lock()
def _refill(self):
"""토큰 자동 충전"""
now = time.time()
elapsed = now - self.last_refill_time
new_tokens = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill_time = now
def consume(self, tokens: int = 1) -> bool:
"""토큰 소비 시도"""
with self._lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def wait_for_token(self, tokens: int = 1, timeout: Optional[float] = None) -> bool:
"""토큰이 있을 때까지 대기"""
start_time = time.time()
while True:
if self.consume(tokens):
return True
if timeout and (time.time() - start_time) >= timeout:
return False
time.sleep(0.01) # 10ms 대기
class HolySheepAIClient:
"""HolySheep AI API 클라이언트 with 令牌桶限流"""
def __init__(self, api_key: str, max_rpm: int = 60):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
# 분당 max_rpm 요청을 초당 rate로 변환
self.rate_limiter = TokenBucket(
capacity=max_rpm,
refill_rate=max_rpm / 60.0
)
self.total_tokens_used = 0
self.total_cost = 0.0
def call_model(self, model: str, messages: list,
timeout: float = 30.0) -> dict:
"""
모델 호출 with 限流
HolySheep AI 모델별 가격:
- GPT-4.1: $8.00/MTok (입력), $8.00/MTok (출력)
- Claude Sonnet 4: $15.00/MTok
- Gemini 2.5 Flash: $2.50/MTok
- DeepSeek V3: $0.42/MTok
"""
import requests
# 限流检查
if not self.rate_limiter.wait_for_token(timeout=timeout):
raise Exception(f"限流 초과: {timeout}초 내 토큰 획득 실패")
endpoint = f"{self.base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": 2000
}
response = requests.post(endpoint,
json=payload,
headers=headers,
timeout=timeout)
if response.status_code == 429:
raise Exception("HolySheep API限流 초과 - 잠시 후 재시도")
response.raise_for_status()
result = response.json()
# コスト集計
usage = result.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
# HolySheep 가격 계산
prices = {
"gpt-4.1": 8.00,
"claude-sonnet-4": 15.00,
"gemini-2.5-flash": 2.50,
"deepseek-v3": 0.42
}
price_per_mtok = prices.get(model, 8.00)
cost = ((prompt_tokens + completion_tokens) / 1_000_000) * price_per_mtok
self.total_tokens_used += prompt_tokens + completion_tokens
self.total_cost += cost
return result
def get_stats(self) -> dict:
"""コスト統計取得"""
return {
"total_tokens": self.total_tokens_used,
"total_cost_usd": round(self.total_cost, 4),
"current_bucket_tokens": round(self.rate_limiter.tokens, 2)
}
使用例
if __name__ == "__main__":
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_rpm=120 # 분당 120요청
)
messages = [
{"role": "system", "content": "당신은 유용한 AI 어시스턴트입니다."},
{"role": "user", "content": "서울의 유명한 관광지를 추천해주세요."}
]
try:
response = client.call_model("gemini-2.5-flash", messages)
print(f"응답: {response['choices'][0]['message']['content']}")
print(f"통계: {client.get_stats()}")
except Exception as e:
print(f"오류: {e}")漏桶(Leaky Bucket) 알고리즘
원리
漏桶 알고리즘은 물이桶から漏れ出すように、一定速度でリクエストを処理します:
- 요청이桶に溜まります（一定容量まで）
- 桶から一定速度でリクエストが漏れ出します（処理）
- 桶が 가득으면새로운 요청は溢出します（拒否）
핵심 특징: 출력 속도가一定하여流量が滑らかになります.하지만 버스트 트래픽은 허용하지 않으며,桶가 비면リクエストは待たされます.
漏桶實現
import time
import threading
from collections import deque
from typing import Callable, Any, Optional
class LeakyBucket:
"""漏桶限流器 - 일정한吐出速度を保証"""
def __init__(self, capacity: int, leak_rate: float):
"""
Args:
capacity: 桶容量 (최대 대기 요청 수)
leak_rate: 초당 처리 속도 (leak rate)
"""
self.capacity = capacity
self.leak_rate = leak_rate
self.queue = deque()
self.last_leak_time = time.time()
self._lock = threading.Lock()
self._leak_thread = None
self._running = False
def add(self, item: Any) -> bool:
"""
桶에 요청 추가
Returns:
True: 추가 성공
False: 桶가 가득 차서 거부
"""
with self._lock:
self._leak() # 先泄漏旧的
if len(self.queue) >= self.capacity:
return False
self.queue.append(item)
return True
def _leak(self):
"""시간 경과에 따라桶から漏れ出す"""
now = time.time()
elapsed = now - self.last_leak_time
leak_count = int(elapsed * self.leak_rate)
if leak_count > 0 and len(self.queue) > 0:
for _ in range(min(leak_count, len(self.queue))):
self.queue.popleft()
self.last_leak_time = now
else:
self.last_leak_time = now
def start_daemon(self, callback: Callable[[Any], None]):
"""백그라운드 처리 데몬 시작"""
self._running = True
self._callback = callback
def daemon():
while self._running:
with self._lock:
self._leak()
if self.queue:
item = self.queue.popleft()
threading.Thread(target=self._callback, args=(item,)).start()
time.sleep(1.0 / self.leak_rate)
self._leak_thread = threading.Thread(target=daemon, daemon=True)
self._leak_thread.start()
def stop(self):
"""데몬 중지"""
self._running = False
if self._leak_thread:
self._leak_thread.join(timeout=2.0)
def get_status(self) -> dict:
"""상태 조회"""
with self._lock:
self._leak()
return {
"queue_size": len(self.queue),
"capacity": self.capacity,
"leak_rate": self.leak_rate,
"utilization": len(self.queue) / self.capacity * 100
}
class AIRequestQueue:
"""AI API 요청 큐 - 漏桶算法实现"""
def __init__(self, max_concurrent: int = 10, requests_per_second: float = 5.0):
"""
Args:
max_concurrent: 최대 동시 처리 수
requests_per_second: 초당 처리 속도 (RPS)
"""
self.bucket = LeakyBucket(
capacity=max_concurrent,
leak_rate=requests_per_second
)
self.pending_requests = []
self.completed_results = []
self._lock = threading.Lock()
self._semaphore = threading.Semaphore(max_concurrent)
def enqueue(self, request_id: str, model: str, messages: list) -> bool:
"""요청 대기열에 추가"""
request = {
"id": request_id,
"model": model,
"messages": messages,
"timestamp": time.time()
}
return self.bucket.add(request)
def process_request(self, request: dict) -> dict:
"""요청 실제 처리 (비동기)"""
import requests
self._semaphore.acquire()
try:
# HolySheep AI API 호출
endpoint = "https://api.holysheep.ai/v1/chat/completions"
headers = {
"Authorization": f"Bearer {self._api_key}",
"Content-Type": "application/json"
}
payload = {
"model": request["model"],
"messages": request["messages"],
"max_tokens": 1500
}
start = time.time()
response = requests.post(endpoint, json=payload, headers=headers, timeout=30)
latency = (time.time() - start) * 1000 # ms 단위
result = {
"id": request["id"],
"status": "success" if response.status_code == 200 else "failed",
"latency_ms": round(latency, 2),
"status_code": response.status_code
}
if response.status_code == 200:
result["content"] = response.json()["choices"][0]["message"]["content"]
return result
except Exception as e:
return {
"id": request["id"],
"status": "error",
"error": str(e)
}
finally:
self._semaphore.release()
def start_processing(self, api_key: str):
"""처리 데몬 시작"""
self._api_key = api_key
self.bucket.start_daemon(self.process_request)
def get_metrics(self) -> dict:
"""메트릭스 조회"""
return self.bucket.get_status()
使用例 - 批量処理
if __name__ == "__main__":
queue = AIRequestQueue(
max_concurrent=50, # 最大50개 동시 처리
requests_per_second=10 # 초당 10개씩処理
)
# テスト用リクエスト生成
test_requests = [
(f"req_{i}", "deepseek-v3", [
{"role": "user", "content": f"테스트 요청 {i}번"}
])
for i in range(100)
]
# 대기열에 추가
accepted = 0
rejected = 0
for req_id, model, messages in test_requests:
if queue.enqueue(req_id, model, messages):
accepted += 1
else:
rejected += 1
print(f"승인: {accepted}, 거부: {rejected}")
print(f"상태: {queue.get_metrics()}")令牌桶 vs 漏桶：어떤 알고리즘을 선택해야 하는가?
| 기준 | 令牌桶 (Token Bucket) | 漏桶 (Leaky Bucket) |
|---|---|---|
| 버스트 트래픽 | 허용 (桶容量内) | 불가 (一定速度만) |
| 流量 특성 | 可变速度 | 一定速度 |
| 적합한 상황 | 突発的な 需要 대응, AI 배치処理 | 항상一定の 처리량 필요 (스트리밍) |
| 메모리 사용 | 버킷 크기만 필요 | 대기열 전체 저장 |
| HolySheep 적용 | 일반적인 AI API 호출 | 실시간 스트리밍 응답 |
저의 경험으로는 대부분의 AI API限流 시나리오에서는 令牌桶이 더 적합합니다.특히 HolySheep AI와 함께 사용할 때 버스트 트래픽을 허용하면서도 평균 사용량을 제어할 수 있어 비용 최적화에 효과적입니다.
실전 패턴：RAG 시스템限流 구현
기업용 RAG(Retrieval-Augmented Generation) 시스템을 구축할 때 저는 다음과 같은 3단계限流 전략을 사용합니다:
import time
import asyncio
from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Optional
class TierType(Enum):
"""사용자 등급"""
FREE = "free"
BASIC = "basic"
PRO = "pro"
ENTERPRISE = "enterprise"
@dataclass
class TierConfig:
"""등급별限流 설정"""
rpm: int # 분당 요청 수
tpm: int # 분당 토큰 수
rpd: int # 일일 요청 수
priority: int # 우선순위 (높을수록 우선)
TIER_CONFIGS: Dict[TierType, TierConfig] = {
TierType.FREE: TierConfig(rpm=10, tpm=10000, rpd=100, priority=1),
TierType.BASIC: TierConfig(rpm=60, tpm=100000, rpd=1000, priority=2),
TierType.PRO: TierConfig(rpm=300, tpm=500000, rpd=10000, priority=3),
TierType.ENTERPRISE: TierConfig(rpm=1000, tpm=2000000, rpd=100000, priority=5)
}
class MultiTierRateLimiter:
"""다중 등급限流기 - RAG 시스템용"""
def __init__(self):
self._user_buckets: Dict[str, Dict] = {}
self._window_start = time.time()
self._daily_counts: Dict[str, int] = {}
self._lock = asyncio.Lock()
async def check_limit(
self,
user_id: str,
tier: TierType,
tokens: int = 0
) -> tuple[bool, Optional[str]]:
"""
限流 체크
Returns:
(allowed, reason_if_denied)
"""
config = TIER_CONFIGS[tier]
now = time.time()
async with self._lock:
# 1분窗口リセット
if now - self._window_start > 60:
self._window_start = now
self._user_buckets.clear()
# 일일 카운트 초기화
today = int(now // 86400)
daily_key = f"{user_id}:{today}"
if daily_key not in self._daily_counts:
self._daily_counts = {k: v for k, v in self._daily_counts.items()
if k.endswith(str(today))}
self._daily_counts[daily_key] = 0
# 일일 요청 수 체크
if self._daily_counts[daily_key] >= config.rpd:
return False, f"일일 요청 수 초과 ({config.rpd})"
# 사용자 버킷 초기화
if user_id not in self._user_buckets:
self._user_buckets[user_id] = {
"tokens": 0,
"requests": 0,
"last_update": now
}
bucket = self._user_buckets[user_id]
# 분당 요청 수 체크
elapsed = now - bucket["last_update"]
refill_tokens = elapsed * (config.tpm / 60)
current_tokens = min(config.tpm, bucket["tokens"] + refill_tokens)
# RPM 체크
if bucket["requests"] >= config.rpm:
return False, f"분당 요청 수 초과 ({config.rpm} RPM)"
# TPM 체크
if tokens > current_tokens:
return False, f"분당 토큰 수 초과 ({config.tpm} TPM)"
# 통과 - 상태 업데이트
bucket["tokens"] = current_tokens - tokens
bucket["requests"] += 1
bucket["last_update"] = now
self._daily_counts[daily_key] += 1
return True, None
async def call_with_limit(
self,
user_id: str,
tier: TierType,
messages: List[dict],
model: str = "deepseek-v3"
) -> dict:
"""限流이 적용된 API 호출"""
import httpx
# 토큰估算 (간단한 heuristic)
estimated_tokens = sum(len(m["content"]) // 4 for m in messages)
allowed, reason = await self.check_limit(user_id, tier, estimated_tokens)
if not allowed:
return {
"error": True,
"reason": reason,
"tier": tier.value,
"retry_after": 60
}
# HolySheep AI