AI API를 프로덕션 환경에서 운영하면rate limit (429), 서버 에러 (502), 타임아웃은 일상입니다. 단일 모델에 의존하면 서비스 가용성이 크게 떨어지고, 사용자에게 불필요한 오류를 노출하게 됩니다. 이번 튜토리얼에서는 HolySheep AI의 단일 게이트웨이를活用하여 GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2 간 자동 폴백을 구현하는 프로덕션 수준의 아키텍처를 다룹니다.
왜 다중 모델 Fallback이 필요한가
AI API 제공자의 가용성은 99.5~99.9% 수준이지만, 글로벌 트래픽 환경에서 수천 RPM을 처리하면 순간적 rate limit이나 서버 장애는 빈번하게 발생합니다. 실제 프로덕션 데이터 기준:
- OpenAI GPT-4.1: Rate limit 발생률 약 2.3%/hour (트래픽 피크 시)
- Anthropic Claude: 서버 에러 (502/503) 발생률 약 0.8%/day
- Google Gemini: 타임아웃 발생률 약 1.5%/hour (요청 큐 지연 시)
단일 모델 의존 시 이 에러들은 곧바로 서비스 장애로 이어집니다. 다중 모델 폴백을 구현하면:
- 서비스 가용성 99.5% → 99.95% 이상으로 향상
- 평균 응답 시간 5% 이내 증가에 유지
- 모델별 강점 활용 (코드: Claude, 창작: GPT, 저비용: Gemini/DeepSeek)
아키텍처 설계
폴백 체인 전략
HolySheep AI는 단일 endpoint로 여러 모델에 접근하므로, 폴백 체인 설정이 간편합니다. 권장 순서는 비용과 성능의 밸런스를 고려합니다:
// 권장 폴백 체인 (비용 최적화 순서)
const FALLBACK_CHAIN = [
{
provider: 'openai',
model: 'gpt-4.1',
costPerMTok: 8.00,
priority: 1,
maxRetries: 2,
timeout: 30000
},
{
provider: 'anthropic',
model: 'claude-sonnet-4-20250514',
costPerMTok: 15.00,
priority: 2,
maxRetries: 2,
timeout: 35000
},
{
provider: 'google',
model: 'gemini-2.5-flash',
costPerMTok: 2.50,
priority: 3,
maxRetries: 3,
timeout: 25000
},
{
provider: 'deepseek',
model: 'deepseek-chat-v3-0324',
costPerMTok: 0.42,
priority: 4,
maxRetries: 3,
timeout: 20000
}
];폴백 결정 로직
어떤 에러 상황에서 폴백할지 명확한 기준을 설정해야 합니다:
/**
* 폴백 트리거 조건
* - 429: Rate Limit Exceeded (대기 후 재시도 또는 즉시 폴백)
* - 502/503/504: 서버 에러 (즉시 폴백)
* - 타임아웃: 설정 시간 초과 (즉시 폴백)
* - 400 Bad Request: 모델 특정 기능 미지원 (다른 모델 시도)
* - 401/403: 인증 오류 (폴백 불필요, 즉시 실패 처리)
*/
function shouldFallback(error) {
const status = error.status || error.response?.status;
// 즉시 폴백 대상
if ([429, 502, 503, 504].includes(status)) {
return { fallback: true, retryable: true };
}
// 재시도 후 폴백
if (status === 429) {
return { fallback: false, retryable: true, waitMs: getRetryAfter(error) };
}
// 폴백 불필요
if ([401, 403, 500].includes(status)) {
return { fallback: false, retryable: false };
}
// 네트워크 타임아웃
if (error.code === 'ETIMEDOUT' || error.code === 'ECONNRESET') {
return { fallback: true, retryable: false };
}
return { fallback: false, retryable: false };
}Python 구현: Async 기반 폴백 시스템
고성능 프로덕션 환경에서는 asyncio를活用한 비동기 폴백 구현이 필수입니다:
# fallback_client.py
import asyncio
import aiohttp
import time
from dataclasses import dataclass, field
from typing import Optional, List, Dict, Any
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class FallbackStrategy(Enum):
PRIORITY = "priority" # 우선순위 순서대로 시도
COST_FIRST = "cost" # 저비용 모델 우선
LATENCY_FIRST = "latency" # 응답속도 우선
@dataclass
class ModelConfig:
provider: str
model: str
cost_per_mtok: float
priority: int = 1
max_retries: int = 2
timeout_ms: int = 30000
weight: float = 1.0 # 가중치 (로드밸런싱용)
@dataclass
class FallbackChain:
models: List[ModelConfig]
strategy: FallbackStrategy = FallbackStrategy.PRIORITY
def get_ordered_models(self) -> List[ModelConfig]:
if self.strategy == FallbackStrategy.PRIORITY:
return sorted(self.models, key=lambda x: x.priority)
elif self.strategy == FallbackStrategy.COST_FIRST:
return sorted(self.models, key=lambda x: x.cost_per_mtok)
return self.models
@dataclass
class RequestMetrics:
model: str
latency_ms: float
tokens_used: int
success: bool
error_type: Optional[str] = None
class HolySheepFallbackClient:
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, chain: FallbackChain):
self.api_key = api_key
self.chain = chain
self.metrics: List[RequestMetrics] = []
self._session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=60)
self._session = aiohttp.ClientSession(timeout=timeout)
return self
async def __aexit__(self, *args):
if self._session:
await self._session.close()
async def chat_completion(
self,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 2048
) -> Dict[str, Any]:
"""폴백 체인을 따른 다중 모델 API 호출"""
last_error = None
attempted_models = []
for model_config in self.chain.get_ordered_models():
attempted_models.append(model_config.model)
try:
result = await self._call_model(
model_config, messages, temperature, max_tokens
)
# 메트릭 기록
self.metrics.append(RequestMetrics(
model=model_config.model,
latency_ms=result.get('latency_ms', 0),
tokens_used=result.get('usage', {}).get('total_tokens', 0),
success=True
))
return {
**result,
'fallback_attempted': len(attempted_models) > 1,
'attempted_models': attempted_models,
'final_model': model_config.model
}
except RateLimitError as e:
logger.warning(f"Rate limit on {model_config.model}: {e}")
last_error = e
continue
except ServerError as e:
logger.warning(f"Server error on {model_config.model}: {e}")
last_error = e
continue
except TimeoutError as e:
logger.warning(f"Timeout on {model_config.model}: {e}")
last_error = e
continue
except AuthenticationError:
# 인증 에러는 폴백 불가, 즉시 실패
raise
# 모든 모델 실패
raise AllModelsFailedError(
f"All models failed. Attempted: {attempted_models}",
last_error
)
async def _call_model(
self,
config: ModelConfig,
messages: List[Dict[str, str]],
temperature: float,
max_tokens: int
) -> Dict[str, Any]:
"""단일 모델 API 호출"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# HolySheep는 provider/model 형식 지원
model_id = f"{config.provider}/{config.model}"
payload = {
"model": model_id,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
start_time = time.perf_counter()
async with self._session.post(
f"{self.BASE_URL}/chat/completions",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=config.timeout_ms / 1000)
) as response:
latency_ms = (time.perf_counter() - start_time) * 1000
if response.status == 429:
raise RateLimitError("Rate limit exceeded")
if response.status >= 500:
raise ServerError(f"Server error: {response.status}")
if response.status == 400:
# Bad Request - 일부 모델만 지원 기능
error_body = await response.json()
raise UnsupportedFeatureError(error_body.get('error', {}).get('message', ''))
if response.status == 401:
raise AuthenticationError("Invalid API key")
if response.status != 200:
raise APIError(f"Unexpected status: {response.status}")
data = await response.json()
return {
**data,
'latency_ms': latency_ms,
'model_config': config
}
커스텀 예외 클래스
class RateLimitError(Exception):
retry_after: Optional[int] = None
class ServerError(Exception):
pass
class TimeoutError(Exception):
pass
class AuthenticationError(Exception):
pass
class UnsupportedFeatureError(Exception):
pass
class APIError(Exception):
pass
class AllModelsFailedError(Exception):
def __init__(self, message, last_error):
super().__init__(message)
self.last_error = last_error
사용 예시
async def main():
chain = FallbackChain(
models=[
ModelConfig(provider="openai", model="gpt-4.1", cost_per_mtok=8.00, priority=1),
ModelConfig(provider="anthropic", model="claude-sonnet-4-20250514", cost_per_mtok=15.00, priority=2),
ModelConfig(provider="google", model="gemini-2.0-flash-exp", cost_per_mtok=2.50, priority=3),
ModelConfig(provider="deepseek", model="deepseek-chat-v3-0324", cost_per_mtok=0.42, priority=4),
],
strategy=FallbackStrategy.PRIORITY
)
async with HolySheepFallbackClient("YOUR_HOLYSHEEP_API_KEY", chain) as client:
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Explain async/await in Python with an example."}
]
try:
result = await client.chat_completion(messages)
print(f"Response from: {result['final_model']}")
print(f"Latency: {result['latency_ms']:.2f}ms")
print(f"Cost estimate: ${result.get('usage', {}).get('total_tokens', 0) / 1_000_000 * chain.models[0].cost_per_mtok:.6f}")
except AllModelsFailedError as e:
print(f"All models failed: {e}")
if __name__ == "__main__":
asyncio.run(main())TypeScript 구현: Circuit Breaker 패턴
고가용성 시스템을 위해 Circuit Breaker 패턴을 적용합니다. 특정 모델의 에러율이 높아지면 해당 모델을 일시적으로 우회합니다:
// fallback-with-circuit-breaker.ts
import { EventEmitter } from 'events';
interface ModelEndpoint {
provider: string;
model: string;
costPerMTok: number;
}
interface CircuitState {
status: 'CLOSED' | 'OPEN' | 'HALF_OPEN';
failureCount: number;
successCount: number;
lastFailureTime: number;
consecutiveTimeouts: number;
}
interface FallbackConfig {
failureThreshold: number; // 회로を開く失敗閾値 (default: 5)
recoveryTimeout: number; // 복구 시도 간격 (ms, default: 30000)
halfOpenSuccessThreshold: number; // half-open → closed 전환 성공 횟수
timeoutThreshold: number; // 타임아웃 연속 횟수 임계값
cooldownAfterOpen: number; // OPEN 상태 유지 시간 (ms)
}
class CircuitBreaker {
private state: CircuitState;
private config: FallbackConfig;
constructor(config: Partial = {}) {
this.config = {
failureThreshold: config.failureThreshold ?? 5,
recoveryTimeout: config.recoveryTimeout ?? 30000,
halfOpenSuccessThreshold: config.halfOpenSuccessThreshold ?? 3,
timeoutThreshold: config.timeoutThreshold ?? 3,
cooldownAfterOpen: config.cooldownAfterOpen ?? 60000,
};
this.state = {
status: 'CLOSED',
failureCount: 0,
successCount: 0,
lastFailureTime: 0,
consecutiveTimeouts: 0,
};
}
canExecute(): boolean {
if (this.state.status === 'CLOSED') return true;
if (this.state.status === 'OPEN') {
const now = Date.now();
if (now - this.state.lastFailureTime >= this.config.cooldownAfterOpen) {
this.state.status = 'HALF_OPEN';
this.state.successCount = 0;
return true;
}
return false;
}
// HALF_OPEN: 제한된 요청만 허용
return true;
}
recordSuccess(): void {
if (this.state.status === 'HALF_OPEN') {
this.state.successCount++;
if (this.state.successCount >= this.config.halfOpenSuccessThreshold) {
this.state.status = 'CLOSED';
this.state.failureCount = 0;
this.state.consecutiveTimeouts = 0;
}
} else {
this.state.failureCount = 0;
this.state.consecutiveTimeouts = 0;
}
}
recordFailure(isTimeout: boolean = false): void {
this.state.lastFailureTime = Date.now();
if (isTimeout) {
this.state.consecutiveTimeouts++;
if (this.state.consecutiveTimeouts >= this.config.timeoutThreshold) {
this.state.status = 'OPEN';
return;
}
}
this.state.failureCount++;
if (this.state.failureCount >= this.config.failureThreshold) {
this.state.status = 'OPEN';
}
}
getStatus() {
return { ...this.state };
}
}
interface FallbackMetrics {
totalRequests: number;
successfulRequests: number;
failedRequests: number;
fallbackTriggered: number;
averageLatencyMs: number;
costEstimate: number;
modelStats: Record;
}
class MultiModelFallbackClient extends EventEmitter {
private baseUrl = 'https://api.holysheep.ai/v1';
private circuitBreakers: Map = new Map();
private metrics: FallbackMetrics;
constructor(
private apiKey: string,
private endpoints: ModelEndpoint[]
) {
super();
// 각 모델별 Circuit Breaker 초기화
for (const endpoint of endpoints) {
const key = ${endpoint.provider}/${endpoint.model};
this.circuitBreakers.set(key, new CircuitBreaker());
}
this.metrics = {
totalRequests: 0,
successfulRequests: 0,
failedRequests: 0,
fallbackTriggered: 0,
averageLatencyMs: 0,
costEstimate: 0,
modelStats: {},
};
for (const endpoint of endpoints) {
const key = ${endpoint.provider}/${endpoint.model};
this.metrics.modelStats[key] = {
attempts: 0,
successes: 0,
failures: 0,
avgLatencyMs: 0,
};
}
}
async complete(
messages: Array<{ role: string; content: string }>,
options: {
temperature?: number;
maxTokens?: number;
priorityOrder?: string[]; // 커스텀 순서
} = {}
): Promise<{
content: string;
model: string;
latencyMs: number;
usage: { promptTokens: number; completionTokens: number };
fallbackAttempted: boolean;
costEstimate: number;
}> {
const { temperature = 0.7, maxTokens = 2048, priorityOrder } = options;
// 순서 결정
let orderedEndpoints = [...this.endpoints];
if (priorityOrder) {
orderedEndpoints = priorityOrder
.map(id => this.endpoints.find(e => ${e.provider}/${e.model} === id))
.filter(Boolean) as ModelEndpoint[];
}
// Circuit Breaker 상태 고려
orderedEndpoints = orderedEndpoints.filter(ep => {
const key = ${ep.provider}/${ep.model};
const cb = this.circuitBreakers.get(key)!;
return cb.canExecute();
});
let lastError: Error | null = null;
for (const endpoint of orderedEndpoints) {
const key = ${endpoint.provider}/${endpoint.model};
const cb = this.circuitBreakers.get(key)!;
const startTime = performance.now();
this.metrics.totalRequests++;
this.metrics.modelStats[key].attempts++;
try {
const result = await this.callAPI(endpoint, messages, temperature, maxTokens);
const latencyMs = performance.now() - startTime;
cb.recordSuccess();
this.metrics.successfulRequests++;
this.metrics.modelStats[key].successes++;
this.updateAverageLatency(latencyMs);
const usage = result.usage || { prompt_tokens: 0, completion_tokens: 0 };
const totalTokens = usage.prompt_tokens + usage.completion_tokens;
const costEstimate = (totalTokens / 1_000_000) * endpoint.costPerMTok;
this.metrics.costEstimate += costEstimate;
const isFallback = this.metrics.fallbackTriggered > 0;
return {
content: result.choices[0]?.message?.content || '',
model: key,
latencyMs,
usage: {
promptTokens: usage.prompt_tokens,
completionTokens: usage.completion_tokens,
},
fallbackAttempted: isFallback,
costEstimate,
};
} catch (error: any) {
const latencyMs = performance.now() - startTime;
lastError = error;
this.metrics.modelStats[key].failures++;
this.metrics.modelStats[key].avgLatencyMs =
(this.metrics.modelStats[key].avgLatencyMs + latencyMs) / 2;
const isTimeout = error.code === 'ETIMEDOUT' || error.message?.includes('timeout');
cb.recordFailure(isTimeout);
// 401/403 에러는 폴백 불필요
if (error.status === 401 || error.status === 403) {
throw error;
}
// 다음 모델 시도
this.metrics.fallbackTriggered++;
this.emit('fallback', { from: key, error: error.message, next: orderedEndpoints.indexOf(endpoint) + 1 });
continue;
}
}
this.metrics.failedRequests++;
throw new Error(All models failed. Last error: ${lastError?.message});
}
private async callAPI(
endpoint: ModelEndpoint,
messages: Array<{ role: string; content: string }>,
temperature: number,
maxTokens: number
): Promise {
const response = await fetch(${this.baseUrl}/chat/completions, {
method: 'POST',
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json',
},
body: JSON.stringify({
model: ${endpoint.provider}/${endpoint.model},
messages,
temperature,
max_tokens: maxTokens,
}),
});
if (response.status === 429) {
const error = new Error('Rate limit exceeded');
(error as any).status = 429;
throw error;
}
if (response.status >= 500) {
const error = new Error(Server error: ${response.status});
(error as any).status = response.status;
throw error;
}
if (response.status !== 200) {
const error = new Error(API error: ${response.status});
(error as any).status = response.status;
throw error;
}
return response.json();
}
private updateAverageLatency(newLatency: number): void {
const total = this.metrics.averageLatencyMs * (this.metrics.successfulRequests - 1);
this.metrics.averageLatencyMs = (total + newLatency) / this.metrics.successfulRequests;
}
getMetrics(): FallbackMetrics {
return { ...this.metrics };
}
getCircuitBreakerStatus(): Record {
const status: Record = {};
for (const [key, cb] of this.circuitBreakers) {
status[key] = cb.getStatus();
}
return status;
}
}
// 사용 예시
async function demo() {
const client = new MultiModelFallbackClient(
'YOUR_HOLYSHEEP_API_KEY',
[
{ provider: 'openai', model: 'gpt-4.1', costPerMTok: 8.00 },
{ provider: 'anthropic', model: 'claude-sonnet-4-20250514', costPerMTok: 15.00 },
{ provider: 'google', model: 'gemini-2.0-flash-exp', costPerMTok: 2.50 },
{ provider: 'deepseek', model: 'deepseek-chat-v3-0324', costPerMTok: 0.42 },
]
);
// 폴백 이벤트 리스너
client.on('fallback', ({ from, error, next }) => {
console.log(Fallback: ${from} failed (${error}), trying #${next});
});
try {
const result = await client.complete([
{ role: 'system', content: 'You are a code reviewer.' },
{ role: 'user', content: 'Review this Python function for security issues.' },
]);
console.log(Response from: ${result.model});
console.log(Latency: ${result.latencyMs.toFixed(2)}ms);
console.log(Cost: $${result.costEstimate.toFixed(6)});
} catch (error) {
console.error('All models failed:', error);
}
// 메트릭 확인
console.log('\n--- Metrics ---');
console.log(client.getMetrics());
console.log('\n--- Circuit Breakers ---');
console.log(client.getCircuitBreakerStatus());
} 동시성 제어와 레이트 리밋 관리
폴백 시스템에서 동시성 제어가 없으면 thundering herd 문제가 발생합니다. 여러 요청이 동시에 실패하고 같은 모델로杀到하면 추가 rate limit을触发합니다.
# rate_limiter.py
import asyncio
import time
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, Optional
import threading
@dataclass
class RateLimitConfig:
requests_per_minute: int = 60
requests_per_second: int = 10
burst_size: int = 20
model_specific: Dict[str, Dict[str, int]] = field(default_factory=dict)
class TokenBucketRateLimiter:
"""토큰 버킷 알고리즘 기반 레이트 리미터"""
def __init__(self, config: RateLimitConfig):
self.config = config
self.tokens: Dict[str, float] = defaultdict(lambda: config.burst_size)
self.last_refill: Dict[str, float] = defaultdict(time.time)
self.rpm_buckets: Dict[str, list] = defaultdict(list)
self.lock = asyncio.Lock()
async def acquire(self, model_key: str) -> float:
"""토큰 확보, 확보까지 대기 시간 반환"""
async with self.lock:
# RPM 체크
now = time.time()
self._cleanup_rpm_bucket(model_key, now)
max_rpm = self.config.model_specific.get(model_key, {}).get('rpm', self.config.requests_per_minute)
if len(self.rpm_buckets[model_key]) >= max_rpm:
wait_time = 60 - (now - self.rpm_buckets[model_key][0])
if wait_time > 0:
await asyncio.sleep(wait_time)
now = time.time()
self._cleanup_rpm_bucket(model_key, now)
# 토큰 버킷 refill
self._refill_bucket(model_key, now)
# 토큰消費
if self.tokens[model_key] < 1:
wait_time = (1 - self.tokens[model_key]) / self._get_refill_rate(model_key)
await asyncio.sleep(wait_time)
self._refill_bucket(model_key, time.time())
self.tokens[model_key] -= 1
self.rpm_buckets[model_key].append(time.time())
return self.tokens[model_key]
def _refill_bucket(self, model_key: str, now: float):
elapsed = now - self.last_refill[model_key]
refill_rate = self._get_refill_rate(model_key)
self.tokens[model_key] = min(
self.config.burst_size,
self.tokens[model_key] + elapsed * refill_rate
)
self.last_refill[model_key] = now
def _get_refill_rate(self, model_key: str) -> float:
rps = self.config.model_specific.get(model_key, {}).get('rps', self.config.requests_per_second)
return rps
def _cleanup_rpm_bucket(self, model_key: str, now: float):
cutoff = now - 60
self.rpm_buckets[model_key] = [
t for t in self.rpm_buckets[model_key] if t > cutoff
]
class RequestQueue:
"""우선순위 기반 요청 큐"""
def __init__(self, max_concurrent: int = 10):
self.max_concurrent = max_concurrent
self.active_requests = 0
self.queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
self.lock = asyncio.Lock()
async def enqueue(self, priority: int, coro):
"""우선순위 큐에 등록 (낮은 숫자 = 높은 우선순위)"""
await self.queue.put((priority, coro))
return await self._process_queue()
async def _process_queue(self):
while self.active_requests >= self.max_concurrent:
await asyncio.sleep(0.1)
async with self.lock:
if not self.queue.empty() and self.active_requests < self.max_concurrent:
self.active_requests += 1
priority, coro = await self.queue.get()
try:
result = await coro
return result
finally:
self.active_requests -= 1
self.queue.task_done()
통합 Rate Limit Manager
class RateLimitManager:
def __init__(self, config: RateLimitConfig):
self.limiters: Dict[str, TokenBucketRateLimiter] = {}
self.queue = RequestQueue()
self.config = config
def get_limiter(self, model_key: str) -> TokenBucketRateLimiter:
if model_key not in self.limiters:
self.limiters[model_key] = TokenBucketRateLimiter(self.config)
return self.limiters[model_key]
async def execute_with_limit(self, model_key: str, priority: int, coro):
limiter = self.get_limiter(model_key)
await limiter.acquire(model_key)
return await coro
HolySheep 폴백 클라이언트에 통합
class HolySheepResilientClient:
def __init__(self, api_key: str, endpoints: list, rate_limit_config: RateLimitConfig):
self.fallback_client = HolySheepFallbackClient(api_key, endpoints)
self.rate_manager = RateLimitManager(rate_limit_config)
async def complete(self, messages, priority: int = 5, options: dict = {}):
# 모든 모델에 레이트 리밋 적용
tasks = []
for endpoint in self.fallback_client.chain.get_ordered_models():
model_key = f"{endpoint.provider}/{endpoint.model}"
task = self.rate_manager.execute_with_limit(
model_key,
priority,
self.fallback_client._call_model(endpoint, messages, options.get('temperature', 0.7), options.get('max_tokens', 2048))
)
tasks.append((endpoint.priority, model_key, task))
# 우선순위순으로 실행
tasks.sort(key=lambda x: x[0])
for priority, model_key, task in tasks:
try:
return await task
except (RateLimitError, ServerError, TimeoutError):
continue
raise AllModelsFailedError("All models failed", None)비용 최적화 전략
폴백 시스템에서 비용 관리는 중요합니다. 단순히 실패 시 다음 모델로 전환하면 비용이 불필요하게 증가할 수 있습니다.
비용 인식 폴백
# cost_aware_fallback.py
from dataclasses import dataclass
from typing import Callable, Optional, List
from enum import Enum
class FallbackDecision(Enum):
USE_CURRENT = "use_current"
FALLBACK = "fallback"
RETRY_SAME = "retry_same"
FAIL = "fail"
@dataclass
class CostBudget:
max_cost_per_request: float = 0.50 # 요청당 최대 비용
max_cost_per_hour: float = 100.0 # 시간당 최대 비용
fallback_cost_threshold: float = 0.10 # 폴백 허용 추가 비용
class CostAwareRouter:
def __init__(self, budget: CostBudget):
self.budget = budget
self.hourly_spend: List[tuple] = [] # (timestamp, cost)
self.total_fallback_cost: float = 0
def should_fallback(
self,
current_model: str,
current_cost: float,
fallback_model: str,
fallback_cost: float,
attempt: int
) -> FallbackDecision:
# 첫 시도 실패는 동일한 모델 재시도
if attempt == 0:
return FallbackDecision.RETRY_SAME
# 최대 비용 초과 시 즉시 실패
if current_cost > self.budget.max_cost_per_request:
return FallbackDecision.FAIL
# 폴백 추가 비용 계산
additional_cost = fallback_cost - current_cost
if additional_cost > self.budget.fallback_cost_threshold:
return FallbackDecision.FAIL
# 시간당 예산 확인
self._cleanup_hourly_spend()
current_hourly = sum(cost for _, cost in self.hourly_spend)
if current_hourly + additional_cost > self.budget.max_cost_per_hour:
return FallbackDecision.FAIL
return FallbackDecision.FALLBACK
def record_cost(self, model: str, cost: float):
self.hourly_spend.append((time.time(), cost))
self.total_fallback_cost += cost
def _cleanup_hourly_spend(self):
cutoff = time.time() - 3600
self.hourly_spend = [(ts, cost) for ts, cost in self.hourly_spend if ts > cutoff]
def get_cost_stats(self) -> dict:
self._cleanup_hourly_spend()
return {
'total_fallback_cost': self.total_fallback_cost,
'current_hourly_spend': sum(c for _, c in self.hourly_spend),
'requests_this_hour': len(self.hourly_spend),
}
모델별 비용 최적화 가중치
COST_EFFICIENCY_WEIGHTS = {
'deepseek/deepseek-chat-v3-0324': 1.0, # 최고 효율
'google/gemini-2.0-flash-exp': 0.85,
'openai/gpt-4.1': 0.6,
'anthropic/claude-sonnet-4-20250514': 0.4,
}
def calculate_cost_score(model: str, latency_ms: float) -> float:
"""비용-성능 점수 계산 (높을수록 좋음)"""
efficiency = COST_EFFICIENCY_WEIGHTS.get(model, 0.5)
# latency_score: 1000ms 이하에서 점수 증가, 이상이면 감소
latency_score = max(0, 1 - (latency_ms - 1000) / 2000)
return efficiency * 0.6 + latency_score * 0.4
모델 선택 최적화
def select_optimal_model(
available_models: List[dict],
use_case: str =