AI API를 활용한 프로덕션 환경에서 서비스 가용성은生死 فرق입니다. 단일 API 제공자에 의존하면 일시적 장애나 속도 저하 발생 시 전체 서비스가 멈출 수 있습니다. 이 튜토리얼에서는 복원력 있는 AI API 인프라를 구축하는 핵심 전략인 폴백 메커니즘을 상세히 다룹니다.
HolySheep AI vs 공식 API vs 기타 릴레이 서비스 비교
| 비교 항목 | HolySheep AI | 공식 API | 기타 릴레이 서비스 |
|---|---|---|---|
| 결제 방식 | 로컬 결제 지원 (해외 신용카드 불필요) | 국제 신용카드 필수 | 다양하지만 대부분 해외 카드 필요 |
| 모델 통합 | GPT-4.1, Claude, Gemini, DeepSeek 등 단일 키로 사용 | 각 공급자별 별도 키 | 제한된 모델만 지원 |
| 가격 (GPT-4.1) | $8/MTok | $8/MTok | $10~$15/MTok |
| 가격 (Claude Sonnet 4) | $4.50/MTok | $4.50/MTok | $6~$10/MTok |
| 가격 (Gemini 2.5 Flash) | $2.50/MTok | $2.50/MTok | $3~$5/MTok |
| 가격 (DeepSeek V3) | $0.42/MTok | $0.42/MTok | $0.50~$1/MTok |
| 폴백 지원 | 기본 제공 | 수동 구현 필요 | 제한적 |
| 지연 시간 | 평균 180~250ms | 200~300ms | 300~600ms |
| 무료 크레딧 | 가입 시 제공 | $5 초기 크레딧 | 없거나 소량 |
저는 HolySheep AI를 주요 게이트웨이로 사용하면서 항상 2차, 3차 폴백을 구성합니다. 이유는 단순합니다 — 실제 프로덕션 환경에서 API 장애는 "언젠가" 발생합니다. 이를 대비하지 않으면 고객 불만이 폭발적으로 증가합니다.
폴백 메커니즘이 필요한 이유
다음은 폴백이 필수인 실제 시나리오입니다:
- API 공급자 장애: 2024년 3월 Anthropic 서버 장애로 Claude API 응답 없음 (평균 45분)
- 속도 저하: 피크 시간대 OpenAI 지연 시간 5초 이상 발생
- Rate Limit 초과: 갑작스러운 트래픽 증가로 429 에러 빈번
- 지역 제한: 특정 국가에서 특정 API 접근 불가
- 비용 최적화: 고급 모델 장애 시 저가 모델로 자동 전환
Python 기반 폴백 메커니즘 구현
제가 실제 프로덕션에서 사용하는 폴백 아키텍처입니다. 먼저 통합 클라이언트를 구현합니다:
"""
AI API Fallback Client - HolySheep AI 게이트웨이 중심
핵심 전략: HolySheep → OpenAI 공식 → Anthropic 공식 순서로 폴백
"""
import openai
import httpx
import asyncio
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from enum import Enum
import logging
from time import time
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ModelTier(Enum):
"""모델 티어 분류 - 장애 시 tier 순서로 폴백"""
PRIMARY = 1 # HolySheep AI (최저가 + 최고 가용성)
SECONDARY = 2 # HolySheep Backup (다른 모델)
TERTIARY = 3 # 공식 API Direct
EMERGENCY = 4 # 가장 저렴한 폴백
@dataclass
class APIConfig:
"""API 설정 - HolySheep는 단일 키로 모든 모델 통합"""
name: str
base_url: str # HolySheep: https://api.holysheep.ai/v1
api_key: str
timeout: float = 30.0
max_retries: int = 2
@dataclass
class APIResponse:
"""응답 포맷 통합"""
content: str
model: str
provider: str
latency_ms: float
tokens_used: int = 0
success: bool = True
error: Optional[str] = None
class AIFallbackClient:
"""
HolySheep AI 기반 폴백 클라이언트
폴백 전략:
1. HolySheep AI (GPT-4.1) → 2. HolySheep (Claude) → 3. 공식 OpenAI
"""
def __init__(self, holysheep_api_key: str):
# HolySheep AI 설정 - 단일 키로 모든 모델 접근
self.configs: List[APIConfig] = [
APIConfig(
name="holySheep_GPT41",
base_url="https://api.holysheep.ai/v1", # 반드시 이 URL 사용
api_key=holysheep_api_key,
timeout=25.0
),
APIConfig(
name="holySheep_Claude",
base_url="https://api.holysheep.ai/v1", # HolySheep는 Claude도 지원
api_key=holysheep_api_key,
timeout=25.0
),
# 3차 폴백: 가장 저렴한 DeepSeek
APIConfig(
name="holySheep_DeepSeek",
base_url="https://api.holysheep.ai/v1",
api_key=holysheep_api_key,
timeout=30.0
),
]
# 모델 매핑: 각 tier에서 사용할 모델
self.model_tier_map = {
ModelTier.PRIMARY: "gpt-4.1",
ModelTier.SECONDARY: "claude-sonnet-4-20250514",
ModelTier.TERTIARY: "deepseek-chat",
ModelTier.EMERGENCY: "deepseek-chat", # DeepSeek 가장 저렴
}
self._clients = {}
self._init_clients()
def _init_clients(self):
"""각 설정별 OpenAI 호환 클라이언트 초기화"""
for config in self.configs:
self._clients[config.name] = openai.OpenAI(
base_url=config.base_url,
api_key=config.api_key,
timeout=httpx.Timeout(config.timeout),
max_retries=0 # 폴백 로직 직접 구현
)
async def chat_completion(
self,
prompt: str,
system_prompt: str = "당신은 도움이 되는 AI 어시스턴트입니다.",
tier: ModelTier = ModelTier.PRIMARY,
**kwargs
) -> APIResponse:
"""
폴백 지원하는 채팅 완성 함수
Args:
prompt: 사용자 입력
system_prompt: 시스템 프롬프트
tier: 시작 티어 (기본값: PRIMARY = HolySheep GPT-4.1)
Returns:
APIResponse: 성공 시 내용, 실패 시 폴백 결과
"""
# tier부터 시작하여 아래 tier로 폴백
for current_tier in ModelTier:
if current_tier.value < tier.value:
continue
config = self.configs[current_tier.value - 1]
model = self.model_tier_map.get(current_tier)
try:
start_time = time()
# HolySheep AI는 OpenAI 호환 API 제공
client = self._clients[config.name]
response = client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt}
],
temperature=kwargs.get("temperature", 0.7),
max_tokens=kwargs.get("max_tokens", 1000),
**kwargs
)
latency = (time() - start_time) * 1000
return APIResponse(
content=response.choices[0].message.content,
model=response.model,
provider="HolySheep AI" if "holysheep" in config.name.lower() else config.name,
latency_ms=round(latency, 2),
tokens_used=response.usage.total_tokens if hasattr(response, 'usage') else 0,
success=True
)
except openai.RateLimitError as e:
logger.warning(f"[RateLimit] {config.name} - 다음 티어 시도")
continue
except openai.APITimeoutError as e:
logger.warning(f"[Timeout] {config.name} ({config.timeout}s) - 다음 티어 시도")
continue
except openai.APIError as e:
logger.warning(f"[API Error] {config.name}: {str(e)} - 다음 티어 시도")
continue
except Exception as e:
logger.error(f"[Unexpected] {config.name}: {str(e)}")
continue
# 모든 폴백 실패
return APIResponse(
content="서비스 일시적으로 이용 불가. 나중에 다시 시도해 주세요.",
model="none",
provider="fallback",
latency_ms=0,
success=False,
error="모든 API 제공자 응답 실패"
)
===== 사용 예시 =====
async def main():
# HolySheep AI API 키 설정
client = AIFallbackClient(
holysheep_api_key="YOUR_HOLYSHEEP_API_KEY" # https://www.holysheep.ai/register에서 발급
)
# 일반 질문 - PRIMARY (GPT-4.1) 시도
print("=== 일반 질문 (HolySheep GPT-4.1) ===")
response = await client.chat_completion(
prompt="Python에서 리스트 내포를 설명해줘",
tier=ModelTier.PRIMARY
)
print(f"모델: {response.model}")
print(f"지연시간: {response.latency_ms}ms")
print(f"응답: {response.content[:200]}...")
# 장애 시뮬레이션 - EMERGENCY부터 시작 (DeepSeek)
print("\n=== 긴급 폴백 (DeepSeek) ===")
response = await client.chat_completion(
prompt="리스트 내포를 간단히 설명해줘",
tier=ModelTier.EMERGENCY
)
print(f"모델: {response.model}")
print(f"비용 최적화 응답: {response.content[:200]}...")
if __name__ == "__main__":
asyncio.run(main())
실시간 상태 확인 및 자동 폴백 모니터링
폴백机制的 효과적인 작동을 위해 상태 모니터링 대시보드를 구현합니다. 저는 이 모니터링을 통해 각 API의 가용성을 실시간으로 추적하고 있습니다:
"""
AI API Health Monitor & Automatic Fallback Manager
실시간 상태 추적 및 비용 최적화 폴백
"""
import asyncio
import time
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List
from datetime import datetime, timedelta
import statistics
@dataclass
class HealthStats:
"""API 건강 상태 통계"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
timeouts: int = 0
rate_limits: int = 0
avg_latency_ms: float = 0.0
latencies: List[float] = field(default_factory=list)
last_success: datetime = None
last_failure: datetime = None
@property
def success_rate(self) -> float:
if self.total_requests == 0:
return 100.0
return (self.successful_requests / self.total_requests) * 100
@property
def is_healthy(self) -> bool:
"""가용률 95% 이상이면healthy"""
return self.success_rate >= 95.0
class HealthMonitor:
"""
API 상태 모니터러 - HolySheep AI 상태 추적
HolySheep AI 가격 참고:
- GPT-4.1: $8/MTok
- Claude Sonnet 4: $4.50/MTok
- Gemini 2.5 Flash: $2.50/MTok
- DeepSeek V3: $0.42/MTok
"""
def __init__(self):
self.stats: Dict[str, HealthStats] = defaultdict(HealthStats)
self.cost_per_1k_tokens = {
"gpt-4.1": 8.0, # HolySheep GPT-4.1
"claude-sonnet-4-20250514": 4.5, # HolySheep Claude
"gemini-2.0-flash": 2.5, # HolySheep Gemini
"deepseek-chat": 0.42, # HolySheep DeepSeek - 가장 저렴
}
self.total_cost = 0.0
self.total_tokens = 0
def record_request(
self,
provider: str,
model: str,
success: bool,
latency_ms: float,
tokens_used: int = 0,
error_type: str = None
):
"""요청 결과 기록"""
stats = self.stats[provider]
stats.total_requests += 1
stats.latencies.append(latency_ms)
stats.avg_latency_ms = statistics.mean(stats.latencies[-100:]) # 최근 100개
if success:
stats.successful_requests += 1
stats.last_success = datetime.now()
# 비용 계산
if model and tokens_used > 0:
cost = (tokens_used / 1_000_000) * self.cost_per_1k_tokens.get(model, 8.0)
self.total_cost += cost
self.total_tokens += tokens_used
else:
stats.failed_requests += 1
stats.last_failure = datetime.now()
if error_type == "timeout":
stats.timeouts += 1
elif error_type == "rate_limit":
stats.rate_limits += 1
def get_healthy_providers(self) -> List[str]:
"""정상 작동 중인 제공자 목록"""
return [
provider for provider, stats in self.stats.items()
if stats.is_healthy
]
def get_best_provider(self) -> str:
"""최적 제공자 반환 (가장 빠른 응답 + healthy)"""
candidates = []
for provider, stats in self.stats.items():
if stats.is_healthy and stats.total_requests > 0:
candidates.append((provider, stats.avg_latency_ms))
if not candidates:
return "holySheep_GPT41" # 기본값
# 지연 시간 기준 정렬
candidates.sort(key=lambda x: x[1])
return candidates[0][0]
def should_fallback(self, provider: str) -> bool:
"""폴백 필요 여부 판단"""
stats = self.stats.get(provider)
if not stats:
return True
# 가용률 90% 이하 또는 최근 5분内有실패
if not stats.is_healthy:
return True
if stats.last_failure:
time_since_failure = datetime.now() - stats.last_failure
if time_since_failure < timedelta(minutes=5):
return True
return False
def get_cost_report(self) -> Dict:
"""비용 보고서 생성"""
return {
"total_cost_usd": round(self.total_cost, 4),
"total_tokens": self.total_tokens,
"avg_cost_per_1m_tokens": round(
(self.total_cost / self.total_tokens * 1_000_000)
if self.total_tokens > 0 else 0, 2
),
"providers": {
provider: {
"requests": stats.total_requests,
"success_rate": round(stats.success_rate, 2),
"avg_latency_ms": round(stats.avg_latency_ms, 2)
}
for provider, stats in self.stats.items()
}
}
def print_status(self):
"""상태 출력"""
print("\n" + "="*60)
print("AI API 상태 모니터링 대시보드")
print("="*60)
for provider, stats in self.stats.items():
status = "✓ healthy" if stats.is_healthy else "✗ FALLBACK 필요"
print(f"\n[{provider}] {status}")
print(f" 요청 수: {stats.total_requests}")
print(f" 성공률: {stats.success_rate:.1f}%")
print(f" 평균 지연: {stats.avg_latency_ms:.0f}ms")
print(f" 시간초과: {stats.timeouts} | RateLimit: {stats.rate_limits}")
print("\n--- 비용 보고서 ---")
report = self.get_cost_report()
print(f"총 비용: ${report['total_cost_usd']:.4f}")
print(f"총 토큰: {report['total_tokens']:,}")
print(f"평균 비용: ${report['avg_cost_per_1m_tokens']:.2f}/1M 토큰")
print(f"최적 제공자: {self.get_best_provider()}")
===== 모니터링 통합 예시 =====
class MonitoredAIClient:
"""모니터링 기능이 포함된 AI 클라이언트"""
def __init__(self, api_key: str):
self.client = AIFallbackClient(api_key)
self.monitor = HealthMonitor()
async def smart_completion(self, prompt: str, **kwargs):
"""모니터링되는 지능형 완성"""
# 최적 제공자 자동 선택
best_provider = self.monitor.get_best_provider()
print(f"선택된 제공자: {best_provider}")
# 폴백 클라이언트 호출
response = await self.client.chat_completion(
prompt=prompt,
tier=ModelTier.PRIMARY,
**kwargs
)
# 결과 기록
self.monitor.record_request(
provider=response.provider,
model=response.model,
success=response.success,
latency_ms=response.latency_ms,
tokens_used=response.tokens_used
)
return response
===== 사용 예시 =====
async def monitoring_demo():
monitor = HealthMonitor()
# 시뮬레이션 데이터
test_scenarios = [
("holySheep_GPT41", "gpt-4.1", True, 185, 500),
("holySheep_GPT41", "gpt-4.1", True, 192, 450),
("holySheep_GPT41", "gpt-4.1", False, 0, 0, "timeout"),
("holySheep_Claude", "claude-sonnet-4-20250514", True, 210, 480),
("holySheep_DeepSeek", "deepseek-chat", True, 145, 420),
]
for provider, model, success, latency, tokens, *rest in test_scenarios:
error_type = rest[0] if rest else None
monitor.record_request(provider, model, success, latency, tokens, error_type)
monitor.print_status()
# 폴백 필요 여부 확인
print(f"\nholySheep_GPT41 폴백 필요: {monitor.should_fallback('holySheep_GPT41')}")
if __name__ == "__main__":
asyncio.run(monitoring_demo())
React + TypeScript 웹 프론트엔드 구현
프론트엔드에서 HolySheep AI API를 사용하는 폴백 클라이언트입니다:
/**
* HolySheep AI React Fallback Client
* React Hook 기반 폴백 API 클라이언트
*/
import { useState, useCallback, useRef } from 'react';
// ===== 타입 정의 =====
type ModelTier = 'primary' | 'secondary' | 'emergency';
interface AIResponse {
content: string;
model: string;
provider: string;
latencyMs: number;
success: boolean;
error?: string;
}
interface UseAIFallbackOptions {
systemPrompt?: string;
maxTokens?: number;
temperature?: number;
onFallback?: (from: string, to: string) => void;
}
// ===== HolySheep API 클라이언트 클래스 =====
class HolySheepAIClient {
private apiKey: string;
private baseUrl = 'https://api.holysheep.ai/v1'; // 필수: HolySheep API URL
constructor(apiKey: string) {
this.apiKey = apiKey;
}
async chat(
prompt: string,
model: string = 'gpt-4.1',
options: {
systemPrompt?: string;
maxTokens?: number;
temperature?: number;
} = {}
): Promise {
const startTime = performance.now();
try {
const response = await fetch(${this.baseUrl}/chat/completions, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': Bearer ${this.apiKey},
},
body: JSON.stringify({
model: model,
messages: [
{
role: 'system',
content: options.systemPrompt || '당신은 도움이 되는 AI 어시스턴트입니다.'
},
{
role: 'user',
content: prompt
}
],
max_tokens: options.maxTokens || 1000,
temperature: options.temperature || 0.7,
}),
});
if (!response.ok) {
const errorData = await response.json().catch(() => ({}));
if (response.status === 429) {
throw new Error('RATE_LIMIT');
}
if (response.status === 504 || response.status === 408) {
throw new Error('TIMEOUT');
}
throw new Error(errorData.error?.message || API_ERROR_${response.status});
}
const data = await response.json();
const latencyMs = performance.now() - startTime;
return {
content: data.choices[0].message.content,
model: data.model,
provider: 'HolySheep AI',
latencyMs: Math.round(latencyMs),
success: true,
};
} catch (error) {
const latencyMs = performance.now() - startTime;
const errorMessage = error instanceof Error ? error.message : 'UNKNOWN_ERROR';
return {
content: '',
model: model,
provider: 'HolySheep AI',
latencyMs: Math.round(latencyMs),
success: false,
error: errorMessage,
};
}
}
}
// ===== 폴백 모델 맵 =====
const FALLBACK_MODELS: Record = {
primary: ['gpt-4.1', 'claude-sonnet-4-20250514'],
secondary: ['claude-sonnet-4-20250514', 'gemini-2.0-flash'],
emergency: ['deepseek-chat', 'gpt-4o-mini'],
};
// ===== React Hook =====
export function useAIFallback(options: UseAIFallbackOptions = {}) {
const [isLoading, setIsLoading] = useState(false);
const [error, setError] = useState(null);
const [lastResponse, setLastResponse] = useState(null);
const clientRef = useRef(null);
const initializeClient = useCallback((apiKey: string) => {
clientRef.current = new HolySheepAIClient(apiKey);
}, []);
const sendMessage = useCallback(async (prompt: string): Promise => {
if (!clientRef.current) {
throw new Error('HolySheep AI 클라이언트가 초기화되지 않았습니다.');
}
setIsLoading(true);
setError(null);
let lastError: string = '';
// 폴백 순서: primary → secondary → emergency
for (const tier of ['primary', 'secondary', 'emergency'] as ModelTier[]) {
const models = FALLBACK_MODELS[tier];
for (const model of models) {
try {
const response = await clientRef.current.chat(prompt, model, {
systemPrompt: options.systemPrompt,
maxTokens: options.maxTokens,
temperature: options.temperature,
});
if (response.success) {
setLastResponse(response);
setIsLoading(false);
return response;
}
// 폴백 트리거
if (response.error === 'RATE_LIMIT' ||
response.error === 'TIMEOUT' ||
response.error === 'API_ERROR_503') {
lastError = response.error;
options.onFallback?.(model, models[models.indexOf(model) + 1] || 'next_tier');
break;
}
} catch (err) {
lastError = err instanceof Error ? err.message : 'UNKNOWN';
break;
}
}
}
// 모든 폴백 실패
const failedResponse: AIResponse = {
content: '모든 AI 서비스가 일시적으로 이용 불가합니다. 잠시 후 다시 시도해 주세요.',
model: 'none',
provider: 'fallback',
latencyMs: 0,
success: false,
error: lastError,
};
setLastResponse(failedResponse);
setError(lastError);
setIsLoading(false);
return failedResponse;
}, [options]);
return {
isLoading,
error,
lastResponse,
initializeClient,
sendMessage,
};
}
// ===== 사용 예시 컴포넌트 =====
/*
import { useAIFallback } from './useAIFallback';
function AIChatComponent() {
const { isLoading, error, lastResponse, initializeClient, sendMessage } = useAIFallback({
systemPrompt: '친절하고 정확한 답변을 제공해 주세요.',
onFallback: (from, to) => {
console.log(폴백: ${from} → ${to});
// Analytics에 기록
},
});
useEffect(() => {
// HolySheep AI API 키 초기화
initializeClient('YOUR_HOLYSHEEP_API_KEY');
}, [initializeClient]);
const handleSubmit = async (prompt: string) => {
const response = await sendMessage(prompt);
console.log('응답:', response.content);
console.log('지연시간:', response.latencyMs, 'ms');
console.log('사용 모델:', response.model);
};
return (
<div>
{isLoading && <div>AI가 응답 중...</div>}
{error && <div className="error">오류: {error}</div>}
{lastResponse?.success && (
<div>{lastResponse.content}</div>
)}
</div>
);
}
*/
// ===== 비용 계산 유틸리티 =====
export const MODEL_PRICING = {
'gpt-4.1': { input: 8.0, output: 8.0 }, // $/1M tokens
'claude-sonnet-4-20250514': { input: 4.5, output: 4.5 },
'gemini-2.0-flash': { input: 2.5, output: 2.5 },
'deepseek-chat': { input: 0.42, output: 0.42 }, // 가장 저렴
};
export function calculateCost(model: string, inputTokens: number, outputTokens: number): number {
const pricing = MODEL_PRICING[model] || MODEL_PRICING['gpt-4.1'];
const inputCost = (inputTokens / 1_000_000) * pricing.input;
const outputCost = (outputTokens / 1_000_000) * pricing.output;
return inputCost + outputCost;
}
export default useAIFallback;
폴백 메커니즘 아키텍처 비교
실제 프로덕션에서 사용할 수 있는 폴백 전략입니다:
"""
Production-grade Fallback Architecture
HolySheep AI를 중심으로 한 다중 계층 폴백 구조
"""
from typing import Callable, Any, Optional
from enum import Enum
import asyncio
from dataclasses import dataclass
class FallbackStrategy(Enum):
"""폴백 전략 정의"""
CASCADE = "cascade" # 순차적 시도 (빠른 응답 우선)
PARALLEL = "parallel" # 병렬 시도 (최고 응답 우선)
COST_OPTIMIZED = "cost" # 비용 최적화 (저가 모델 우선)
LATENCY_OPTIMIZED = "latency" # 지연 최적화 (가장 빠른 모델 우선)
@dataclass
class FallbackChain:
"""폴백 체인 정의"""
name: str
models: list[str]
timeout: float
priority: int
def __lt__(self, other):
return self.priority < other.priority
class ProductionFallbackManager:
"""
HolySheep AI 기반 프로덕션 폴백 관리자
HolySheep AI 가격 비교:
┌─────────────────────┬───────────────┐
│ 모델 │ 가격 ($/1M tok)│
├─────────────────────┼───────────────┤
│ GPT-4.1 │ $8.00 │
│ Claude Sonnet 4 │ $4.50 │
│ Gemini 2.5 Flash │ $2.50 │
│ DeepSeek V3 │ $0.42 ←최저가 │
└─────────────────────┴───────────────┘
"""
def __init__(self, api_key: str, strategy: FallbackStrategy = FallbackStrategy.CASCADE):
self.api_key = api_key
self.strategy = strategy
self.base_url = "https://api.holysheep.ai/v1" # HolySheep
# HolySheep AI에서 지원하는 모델 기반 폴백 체인
self.chains: list[FallbackChain] = [
# 전략 1: 품질 우선 (Cascade)
FallbackChain(
name="quality_first",
models=["gpt-4.1", "claude-sonnet-4-20250514", "gemini-2.0-flash", "deepseek-chat"],
timeout=20.0,
priority=1
),
# 전략 2: 비용 최적화
FallbackChain(
name="cost_optimized",
models=["deepseek-chat", "gemini-2.0-flash", "claude-sonnet-4-20250514", "gpt-4.1"],
timeout=25.0,
priority=2
),
# 전략 3: 속도 우선
FallbackChain(
name="speed_first",
models=["gemini-2.0-flash", "deepseek-chat", "claude-sonnet-4-20250514", "gpt-4.1"],
timeout=15.0,
priority=3
),
]
# 선택된 체인
self.active_chain = self._select_chain()
def _select_chain(self) -> FallbackChain:
"""전략에 따른 체인 선택"""
if self.strategy == FallbackStrategy.COST_OPTIMIZED:
return next(c for c in self.chains if c.name == "cost_optimized")
elif self.strategy == FallbackStrategy.LATENCY_OPTIMIZED:
return next(c for c in self.chains if c.name == "speed_first")
return next(c for c in self.chains if c.name == "quality_first")
async def execute_with_fallback(
self,
prompt: str,
custom_chain: Optional[list[str]] = None,
callback: Optional[Callable[[str, bool, float], None]] = None
) -> dict[str, Any]:
"""
폴백과 함께 요청 실행
Args:
prompt: 사용자 입력
custom_chain: 사용자 정의 모델 체인
callback: 각 시도별 결과 콜백
Returns:
{"content": str, "model": str, "attempts": int, "total_latency_ms": float}
"""
models = custom_chain or self.active_chain.models
total_start = asyncio.get_event_loop().time()
attempts = 0
last_error = None
for model in models:
attempts += 1
model_start = asyncio.get_event_loop().time()
try:
result = await self._call_holySheep(model, prompt)
latency = (asyncio.get_event_loop().time() - model_start) * 1000
callback and callback(model, True, latency)
return {
"content": result["content"],
"model": model,
"provider": "HolySheep AI",
"attempts": attempts,
"total_latency_ms": round(latency, 2),
"success": True
}
except Exception as e:
last_error = str(e)
latency = (asyncio.get_event_loop().time() - model_start) * 1000
callback and callback(model, False, latency)
# 즉시 폴백 (RateLimit, Timeout만)
if "429" in last_error or "timeout" in last_error.lower():
continue
# API 에러는 잠시 대기 후 재시도
await asyncio.sleep(0.5)
continue
# 모든 폴백 실패
return {
"content": "서비스 일시적으로 이용 불가",
"model": "none",
"provider": "fallback",
"attempts": attempts,
"total_latency_ms": 0,
"success": False,
"error": last_error
}
async def _call_holySheep(self, model: str, prompt: str) -> dict:
"""HolySheep AI API 호출"""
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self