저는 3년간 AI API 게이트웨이 아키텍처를 설계하며 수많은 레이트 리밋 장애를 경험했습니다. 본 기사에서는 스타트업 환경에서 AI API를 안정적으로 운영하기 위한 레이트 리밋 전략을 프로덕션 수준의 코드와 실제 벤치마크 데이터와 함께 설명드리겠습니다.
왜 레이트 리밋이 스타트업 성장의 병목인가
AI API는 전통적인 REST API와 달리 토큰 기반 과금 구조를 가집니다. 요청 수 제한뿐 아니라 토큰 소비량, 동시 연결 수, 시간당 호출 횟수 등 다차원적인 제어가 필요합니다. HolySheep AI를 통해 단일 API 키로 여러 모델을 통합할 때, 각 모델의 레이트 리밋 정책差异를 효과적으로 관리하는 것이 핵심입니다.
- 동시성 폭주: 사용자 증가 시 동시 요청이 기하급수적으로 증가
- 토큰 과소비: 잘못된 프롬프트 설계로 불필요한 토큰 낭비
- 비용 불안정: 예측 불가능한 사용량으로 월별 비용 변동
- 서비스 가용성: 레이트 리밋 초과 시 서비스 중단 위험
레이트 리밋 아키텍처 패턴 3가지
1. 토큰 버킷 알고리즘 (Token Bucket)
가장 널리 사용되는 알고리즘입니다. 정해진 속도로 토큰이 충전되고, 요청 시 토큰을 소비합니다. 버스트 트래픽을 허용하면서도 평균 사용량을 통제할 수 있습니다.
import time
import threading
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, Optional
import asyncio
@dataclass
class TokenBucketRateLimiter:
"""
토큰 버킷 기반 레이트 리밋러
HolySheep AI API 연동 시 권장하는 기본 패턴
"""
capacity: int # 버킷 최대 용량
refill_rate: float # 초당 충전 토큰 수
tokens: float = field(init=False)
last_refill: float = field(init=False)
lock: threading.Lock = field(default_factory=threading.Lock)
def __post_init__(self):
self.tokens = float(self.capacity)
self.last_refill = time.time()
def _refill(self):
"""토큰 자동 충전 로직"""
now = time.time()
elapsed = now - self.last_refill
self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
self.last_refill = now
def acquire(self, tokens: int = 1, timeout: float = 30.0) -> bool:
"""토큰 획득 시도, 성공 시 True 반환"""
start_time = time.time()
while True:
with self.lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
if time.time() - start_time >= timeout:
return False
time.sleep(0.01) # 10ms 대기 후 재시도
def get_wait_time(self) -> float:
"""필요한 대기 시간 계산 (초 단위)"""
with self.lock:
self._refill()
if self.tokens >= 1:
return 0.0
return (1.0 - self.tokens) / self.refill_rate
HolySheep AI 모델별 레이트 리밋 설정 예시
RATE_LIMITS = {
"gpt-4.1": TokenBucketRateLimiter(capacity=500, refill_rate=50), # 500 TPM → 50 TPS
"claude-sonnet-4": TokenBucketRateLimiter(capacity=1000, refill_rate=100), # Claude 4.5는 1000 TPM
"gemini-2.5-flash": TokenBucketRateLimiter(capacity=2000, refill_rate=200), # Gemini Flash 고속
"deepseek-v3.2": TokenBucketRateLimiter(capacity=3000, refill_rate=300), # DeepSeek 高并发
}
async def call_with_rate_limit(
model: str,
prompt: str,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1"
):
"""레이트 리밋 적용 AI API 호출"""
limiter = RATE_LIMITS.get(model)
if not limiter:
raise ValueError(f"Unknown model: {model}")
# 토큰 획득 대기
wait_time = limiter.get_wait_time()
if wait_time > 0:
await asyncio.sleep(wait_time)
# HolySheep AI API 호출
import aiohttp
async with aiohttp.ClientSession() as session:
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 1024,
"temperature": 0.7
}
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async with session.post(
f"{base_url}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
if response.status == 429:
retry_after = int(response.headers.get("Retry-After", 5))
await asyncio.sleep(retry_after)
return await call_with_rate_limit(model, prompt, api_key, base_url)
return await response.json()
벤치마크 테스트
async def benchmark_rate_limiter():
"""레이트 리밋 성능 벤치마크"""
import statistics
latencies = []
limiter = TokenBucketRateLimiter(capacity=100, refill_rate=10)
for i in range(100):
start = time.time()
limiter.acquire(tokens=1, timeout=5.0)
latencies.append((time.time() - start) * 1000)
print(f"평균 대기 시간: {statistics.mean(latencies):.2f}ms")
print(f"중앙값: {statistics.median(latencies):.2f}ms")
print(f"최대 대기 시간: {max(latencies):.2f}ms")
print(f"P99 지연: {sorted(latencies)[98]:.2f}ms")
if __name__ == "__main__":
asyncio.run(benchmark_rate_limiter())2. 슬라이딩 윈도우 카운터 (Sliding Window Counter)
시간 기반 윈도우를 슬라이딩하여 더 정확한 레이트 제어가 가능합니다. HolySheep AI의 분당 요청 수(RPM) 제한에 최적화된 패턴입니다.
interface RateLimitConfig {
windowMs: number; // 윈도우 크기 (밀리초)
maxRequests: number; // 윈도우 내 최대 요청 수
}
interface SlidingWindowCounter {
timestamps: number[]; // 요청 타임스탬프 배열
windowMs: number;
maxRequests: number;
}
class SlidingWindowRateLimiter {
private windows: Map = new Map();
private cleanupInterval: NodeJS.Timeout;
constructor(
private config: RateLimitConfig,
private onLimitExceeded?: (key: string, waitMs: number) => void
) {
// 1분마다 만료된 타임스탬프 정리
this.cleanupInterval = setInterval(() => this.cleanup(), 60000);
}
private getWindow(key: string): SlidingWindowCounter {
if (!this.windows.has(key)) {
this.windows.set(key, {
timestamps: [],
windowMs: this.config.windowMs,
maxRequests: this.config.maxRequests,
});
}
return this.windows.get(key)!;
}
private cleanup(): void {
const now = Date.now();
for (const [key, window] of this.windows.entries()) {
const cutoff = now - window.windowMs;
window.timestamps = window.timestamps.filter(ts => ts > cutoff);
if (window.timestamps.length === 0) {
this.windows.delete(key);
}
}
}
/**
* 레이트 리밋 확인 및 요청 허용
* @returns { allowed: boolean, remaining: number, resetMs: number }
*/
check(key: string): { allowed: boolean; remaining: number; resetMs: number } {
const window = this.getWindow(key);
const now = Date.now();
const windowStart = now - window.windowMs;
// 윈도우 내 요청 필터링
window.timestamps = window.timestamps.filter(ts => ts > windowStart);
if (window.timestamps.length >= window.maxRequests) {
// 가장 오래된 요청이 만료되는 시간 계산
const oldestTimestamp = Math.min(...window.timestamps);
const resetMs = oldestTimestamp + window.windowMs - now;
this.onLimitExceeded?.(key, resetMs);
return {
allowed: false,
remaining: 0,
resetMs,
};
}
// 요청 허용 및 타임스탬프 기록
window.timestamps.push(now);
return {
allowed: true,
remaining: window.maxRequests - window.timestamps.length,
resetMs: window.windowMs,
};
}
/**
* HolySheep AI API 호출 래퍼
*/
async executeWithLimit(
key: string,
fn: () => Promise,
maxRetries: number = 3
): Promise {
const result = this.check(key);
if (!result.allowed) {
if (maxRetries <= 0) {
throw new Error(Rate limit exceeded for ${key}. Wait ${result.resetMs}ms);
}
console.log(Rate limited. Waiting ${result.resetMs}ms before retry...);
await new Promise(resolve => setTimeout(resolve, result.resetMs));
return this.executeWithLimit(key, fn, maxRetries - 1);
}
try {
return await fn();
} catch (error: any) {
// HolySheep AI의 429 에러 처리
if (error?.status === 429) {
const retryAfter = parseInt(error?.headers?.['retry-after'] || '5', 10) * 1000;
await new Promise(resolve => setTimeout(resolve, retryAfter));
return this.executeWithLimit(key, fn, maxRetries - 1);
}
throw error;
}
}
destroy(): void {
clearInterval(this.cleanupInterval);
this.windows.clear();
}
}
// HolySheep AI 모델별 레이트 리밋 설정
const HOLYSHEEP_RATE_LIMITS: Record = {
'gpt-4.1': { windowMs: 60000, maxRequests: 500 }, // 500 RPM
'claude-sonnet-4': { windowMs: 60000, maxRequests: 1000 }, // 1000 RPM
'gemini-2.5-flash': { windowMs: 60000, maxRequests: 2000 }, // 2000 RPM
'deepseek-v3.2': { windowMs: 60000, maxRequests: 3000 }, // 3000 RPM
};
// 사용 예시
async function main() {
const limiter = new SlidingWindowRateLimiter(
HOLYSHEEP_RATE_LIMITS['gpt-4.1'],
(key, waitMs) => console.warn(Rate limit warning: ${key}, wait ${waitMs}ms)
);
// 배치 처리 예시
const prompts = [
"Startup 성장 전략은?",
"AI API 비용 최적화 방법은?",
"레이트 리밋 구현 패턴은?",
];
const results = await Promise.all(
prompts.map((prompt, index) =>
limiter.executeWithLimit(
user-session-123,
async () => {
const response = await fetch('https://api.holysheep.ai/v1/chat/completions', {
method: 'POST',
headers: {
'Authorization': 'Bearer YOUR_HOLYSHEEP_API_KEY',
'Content-Type': 'application/json',
},
body: JSON.stringify({
model: 'gpt-4.1',
messages: [{ role: 'user', content: prompt }],
max_tokens: 500,
}),
});
return response.json();
}
)
)
);
console.log('Results:', results);
limiter.destroy();
}
export { SlidingWindowRateLimiter, RateLimitConfig, HOLYSHEEP_RATE_LIMITS }; 실전 비용 최적화: HolySheep AI 활용 전략
스타트업에서 AI 비용을 절감하려면 모델 선택과 요청 최적화가 핵심입니다. HolySheep AI는 다양한 모델을 단일 API 키로 제공하여 모델 스위칭을 유연하게 처리할 수 있습니다.
모델별 비용 비교 (2024년 12월 기준)
- DeepSeek V3.2: $0.42/MTok — 가장 경제적, 복잡하지 않은 태스크에 최적
- Gemini 2.5 Flash: $2.50/MTok — 고속 응답, 대량 처리 시 비용 효율적
- Claude Sonnet 4: $15/MTok — 균형 잡힌 성능, 코딩 작업에 강점
- GPT-4.1: $8/MTok — 최고 품질, 복잡한 추론 작업
"""
HolySheep AI 비용 최적화 라우터
작업 복잡도에 따라 최적 모델 자동 선택
"""
import time
from enum import Enum
from dataclasses import dataclass
from typing import List, Dict, Callable, Optional
from abc import ABC, abstractmethod
class TaskComplexity(Enum):
SIMPLE = "simple" # 단순 질문, 번역, 요약
MODERATE = "moderate" # 코드 생성, 분석, 설명
COMPLEX = "complex" # 복잡한 추론, 다단계 문제
@dataclass
class ModelConfig:
name: str
cost_per_mtok: float # 달러 단위
avg_latency_ms: float # 평균 응답 시간
max_tokens: int
provider: str = "holysheep"
@dataclass
class CostEstimate:
input_tokens: int
output_tokens: int
model: str
cost: float
latency_ms: float
class CostAwareRouter:
"""
비용과 품질 트레이드오프를 고려한 모델 라우터
HolySheep AI의 다중 모델 통합 기능을 활용
"""
# HolySheep AI 모델 설정
MODELS: Dict[TaskComplexity, List[ModelConfig]] = {
TaskComplexity.SIMPLE: [
ModelConfig("deepseek-v3.2", 0.00042, 800, 4096),
ModelConfig("gemini-2.5-flash", 0.00250, 600, 8192),
],
TaskComplexity.MODERATE: [
ModelConfig("gemini-2.5-flash", 0.00250, 600, 8192),
ModelConfig("claude-sonnet-4", 0.015, 1200, 8192),
ModelConfig("gpt-4.1", 0.008, 1500, 8192),
],
TaskComplexity.COMPLEX: [
ModelConfig("claude-sonnet-4", 0.015, 1200, 8192),
ModelConfig("gpt-4.1", 0.008, 1500, 8192),
],
}
def __init__(self, cost_threshold: float = 0.10):
"""
Args:
cost_threshold: 단일 요청 최대 비용 한도 (달러)
"""
self.cost_threshold = cost_threshold
self.request_stats: List[CostEstimate] = []
def estimate_cost(
self,
complexity: TaskComplexity,
estimated_input_tokens: int,
estimated_output_tokens: int
) -> List[CostEstimate]:
"""가능한 모델들의 비용 추정"""
estimates = []
for model in self.MODELS.get(complexity, []):
input_cost = (estimated_input_tokens / 1_000_000) * model.cost_per_mtok
output_cost = (estimated_output_tokens / 1_000_000) * model.cost_per_mtok
total_cost = input_cost + output_cost
estimates.append(CostEstimate(
input_tokens=estimated_input_tokens,
output_tokens=estimated_output_tokens,
model=model.name,
cost=total_cost,
latency_ms=model.avg_latency_ms,
))
return sorted(estimates, key=lambda x: x.cost)
def select_optimal_model(
self,
complexity: TaskComplexity,
estimated_input_tokens: int,
estimated_output_tokens: int,
prefer_speed: bool = False
) -> ModelConfig:
"""비용 및 속도 기반 최적 모델 선택"""
estimates = self.estimate_cost(
complexity,
estimated_input_tokens,
estimated_output_tokens
)
# 비용 한도 내 모델 필터링
candidates = [e for e in estimates if e.cost <= self.cost_threshold]
if not candidates:
# 비용 한도 초과 시 가장 저렴한 모델 강제 선택
return ModelConfig(
name=estimates[0].model,
cost_per_mtok=0.00042, # DeepSeek最低价
avg_latency_ms=800,
max_tokens=4096,
)
if prefer_speed:
return ModelConfig(
name=min(candidates, key=lambda x: x.latency_ms).model,
cost_per_mtok=0.00042,
avg_latency_ms=600,
max_tokens=8192,
)
return ModelConfig(
name=candidates[0].model,
cost_per_mtok=0.00042,
avg_latency_ms=1500,
max_tokens=8192,
)
def calculate_monthly_cost(
self,
daily_requests: int,
avg_input_tokens: int,
avg_output_tokens: int,
complexity_distribution: Dict[TaskComplexity, float]
) -> Dict[str, float]:
"""월간 비용 추정"""
daily_costs = {}
for complexity, ratio in complexity_distribution.items():
requests_for_complexity = int(daily_requests * ratio)
estimates = self.estimate_cost(
complexity,
avg_input_tokens,
avg_output_tokens
)
# 가장 비용 효율적인 모델 선택
daily_cost = estimates[0].cost * requests_for_complexity
daily_costs[complexity.value] = daily_cost
total_monthly = sum(daily_costs.values()) * 30
return {
"daily_costs": daily_costs,
"monthly_total": total_monthly,
"monthly_total_cents": int(total_monthly * 100),
}
벤치마크 및 비용 시뮬레이션
def run_cost_simulation():
"""비용 최적화 시뮬레이션"""
router = CostAwareRouter(cost_threshold=0.05)
# 월간 10,000건 요청 시나리오
scenario = {
"daily_requests": 10000,
"avg_input_tokens": 500,
"avg_output_tokens": 300,
"complexity_distribution": {
TaskComplexity.SIMPLE: 0.5,
TaskComplexity.MODERATE: 0.35,
TaskComplexity.COMPLEX: 0.15,
}
}
costs = router.calculate_monthly_cost(**scenario)
print("=" * 50)
print("월간 비용 추정 (일 10,000건 요청)")
print("=" * 50)
print(f"단순 작업 (50%): ${costs['daily_costs']['simple']:.4f}/일")
print(f"중간 작업 (35%): ${costs['daily_costs']['moderate']:.4f}/일")
print(f"복잡 작업 (15%): ${costs['daily_costs']['complex']:.4f}/일")
print(f"월간 총액: ${costs['monthly_total']:.2f}")
print(f"센트 단위: {costs['monthly_total_cents']}¢")
print("=" * 50)
# 모델 선택 예시
print("\n모델 선택 예시:")
model = router.select_optimal_model(
TaskComplexity.MODERATE,
estimated_input_tokens=800,
estimated_output_tokens=500,
)
print(f"선택된 모델: {model.name}")
if __name__ == "__main__":
run_cost_simulation()동시성 제어: 연결 풀과 세마포어 패턴
다중 요청 처리 시 HolySheep AI의 동시 연결 제한을 초과하지 않으면서 최대 처리량을 달성하는 것이 중요합니다.
import asyncio
import aiohttp
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class ConnectionPoolConfig:
max_concurrent_requests: int = 10 # 동시 요청 최대 수
max_connections_per_host: int = 5 # 호스트별 최대 연결
request_timeout: float = 60.0 # 요청 타임아웃 (초)
retry_attempts: int = 3 # 재시도 횟수
retry_backoff: float = 1.5 # 지수 백오프
class HolySheepAIClient:
"""
HolySheep AI API 전용 동시성 제어 클라이언트
레이트 리밋과 연결 풀을 통합 관리
"""
def __init__(
self,
api_key: str,
config: Optional[ConnectionPoolConfig] = None
):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.config = config or ConnectionPoolConfig()
# 세마포어로 동시성 제어
self.semaphore = asyncio.Semaphore(self.config.max_concurrent_requests)
# 레이트 리밋 상태
self.rate_limit_state = {
"gpt-4.1": {"tokens": 0, "last_reset": time.time()},
"claude-sonnet-4": {"tokens": 0, "last_reset": time.time()},
"gemini-2.5-flash": {"tokens": 0, "last_reset": time.time()},
"deepseek-v3.2": {"tokens": 0, "last_reset": time.time()},
}
async def _create_session(self) -> aiohttp.ClientSession:
"""연결 풀 설정이 적용된 세션 생성"""
connector = aiohttp.TCPConnector(
limit=self.config.max_concurrent_requests,
limit_per_host=self.config.max_connections_per_host,
keepalive_timeout=30,
)
timeout = aiohttp.ClientTimeout(
total=self.config.request_timeout,
connect=10,
sock_read=30,
)
return aiohttp.ClientSession(
connector=connector,
timeout=timeout,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
)
async def _handle_rate_limit(
self,
response: aiohttp.ClientResponse,
model: str
) -> float:
"""레이트 리밋 응답 처리"""
if response.status == 429:
retry_after = int(response.headers.get("Retry-After", 5))
reset_time = time.time() + retry_after
logger.warning(
f"Rate limit hit for {model}. "
f"Retrying after {retry_after}s"
)
self.rate_limit_state[model] = {
"tokens": 0,
"last_reset": reset_time
}
return retry_after
return 0
async def _execute_request(
self,
session: aiohttp.ClientSession,
model: str,
messages: List[Dict[str, str]],
**kwargs
) -> Dict[str, Any]:
"""실제 API 요청 실행"""
payload = {
"model": model,
"messages": messages,
"max_tokens": kwargs.get("max_tokens", 1024),
"temperature": kwargs.get("temperature", 0.7),
}
async with session.post(
f"{self.base_url}/chat/completions",
json=payload
) as response:
if response.status == 429:
wait_time = await self._handle_rate_limit(response, model)
raise aiohttp.ClientResponseError(
request_info=response.request_info,
history=response.history,
status=429,
message="Rate limit exceeded"
)
if response.status != 200:
error_body = await response.text()
raise Exception(
f"API Error {response.status}: {error_body}"
)
return await response.json()
async def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
**kwargs
) -> Dict[str, Any]:
"""세마포어 기반 동시성 제어 요청"""
async with self.semaphore:
async with await self._create_session() as session:
last_exception = None
for attempt in range(self.config.retry_attempts):
try:
result = await self._execute_request(
session, model, messages, **kwargs
)
logger.info(
f"Request success: model={model}, "
f"attempt={attempt + 1}"
)
return result
except aiohttp.ClientResponseError as e:
if e.status == 429:
wait_time = self.rate_limit_state[model]["last_reset"] - time.time()
if wait_time > 0:
await asyncio.sleep(wait_time)
last_exception = e
else:
raise
except Exception as e:
last_exception = e
if attempt < self.config.retry_attempts - 1:
backoff = self.config.retry_backoff ** attempt
logger.warning(
f"Request failed (attempt {attempt + 1}): {e}. "
f"Retrying in {backoff}s"
)
await asyncio.sleep(backoff)
raise last_exception
async def batch_chat(
self,
requests: List[Dict[str, Any]],
progress_callback: Optional[callable] = None
) -> List[Dict[str, Any]]:
"""
배치 요청 처리
모든 요청을 동시 실행하되 동시성 제한 준수
"""
tasks = []
for req in requests:
task = self.chat_completion(
model=req["model"],
messages=req["messages"],
max_tokens=req.get("max_tokens", 1024),
)
tasks.append(task)
results = []
completed = 0
for coro in asyncio.as_completed(tasks):
try:
result = await coro
results.append({"success": True, "data": result})
except Exception as e:
results.append({"success": False, "error": str(e)})
completed += 1
if progress_callback:
progress_callback(completed, len(tasks))
return results
벤치마크: 동시 요청 성능 테스트
async def benchmark_concurrent_requests():
"""동시성 제어 성능 벤치마크"""
import statistics
client = HolySheepAIClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
config=ConnectionPoolConfig(
max_concurrent_requests=5,
request_timeout=30.0,
)
)
test_requests = [
{
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": f"테스트 요청 {i}"}],
"max_tokens": 100,
}
for i in range(20)
]
start_time = time.time()
results = await client.batch_chat(test_requests)
total_time = time.time() - start_time
success_count = sum(1 for r in results if r["success"])
print("=" * 50)
print("동시성 벤치마크 결과")
print("=" * 50)
print(f"총 요청 수: {len(test_requests)}")
print(f"성공: {success_count}")
print(f"실패: {len(results) - success_count}")
print(f"총 소요 시간: {total_time:.2f}초")
print(f"평균 응답 시간: {total_time/len(test_requests):.2f}초/요청")
print(f"처리량: {len(test_requests)/total_time:.2f} 요청/초")
print("=" * 50)
if __name__ == "__main__":
asyncio.run(benchmark_concurrent_requests())모니터링 및 알림 시스템
프로덕션 환경에서 레이트 리밋 상태를 실시간 모니터링하고 비용 이상 징후를 감지하는 시스템이 필수적입니다.
"""
HolySheep AI 사용량 모니터링 및 알림 시스템
실시간 대시보드 연동 가능한 Prometheus 메트릭 내보내기
"""
import time
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from collections import defaultdict
from datetime import datetime, timedelta
import threading
import json
@dataclass
class RequestMetrics:
"""단일 요청 메트릭"""
timestamp: float
model: str
input_tokens: int
output_tokens: int
latency_ms: float
status: str # success, rate_limited, error
cost_usd: float
@dataclass
class RateLimitAlert:
alert_type: str # approaching_limit, exceeded, cost_spike
model: str
current_usage: float
limit: float
timestamp: float
message: str
class MetricsCollector:
"""
HolySheep AI API 사용량 수집기
레이트 리밋 근접 상황 및 비용 이상 감지
"""
def __init__(
self,
rate_limits: Dict[str, Dict[str, float]],
cost_per_mtok: Dict[str, float]
):
"""
Args:
rate_limits: 모델별 분당 제한 (RPM, TPM)
cost_per_mtok: 모델별 토큰 단가 (달러)
"""
self.rate_limits = rate_limits
self.cost_per_mtok = cost_per_mtok
self.metrics: List[RequestMetrics] = []
self.alerts: List[RateLimitAlert] = []
# 윈도우별 집계
self.window_duration = 60 # 1분 윈도우
self.lock = threading.Lock()
def record_request(
self,
model: str,
input_tokens: int,
output_tokens: int,
latency_ms: float,
status: str
):
"""요청 메트릭 기록"""
cost = (
(input_tokens + output_tokens) / 1_000_000
) * self.cost_per_mtok.get(model, 0.001)
metric = RequestMetrics(
timestamp=time.time(),
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
latency_ms=latency_ms,
status=status,
cost_usd=cost
)
with self.lock:
self.metrics.append(metric)
self._check_alerts(model)
def _check_alerts(self, model: str):
"""알림 조건 체크"""
now = time.time()
window_start = now - self.window_duration
# 윈도우 내 요청 집계
recent_metrics = [
m for m in self.metrics
if m.model == model and m.timestamp >= window_start
]
total_tokens = sum(m.input_tokens + m.output_tokens for m in recent_metrics)
request_count = len(recent_metrics)
# 분당 토큰 제한 체크
if model in self.rate_limits:
tpm_limit = self.rate_limits[model].get("tpm", float('inf'))
rpm_limit = self.rate_limits[model].get("rpm", float('inf'))
usage_ratio = total_tokens / tpm_limit if tpm_limit else 0
if usage_ratio >= 1.0:
self.alerts.append(RateLimitAlert(
alert_type="exceeded",
model=model,
current_usage=total_tokens,
limit=tpm_limit,
timestamp=now,
message=f"{model} TPM 제한 초과: {total_tokens}/{tpm_limit}"
))
elif usage_ratio >= 0.8:
self.alerts.append(RateLimitAlert(
alert_type="approaching_limit",
model=model,
current_usage=total_tokens,
limit=tpm_limit,
timestamp=now,
message=f"{model} TPM 제한 근접: {usage_ratio*100:.1f}%"
))
def get_current_usage(self, model: str) -> Dict:
"""현재 사용량 조회"""
now = time.time()
window_start = now - self.window_duration
with self.lock:
recent = [
m for m in self.metrics
if m.model == model and m.timestamp >= window_start
]
return {
"model": model,
"requests_in_window": len(recent),
"tokens_in_window": sum(
m.input_tokens + m.output_tokens for m in recent
),
"cost_in_window": sum(m.cost_usd for m in recent),
"avg_latency_ms": (
sum(m.latency_ms for m in recent) / len(recent)
if recent else 0
),
"rate_limited_count": sum(1 for m in recent if m.status == "rate_limited")
}
def get_hourly_cost(self) -> float:
"""시간당 비용 계산"""
now = time.time()
hour_ago = now - 3600
with self.lock:
recent = [
m for m in self.metrics
if m.timestamp >= hour_ago
]
return sum(m.cost_usd for m in recent)
def get_cost_projection(self) -> Dict[str, float]:
"""일일/월간 비용 예측"""
hourly_cost = self.get_hourly_cost()
return {
"hourly_cost_usd": hourly_cost,
"daily_projection_us