저는 3년째 AI API 게이트웨이 개발을 맡고 있는 엔지니어입니다. 이번 글에서는 AI API를 안정적으로 배포하고 운영하는 데 필요한 핵심 전략을 실제 경험 바탕으로 정리하겠습니다. 특히 HolySheep AI를 활용한 비용 최적화 및 고가용성 아키텍처에 초점을 맞추겠습니다.
시작하기 전에: 왜 AI API 배포 전략이 중요한가
최근 이커머스 플랫폼에서 AI 고객 서비스 봇을 출시한 경험이 있습니다. 초기에는 단일 모델 호출로 간단히 구현했지만, 트래픽이 10배 증가하면서 세 가지 심각한 문제에 직면했습니다:
- 예측 불가능한 비용 폭증
- 특정 모델의 지연 시간 증가
- 일부 모델 서비스 중단 시 전체 시스템 영향
이 문제를 해결하기 위해 다중 모델 라우팅과 자동 장애 조치 아키텍처를 도입했습니다. 결과적으로:
- 응답 시간: 평균 1,200ms → 450ms 개선
- API 비용: 월 $3,200 → $1,850 절감 (42% 감소)
- 가용성: 99.2% → 99.95% 향상
1. 다중 모델 라우팅 아키텍처
AI API 배포의 핵심은 단일 모델 의존도를 낮추고, 요청 유형에 따라 최적 모델을 선택하는 것입니다. HolySheep AI를 사용하면 단일 API 키로 다양한 모델에 접근할 수 있어 이架构을 쉽게 구현할 수 있습니다.
1.1 기본 라우팅 시스템 구현
#!/usr/bin/env python3
"""
AI API Multi-Model Router
HolySheep AI 게이트웨이 기반 요청 라우팅
"""
import httpx
import asyncio
import time
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
class TaskType(Enum):
COMPLEX_REASONING = "complex"
FAST_RESPONSE = "fast"
COST_SENSITIVE = "budget"
@dataclass
class ModelConfig:
name: str
provider: str
cost_per_1m_tokens: float
avg_latency_ms: float
max_tokens: int
capabilities: list
class AIRouter:
"""HolySheep AI 기반 다중 모델 라우터"""
BASE_URL = "https://api.holysheep.ai/v1"
# HolySheep AI 지원 모델 설정
MODELS = {
"gpt-4.1": ModelConfig(
name="gpt-4.1",
provider="openai",
cost_per_1m_tokens=8.00, # $8/MTok
avg_latency_ms=850,
max_tokens=128000,
capabilities=["reasoning", "coding", "analysis"]
),
"claude-sonnet-4.5": ModelConfig(
name="claude-sonnet-4.5",
provider="anthropic",
cost_per_1m_tokens=15.00, # $15/MTok
avg_latency_ms=920,
max_tokens=200000,
capabilities=["long-context", "writing", "analysis"]
),
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
provider="google",
cost_per_1m_tokens=2.50, # $2.50/MTok
avg_latency_ms=380,
max_tokens=1000000,
capabilities=["fast", "multimodal", "batch"]
),
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
provider="deepseek",
cost_per_1m_tokens=0.42, # $0.42/MTok
avg_latency_ms=520,
max_tokens=64000,
capabilities=["coding", "math", "cost-effective"]
)
}
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.AsyncClient(timeout=30.0)
self.request_count = {"success": 0, "fallback": 0, "error": 0}
self.cost_tracker = {"total_tokens": 0, "total_cost": 0.0}
async def route_request(
self,
task_type: TaskType,
prompt: str,
fallback_enabled: bool = True
) -> Dict[str, Any]:
"""요청 유형에 따라 최적 모델 선택 및 호출"""
# 1. 작업 유형별 모델 선택
selected_model = self._select_model(task_type, prompt)
# 2. 기본 모델로 요청
result = await self._call_model(selected_model, prompt)
if result["success"] or not fallback_enabled:
return result
# 3. 폴백: 다른 모델로 재시도
self.request_count["fallback"] += 1
fallback_model = self._select_fallback(selected_model, task_type)
return await self._call_model(fallback_model, prompt)
def _select_model(self, task_type: TaskType, prompt: str) -> str:
"""작업 유형별 최적 모델 선택 로직"""
if task_type == TaskType.COMPLEX_REASONING:
# 복잡한 추론은 GPT-4.1 또는 Claude Sonnet
return "gpt-4.1"
elif task_type == TaskType.FAST_RESPONSE:
# 빠른 응답은 Gemini Flash
return "gemini-2.5-flash"
elif task_type == TaskType.COST_SENSITIVE:
# 비용 최적화는 DeepSeek
return "deepseek-v3.2"
# 기본: 복잡도 추정
prompt_length = len(prompt)
if prompt_length > 10000:
return "claude-sonnet-4.5"
elif prompt_length > 3000:
return "gemini-2.5-flash"
else:
return "deepseek-v3.2"
def _select_fallback(self, primary: str, task_type: TaskType) -> str:
"""폴백 모델 선택"""
fallbacks = {
"gpt-4.1": ["claude-sonnet-4.5", "gemini-2.5-flash"],
"claude-sonnet-4.5": ["gpt-4.1", "gemini-2.5-flash"],
"gemini-2.5-flash": ["deepseek-v3.2", "gpt-4.1"],
"deepseek-v3.2": ["gemini-2.5-flash", "gpt-4.1"]
}
return fallbacks.get(primary, ["gemini-2.5-flash"])[0]
async def _call_model(self, model: str, prompt: str) -> Dict[str, Any]:
"""HolySheep AI API 호출"""
start_time = time.time()
try:
response = await self.client.post(
f"{self.BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 2048
}
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
data = response.json()
usage = data.get("usage", {})
tokens_used = usage.get("total_tokens", 0)
# 비용 계산
model_config = self.MODELS.get(model)
cost = (tokens_used / 1_000_000) * model_config.cost_per_1m_tokens
self.request_count["success"] += 1
self.cost_tracker["total_tokens"] += tokens_used
self.cost_tracker["total_cost"] += cost
return {
"success": True,
"model": model,
"response": data["choices"][0]["message"]["content"],
"latency_ms": round(latency_ms, 2),
"tokens": tokens_used,
"cost_usd": round(cost, 4)
}
else:
return {
"success": False,
"error": f"API Error: {response.status_code}",
"model": model
}
except Exception as e:
return {
"success": False,
"error": str(e),
"model": model
}
def get_stats(self) -> Dict[str, Any]:
"""라우팅 통계 반환"""
return {
**self.request_count,
**self.cost_tracker,
"success_rate": (
self.request_count["success"] /
max(sum(self.request_count.values()), 1) * 100
)
}
사용 예제
async def main():
router = AIRouter(api_key="YOUR_HOLYSHEEP_API_KEY")
# 다양한 요청 유형 테스트
test_cases = [
(TaskType.COMPLEX_REASONING, "왜 하늘은 파란색인가에 대한 과학적 설명을 작성해줘"),
(TaskType.FAST_RESPONSE, "안녕하세요 간단히 인사해줘"),
(TaskType.COST_SENSITIVE, "1+1은 몇인가요"),
]
for task_type, prompt in test_cases:
result = await router.route_request(task_type, prompt)
print(f"Task: {task_type.value} | Model: {result.get('model')} | "
f"Latency: {result.get('latency_ms')}ms | Cost: ${result.get('cost_usd')}")
print(f"\nTotal Stats: {router.get_stats()}")
if __name__ == "__main__":
asyncio.run(main())2. 비용 최적화 전략
AI API 운영에서 비용 관리는 매우 중요합니다. HolySheep AI의 모델별 가격표를 활용하면 요청 특성에 따라 연간 수천 달러를 절감할 수 있습니다.
2.1 스마트 캐싱 시스템
#!/usr/bin/env python3
"""
AI API 비용 최적화: 스마트 캐싱 및 토큰 관리
"""
import hashlib
import json
import time
from typing import Optional, Dict, Any, Tuple
from collections import OrderedDict
import asyncio
class SmartCache:
"""
LRU 캐시 + 임베딩 유사도 기반 스마트 캐싱
비용 절감 및 응답 시간 단축
"""
def __init__(self, max_size: int = 1000, ttl_seconds: int = 3600):
self.cache = OrderedDict()
self.max_size = max_size
self.ttl = ttl_seconds
self.hits = 0
self.misses = 0
self.cost_saved = 0.0
def _generate_key(self, prompt: str, model: str) -> str:
"""프로프트 해시 키 생성"""
content = f"{model}:{prompt.lower().strip()}"
return hashlib.sha256(content.encode()).hexdigest()[:32]
def get(self, prompt: str, model: str) -> Optional[Dict[str, Any]]:
"""캐시 조회"""
key = self._generate_key(prompt, model)
if key in self.cache:
entry = self.cache[key]
# TTL 체크
if time.time() - entry["timestamp"] < self.ttl:
self.cache.move_to_end(key)
self.hits += 1
# 예상 비용 절감량 (평균 500 토큰 가정)
avg_tokens = 500
avg_cost_per_token = 0.00001 # 평균 비용
self.cost_saved += (avg_tokens / 1_000_000) * 15.0 # 평균 모델 비용
return entry["response"]
else:
# TTL 만료 - 제거
del self.cache[key]
self.misses += 1
return None
def set(self, prompt: str, model: str, response: Dict[str, Any]):
"""캐시 저장"""
key = self._generate_key(prompt, model)
# LRU 정책: 최대 크기 초과 시 가장 오래된 항목 제거
if len(self.cache) >= self.max_size:
self.cache.popitem(last=False)
self.cache[key] = {
"response": response,
"timestamp": time.time()
}
self.cache.move_to_end(key)
def get_stats(self) -> Dict[str, Any]:
"""캐시 통계"""
total = self.hits + self.misses
hit_rate = (self.hits / total * 100) if total > 0 else 0
return {
"hits": self.hits,
"misses": self.misses,
"hit_rate_percent": round(hit_rate, 2),
"cost_saved_usd": round(self.cost_saved, 2),
"cache_size": len(self.cache)
}
class TokenOptimizer:
"""
토큰 사용량 최적화
프로프트 압축 및 컨텍스트 관리
"""
# 불필요한 토큰 패턴
REMOVE_PATTERNS = [
r"당신은\s+",
r"너는\s+",
r"，请你",
r"请帮我",
r"plz\s+",
r"plz\s+",
r"plz",
]
@staticmethod
def estimate_tokens(text: str) -> int:
"""대략적인 토큰 수 추정 (한국어: 1토큰 ≈ 2글자)"""
# 실제는 tiktoken 권장
return len(text) // 2
@staticmethod
def compress_prompt(prompt: str) -> str:
"""프로프트 압축 - 불필요한 표현 제거"""
import re
compressed = prompt
for pattern in TokenOptimizer.REMOVE_PATTERNS:
compressed = re.sub(pattern, "", compressed, flags=re.IGNORECASE)
# 연속 공백 제거
compressed = re.sub(r'\s+', ' ', compressed).strip()
return compressed
@staticmethod
def truncate_for_context(
prompt: str,
max_tokens: int,
include_system: str = ""
) -> Tuple[str, str]:
"""
컨텍스트 창 관리를 위한 프로프트 자르기
반환: (system_prompt, user_prompt)
"""
MAX_CONTEXT = max_tokens - 500 # 응답 공간 확보
# 시스템 프롬프트 고정
system_part = include_system[:1000] if include_system else ""
# 사용 가능 공간 계산
system_tokens = TokenOptimizer.estimate_tokens(system_part)
available_tokens = MAX_CONTEXT - system_tokens
if TokenOptimizer.estimate_tokens(prompt) <= available_tokens:
return system_part, prompt
# 프로프트 자르기 (뒤쪽 우선)
truncated_prompt = prompt[-available_tokens * 2:]
return system_part, truncated_prompt
비용 절감 시뮬레이션
def simulate_cost_savings():
"""월간 비용 절감 시뮬레이션"""
# 월간 요청 예상
monthly_requests = 50000
# 캐시 히트율별 비용 비교
scenarios = [
("캐시 없음", 0),
("저조한 캐시", 15),
("적절한 캐시", 35),
("스마트 캐시", 55),
]
# 평균 요청 비용 ($0.001 per request with average model)
avg_request_cost = 0.001
print("=" * 60)
print("월간 AI API 비용 절감 시뮬레이션")
print("=" * 60)
print(f"월간 요청 수: {monthly_requests:,} requests")
print(f"평균 요청 비용: ${avg_request_cost:.4f}")
print()
for name, hit_rate in scenarios:
original_cost = monthly_requests * avg_request_cost
cache_benefit = original_cost * (hit_rate / 100)
final_cost = original_cost - cache_benefit
print(f"시나리오: {name}")
print(f" 캐시 히트율: {hit_rate}%")
print(f" 원래 비용: ${original_cost:.2f}")
print(f" 절감액: ${cache_benefit:.2f}")
print(f" 최종 비용: ${final_cost:.2f}")
print()
if __name__ == "__main__":
# 캐시 테스트
cache = SmartCache(max_size=100, ttl_seconds=300)
# 테스트 캐시 조회
test_prompt = "한국의 수도는 어디인가요?"
cached_result = cache.get(test_prompt, "gpt-4.1")
if cached_result:
print(f"캐시 히트: {cached_result}")
else:
print("캐시 미스 - API 호출 필요")
# 실제 API 호출 후 캐시 저장
cache.set(test_prompt, "gpt-4.1", {"answer": "서울"})
print(f"캐시 통계: {cache.get_stats()}")
# 비용 절감 시뮬레이션
simulate_cost_savings()3. 모니터링 및 알림 시스템
안정적인 AI API 운영을 위해서는 실시간 모니터링과 비용 알림이 필수입니다. HolySheep AI 대시보드와 연동하는 모니터링 시스템을 구축해보겠습니다.
#!/usr/bin/env python3
"""
AI API 모니터링 및 알림 시스템
"""
import time
import asyncio
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from collections import defaultdict
import smtplib
from email.mime.text import MIMEText
@dataclass
class RequestMetric:
"""개별 요청 메트릭"""
timestamp: float
model: str
latency_ms: float
tokens: int
cost_usd: float
success: bool
error_type: Optional[str] = None
@dataclass
class DailyReport:
"""일일 보고서"""
date: str
total_requests: int
successful_requests: int
failed_requests: int
total_tokens: int
total_cost_usd: float
avg_latency_ms: float
model_usage: Dict[str, int] = field(default_factory=dict)
errors_by_type: Dict[str, int] = field(default_factory=dict)
class AIMonitor:
"""
AI API 모니터링 시스템
- 실시간 메트릭 수집
- 비용 추적
- 알림 발송
"""
def __init__(self, warning_threshold_daily: float = 50.0):
self.metrics: List[RequestMetric] = []
self.warning_threshold_daily = warning_threshold_daily
self.cost_alerts: List[Dict] = []
# 시간대별 요청 분포
self.hourly_distribution = defaultdict(int)
# 모델별 응답 시간 추적
self.model_latencies: Dict[str, List[float]] = defaultdict(list)
def record_request(self, metric: RequestMetric):
"""요청 메트릭 기록"""
self.metrics.append(metric)
# 시간대별 분포
hour = datetime.fromtimestamp(metric.timestamp).hour
self.hourly_distribution[hour] += 1
# 모델별 지연 시간
if metric.success:
self.model_latencies[metric.model].append(metric.latency_ms)
# 비용 임계값 체크
daily_cost = self.get_daily_cost()
if daily_cost >= self.warning_threshold_daily:
self.cost_alerts.append({
"timestamp": datetime.now().isoformat(),
"daily_cost": daily_cost,
"threshold": self.warning_threshold_daily
})
def get_daily_cost(self) -> float:
"""오늘 비용 합계 계산"""
today_start = datetime.now().replace(
hour=0, minute=0, second=0, microsecond=0
).timestamp()
return sum(
m.cost_usd for m in self.metrics
if m.timestamp >= today_start
)
def get_current_costs(self) -> Dict[str, float]:
"""모델별 현재 비용"""
costs = defaultdict(float)
today_start = datetime.now().replace(
hour=0, minute=0, second=0, microsecond=0
).timestamp()
for m in self.metrics:
if m.timestamp >= today_start:
costs[m.model] += m.cost_usd
return dict(costs)
def generate_daily_report(self) -> DailyReport:
"""일일 보고서 생성"""
today_start = datetime.now().replace(
hour=0, minute=0, second=0, microsecond=0
).timestamp()
today_metrics = [m for m in self.metrics if m.timestamp >= today_start]
if not today_metrics:
return DailyReport(
date=datetime.now().strftime("%Y-%m-%d"),
total_requests=0,
successful_requests=0,
failed_requests=0,
total_tokens=0,
total_cost_usd=0.0,
avg_latency_ms=0.0
)
successful = [m for m in today_metrics if m.success]
failed = [m for m in today_metrics if not m.success]
# 모델 사용량
model_usage = defaultdict(int)
for m in today_metrics:
model_usage[m.model] += 1
# 에러 유형
errors_by_type = defaultdict(int)
for m in failed:
errors_by_type[m.error_type or "unknown"] += 1
return DailyReport(
date=datetime.now().strftime("%Y-%m-%d"),
total_requests=len(today_metrics),
successful_requests=len(successful),
failed_requests=len(failed),
total_tokens=sum(m.tokens for m in today_metrics),
total_cost_usd=sum(m.cost_usd for m in today_metrics),
avg_latency_ms=sum(m.latency_ms for m in successful) / len(successful) if successful else 0,
model_usage=dict(model_usage),
errors_by_type=dict(errors_by_type)
)
def get_model_performance(self) -> Dict[str, Dict]:
"""모델별 성능 지표"""
performance = {}
for model, latencies in self.model_latencies.items():
if latencies:
performance[model] = {
"avg_latency_ms": round(sum(latencies) / len(latencies), 2),
"min_latency_ms": round(min(latencies), 2),
"max_latency_ms": round(max(latencies), 2),
"p95_latency_ms": round(sorted(latencies)[int(len(latencies) * 0.95)], 2),
"request_count": len(latencies)
}
return performance
def check_anomalies(self) -> List[Dict]:
"""이상 상황 감지"""
anomalies = []
today_start = datetime.now().replace(
hour=0, minute=0, second=0, microsecond=0
).timestamp()
today_metrics = [m for m in self.metrics if m.timestamp >= today_start]
if not today_metrics:
return anomalies
# 실패율 체크 (5% 이상)
failed = [m for m in today_metrics if not m.success]
failure_rate = len(failed) / len(today_metrics)
if failure_rate > 0.05:
anomalies.append({
"type": "high_failure_rate",
"message": f"실패율 {failure_rate * 100:.1f}% (임계값: 5%)",
"severity": "warning"
})
# 평균 지연 시간 체크 (2초 이상)
successful = [m for m in today_metrics if m.success]
if successful:
avg_latency = sum(m.latency_ms for m in successful) / len(successful)
if avg_latency > 2000:
anomalies.append({
"type": "high_latency",
"message": f"평균 지연 {avg_latency:.0f}ms (임계값: 2000ms)",
"severity": "warning"
})
# 비용 급증 체크
if self.get_daily_cost() > self.warning_threshold_daily:
anomalies.append({
"type": "cost_threshold_exceeded",
"message": f"일일 비용 ${self.get_daily_cost():.2f} 임계값 초과",
"severity": "critical"
})
return anomalies
def print_dashboard(self):
"""모니터링 대시보드 출력"""
report = self.generate_daily_report()
performance = self.get_model_performance()
anomalies = self.check_anomalies()
print("=" * 70)
print(f"📊 AI API 모니터링 대시보드 - {report.date}")
print("=" * 70)
print(f"\n💰 비용 현황")
print(f" 일일 비용: ${report.total_cost_usd:.4f}")
print(f" 예상 월간 비용: ${report.total_cost_usd * 30:.2f}")
print(f"\n📈 요청 현황")
print(f" 총 요청: {report.total_requests:,}")
print(f" 성공: {report.successful_requests:,} ({report.successful_requests/max(report.total_requests,1)*100:.1f}%)")
print(f" 실패: {report.failed_requests:,}")
print(f" 토큰 사용: {report.total_tokens:,}")
print(f"\n⚡ 모델별 성능")
for model, perf in performance.items():
print(f" {model}:")
print(f" 평균 지연: {perf['avg_latency_ms']}ms")
print(f" P95 지연: {perf['p95_latency_ms']}ms")
print(f" 요청 수: {perf['request_count']:,}")
if anomalies:
print(f"\n🚨 이상 상황 ({len(anomalies)}건)")
for a in anomalies:
print(f" [{a['severity'].upper()}] {a['message']}")
print("=" * 70)
모니터링 시스템 데모
if __name__ == "__main__":
monitor = AIMonitor(warning_threshold_daily=100.0)
# 샘플 메트릭 추가
sample_metrics = [
RequestMetric(
timestamp=time.time() - 3600,
model="gpt-4.1",
latency_ms=820,
tokens=1500,
cost_usd=0.012,
success=True
),
RequestMetric(
timestamp=time.time() - 1800,
model="gemini-2.5-flash",
latency_ms=350,
tokens=800,
cost_usd=0.002,
success=True
),
RequestMetric(
timestamp=time.time() - 900,
model="deepseek-v3.2",
latency_ms=480,
tokens=1200,
cost_usd=0.0005,
success=True
),
RequestMetric(
timestamp=time.time() - 300,
model="claude-sonnet-4.5",
latency_ms=950,
tokens=2000,
cost_usd=0.030,
success=False,
error_type="rate_limit"
),
]
for metric in sample_metrics:
monitor.record_request(metric)
# 대시보드 출력
monitor.print_dashboard()
# 모델별 비용 분포
print(f"\n📊 모델별 비용 분포:")
for model, cost in monitor.get_current_costs().items():
print(f" {model}: ${cost:.4f}")4. HolySheep AI 실전 가격 비교
HolySheep AI를 통해 실제 비용을 비교해보면 모델 선택이 얼마나 중요한지 명확히 알 수 있습니다.
| 모델 | 입력 비용 ($/MTok) | 출력 비용 ($/MTok) | 평균 응답 시간 | 적합한 사용 사례 |
|---|---|---|---|---|
| GPT-4.1 | $8.00 | $8.00 | ~850ms | 복잡한 추론, 코딩, 분석 |
| Claude Sonnet 4.5 | $15.00 | $15.00 | ~920ms | 장문 분석, 창작, 긴 컨텍스트 |
| Gemini 2.5 Flash | $2.50 | $2.50 | ~380ms | 빠른 응답, 일괄 처리, multimodal |
| DeepSeek V3.2 | $0.42 | $0.42 | ~520ms | 비용 최적화, 코딩, 수학 |
비용 비교 시나리오:
- 100만 토큰 처리에 GPT-4.1 사용 시: $8.00
- 동일 작업을 DeepSeek V3.2 사용 시: $0.42 (95% 절감)
- Gemini 2.5 Flash 사용 시: $2.50 (69% 절감)
단순 질문에는 DeepSeek, 복잡한 분석에는 GPT-4.1, 빠른 응답에는 Gemini Flash를 사용하면 비용을 크게 절감할 수 있습니다.
자주 발생하는 오류와 해결책
오류 1: Rate Limit 초과 (429 Error)
# 문제: API 요청 시 429 Too Many Requests 오류
원인: HolySheep AI의 요청 빈도 제한 초과
import asyncio
import httpx
from datetime import datetime, timedelta
class RateLimitHandler:
"""Rate Limit 우회 및 재시도 로직"""
def __init__(self, max_retries: int = 3):
self.max_retries = max_retries
self.request_times = []
self.retry_after = 1 # 초기 대기 시간 (초)
async def call_with_retry(self, url: str, headers: dict, payload: dict) -> dict:
"""지수 백오프를 통한 재시도 로직"""
async with httpx.AsyncClient() as client:
for attempt in range(self.max_retries):
try:
response = await client.post(
url,
headers=headers,
json=payload,
timeout=30.0
)
if response.status_code == 200:
self.retry_after = 1 # 성공 시 초기화
return response.json()
elif response.status_code == 429:
# Rate Limit 초과 - Retry-After 헤더 확인
retry_after = response.headers.get("Retry-After")
if retry_after:
wait_time = int(retry_after)
else:
# 지수 백오프 적용
wait_time = self.retry_after * (2 ** attempt)
self.retry_after = min(self.retry_after * 2, 60)
print(f"Rate Limit 초과. {wait_time}초 후 재시도... (시도 {attempt + 1}/{self.max_retries})")
await asyncio.sleep(wait_time)
elif response.status_code == 500:
# 서버 오류 - 1초 후 재시도
await asyncio.sleep(1)
else:
return {"error": f"HTTP {response.status_code}", "details": response.text}
except httpx.TimeoutException:
print(f"요청 시간 초과. 재시도... (시도 {attempt + 1}/{self.max_retries})")
await asyncio.sleep(2)
except Exception as e:
print(f"예상치 못한 오류: {e}")
raise
return {"error": "최대 재시도 횟수 초과"}
Rate Limit 예방: 요청 간격 관리
class RequestThrottler:
"""요청 스로틀링 - Rate Limit 사전 예방"""
def __init__(self, requests_per_minute: int = 60):
self.rpm = requests_per_minute
self.min_interval = 60.0 / requests_per_minute
self.last_request_time = 0
async def wait_if_needed(self):
"""필요시 요청 전 대기"""
elapsed = datetime.now().timestamp() - self.last_request_time
if elapsed < self.min_interval:
wait_time = self.min_interval - elapsed
await asyncio.sleep(wait_time)
self.last_request_time = datetime.now().timestamp()오류 2: 토큰 초과 (400 Error - max_tokens)
# 문제: 요청 시 "max_tokens exceeded" 또는 컨텍스트 창 초과 오류
원인: 입력 토큰 + 출력 토큰이 모델 제한 초과
class TokenLimitHandler:
"""토큰 제한 관리 및 최적화"""
MODEL_LIMITS = {
"gpt-4.1": 128000,
"claude-sonnet-4.5": 200000,
"gemini-2.5-flash": 1000000,
"deepseek-v3.2": 64000
}
@staticmethod
def estimate_tokens(text: str) -> int:
"""토큰 수 추정 (한국어 최적화)"""
# 한국어: 대략 1토큰 ≈ 1.5글자
# 영어: 대략 1토큰 ≈ 4글자
korean_chars = sum(1 for c in text if '\uAC00' <= c <= '\uD7A3')
english_chars = len(text) - korean_chars
return int(korean_chars / 1.5) + int(english_chars / 4)
@staticmethod
def truncate_for_model(
prompt: str,
model: str,
reserved_output_tokens: int = 1000
) -> str:
"""모델 컨텍스트 창에 맞게 프로프트 자르기"""
max_tokens = TokenLimitHandler.MODEL_LIMITS.get(
model,
TokenLimitHandler.MODEL_L