10장: API 게이트웨이와 프로덕션 인프라

model_list:
  - model_name: "chat-default"
    litellm_params:
      model: "anthropic/claude-4"
      api_key: "sk-ant-..."
    model_info:
      max_tokens: 8192
      input_cost_per_token: 0.000003
      output_cost_per_token: 0.000015
 
  - model_name: "chat-default"
    litellm_params:
      model: "openai/gpt-4o"
      api_key: "sk-..."
    model_info:
      max_tokens: 4096
      input_cost_per_token: 0.000005
      output_cost_per_token: 0.000015
 
  - model_name: "chat-fast"
    litellm_params:
      model: "anthropic/claude-4-haiku"
      api_key: "sk-ant-..."
 
  - model_name: "embedding"
    litellm_params:
      model: "openai/text-embedding-3-large"
      api_key: "sk-..."
 
  - model_name: "coding"
    litellm_params:
      model: "anthropic/claude-4"
      api_key: "sk-ant-..."
 
router_settings:
  routing_strategy: "latency-based-routing"
  num_retries: 3
  retry_after: 5
  allowed_fails: 2
  cooldown_time: 60
 
litellm_settings:
  drop_params: true
  set_verbose: false
  cache: true
  cache_params:
    type: "redis"
    host: "redis"
    port: 6379

import litellm
 
# OpenAI 호환 인터페이스로 어떤 모델이든 호출
response = await litellm.acompletion(
    model="anthropic/claude-4",
    messages=[{"role": "user", "content": "Hello"}],
    max_tokens=1024,
)
 
# 프록시 서버를 통한 호출 (어떤 SDK로든)
import openai
 
client = openai.OpenAI(
    api_key="sk-internal-key",
    base_url="http://litellm-proxy:4000",
)
 
response = client.chat.completions.create(
    model="chat-default",  # 라우팅 규칙에 따라 프로바이더 결정
    messages=[{"role": "user", "content": "Hello"}],
)

server:
  port: 8080
  read_timeout: 120s
  write_timeout: 120s
 
providers:
  anthropic:
    api_key: "${ANTHROPIC_API_KEY}"
    models:
      - claude-4
      - claude-4-haiku
  openai:
    api_key: "${OPENAI_API_KEY}"
    models:
      - gpt-4o
      - gpt-4o-mini
 
routing:
  default_strategy: "round-robin"
  rules:
    - match:
        model: "chat-*"
      backends:
        - provider: anthropic
          model: claude-4
          weight: 70
        - provider: openai
          model: gpt-4o
          weight: 30
    - match:
        model: "fast-*"
      backends:
        - provider: anthropic
          model: claude-4-haiku
 
fallback:
  enabled: true
  max_retries: 2
  retry_on:
    - 429  # Rate limit
    - 500  # Server error
    - 503  # Service unavailable

from enum import Enum
from dataclasses import dataclass
import random
 
 
class RoutingStrategy(str, Enum):
    round_robin = "round-robin"
    weighted = "weighted"
    latency_based = "latency-based"
    cost_based = "cost-based"
    capability_based = "capability-based"
 
 
@dataclass
class ProviderBackend:
    provider: str
    model: str
    weight: int = 1
    avg_latency_ms: float = 0
    cost_per_token: float = 0
    capabilities: set[str] = None
    healthy: bool = True
 
 
class SmartRouter:
    """지능형 멀티 프로바이더 라우터"""
    
    def __init__(self, backends: list[ProviderBackend]):
        self.backends = backends
    
    def route(
        self,
        strategy: RoutingStrategy,
        request: dict,
    ) -> ProviderBackend:
        healthy = [b for b in self.backends if b.healthy]
        if not healthy:
            raise NoHealthyBackendError("사용 가능한 백엔드가 없습니다")
        
        if strategy == RoutingStrategy.weighted:
            return self._weighted_route(healthy)
        elif strategy == RoutingStrategy.latency_based:
            return self._latency_route(healthy)
        elif strategy == RoutingStrategy.cost_based:
            return self._cost_route(healthy, request)
        elif strategy == RoutingStrategy.capability_based:
            return self._capability_route(healthy, request)
        else:
            return self._round_robin(healthy)
    
    def _weighted_route(
        self, backends: list[ProviderBackend]
    ) -> ProviderBackend:
        weights = [b.weight for b in backends]
        return random.choices(backends, weights=weights, k=1)[0]
    
    def _latency_route(
        self, backends: list[ProviderBackend]
    ) -> ProviderBackend:
        return min(backends, key=lambda b: b.avg_latency_ms)
    
    def _cost_route(
        self,
        backends: list[ProviderBackend],
        request: dict,
    ) -> ProviderBackend:
        estimated_tokens = estimate_tokens(request)
        return min(
            backends,
            key=lambda b: b.cost_per_token * estimated_tokens,
        )
    
    def _capability_route(
        self,
        backends: list[ProviderBackend],
        request: dict,
    ) -> ProviderBackend:
        required = set()
        if has_images(request):
            required.add("vision")
        if has_tools(request):
            required.add("tool_calling")
        if request.get("response_format"):
            required.add("structured_output")
        
        capable = [
            b for b in backends
            if required.issubset(b.capabilities or set())
        ]
        if not capable:
            raise NoCapableBackendError(
                f"필요한 기능을 지원하는 백엔드가 없습니다: {required}"
            )
        
        return self._latency_route(capable)

class FallbackHandler:
    """프로바이더 장애 시 자동 폴백"""
    
    def __init__(
        self,
        primary: ProviderBackend,
        fallbacks: list[ProviderBackend],
        max_retries: int = 2,
    ):
        self.chain = [primary] + fallbacks
        self.max_retries = max_retries
    
    async def execute(self, request: dict) -> dict:
        last_error = None
        
        for backend in self.chain:
            if not backend.healthy:
                continue
            
            for attempt in range(self.max_retries + 1):
                try:
                    response = await call_provider(
                        backend, request
                    )
                    return response
                except RateLimitError:
                    # 다음 프로바이더로 즉시 폴백
                    logger.warning(
                        f"{backend.provider} 레이트 리밋, "
                        f"폴백 시도"
                    )
                    break
                except ProviderError as e:
                    last_error = e
                    if attempt < self.max_retries:
                        await asyncio.sleep(
                            0.5 * (2 ** attempt)
                        )
                    else:
                        # 백엔드를 비정상으로 표시
                        backend.healthy = False
                        schedule_health_check(backend)
        
        raise AllProvidersFailedError(
            f"모든 프로바이더 실패: {last_error}"
        )

from fastapi import Security, HTTPException
from fastapi.security import APIKeyHeader
 
api_key_header = APIKeyHeader(name="Authorization", auto_error=False)
 
 
async def verify_api_key(
    authorization: str | None = Security(api_key_header),
) -> APIKeyInfo:
    if not authorization:
        raise HTTPException(status_code=401, detail="API 키가 필요합니다")
    
    # "Bearer sk-..." 형식에서 토큰 추출
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="잘못된 인증 형식입니다")
    
    api_key = authorization[7:]
    
    # API 키 검증 (해시 비교)
    key_info = await key_store.verify(api_key)
    if not key_info:
        raise HTTPException(status_code=401, detail="유효하지 않은 API 키입니다")
    
    if key_info.revoked:
        raise HTTPException(status_code=401, detail="폐기된 API 키입니다")
    
    if key_info.expires_at and key_info.expires_at < datetime.now():
        raise HTTPException(status_code=401, detail="만료된 API 키입니다")
    
    return key_info

import jwt
from datetime import datetime, timedelta
 
 
class JWTAuth:
    def __init__(self, secret: str, algorithm: str = "HS256"):
        self.secret = secret
        self.algorithm = algorithm
    
    def create_token(
        self,
        user_id: str,
        org_id: str,
        scopes: list[str],
        expires_in: timedelta = timedelta(hours=1),
    ) -> str:
        payload = {
            "sub": user_id,
            "org": org_id,
            "scopes": scopes,
            "iat": datetime.utcnow(),
            "exp": datetime.utcnow() + expires_in,
        }
        return jwt.encode(payload, self.secret, algorithm=self.algorithm)
    
    def verify_token(self, token: str) -> dict:
        try:
            payload = jwt.decode(
                token, self.secret, algorithms=[self.algorithm]
            )
            return payload
        except jwt.ExpiredSignatureError:
            raise HTTPException(status_code=401, detail="토큰이 만료되었습니다")
        except jwt.InvalidTokenError:
            raise HTTPException(status_code=401, detail="유효하지 않은 토큰입니다")
 
 
# 스코프 기반 인가
def require_scopes(*required_scopes: str):
    async def dependency(token_data: dict = Depends(verify_jwt)):
        user_scopes = set(token_data.get("scopes", []))
        if not set(required_scopes).issubset(user_scopes):
            raise HTTPException(
                status_code=403,
                detail=f"필요한 권한: {required_scopes}",
            )
        return token_data
    return dependency
 
 
# 엔드포인트에 스코프 적용
@app.post("/v1/chat/completions")
async def create_completion(
    request: CompletionRequest,
    user: dict = Depends(require_scopes("chat:write")),
):
    pass
 
 
@app.get("/v1/usage")
async def get_usage(
    user: dict = Depends(require_scopes("usage:read")),
):
    pass

import hashlib
import json
 
 
class AIResponseCache:
    """AI 응답 캐싱"""
    
    def __init__(self, redis_client, ttl: int = 3600):
        self.redis = redis_client
        self.ttl = ttl
    
    def _cache_key(self, request: dict) -> str:
        """캐시 키 생성 — 결정적 파라미터만 사용"""
        cacheable = {
            "model": request["model"],
            "messages": request["messages"],
            "temperature": request.get("temperature", 1.0),
            "max_tokens": request.get("max_tokens"),
            "seed": request.get("seed"),
            "tools": request.get("tools"),
            "response_format": request.get("response_format"),
        }
        
        # temperature=0이고 seed가 고정이면 결정적
        content = json.dumps(cacheable, sort_keys=True, ensure_ascii=False)
        return f"ai_cache:{hashlib.sha256(content.encode()).hexdigest()}"
    
    def is_cacheable(self, request: dict) -> bool:
        """캐싱 가능한 요청인지 판별"""
        # temperature=0 또는 seed가 지정된 경우만 캐싱
        temp = request.get("temperature", 1.0)
        seed = request.get("seed")
        stream = request.get("stream", False)
        
        return (temp == 0 or seed is not None) and not stream
    
    async def get(self, request: dict) -> dict | None:
        if not self.is_cacheable(request):
            return None
        
        key = self._cache_key(request)
        cached = await self.redis.get(key)
        
        if cached:
            return json.loads(cached)
        return None
    
    async def set(self, request: dict, response: dict) -> None:
        if not self.is_cacheable(request):
            return
        
        key = self._cache_key(request)
        await self.redis.setex(
            key,
            self.ttl,
            json.dumps(response, ensure_ascii=False),
        )

import structlog
 
logger = structlog.get_logger()
 
 
async def log_completion(
    request_id: str,
    model: str,
    user_id: str,
    prompt_tokens: int,
    completion_tokens: int,
    latency_ms: float,
    status: str,
    error: str | None = None,
):
    logger.info(
        "completion",
        request_id=request_id,
        model=model,
        user_id=user_id,
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        total_tokens=prompt_tokens + completion_tokens,
        latency_ms=round(latency_ms, 1),
        status=status,
        error=error,
        cost_usd=calculate_cost(model, prompt_tokens, completion_tokens),
    )

from prometheus_client import (
    Counter, Histogram, Gauge, Summary,
)
 
# 요청 카운터
request_total = Counter(
    "ai_api_requests_total",
    "Total AI API requests",
    ["model", "status", "provider"],
)
 
# 토큰 사용량
tokens_total = Counter(
    "ai_api_tokens_total",
    "Total tokens processed",
    ["model", "type"],  # type: prompt/completion
)
 
# 지연시간 분포
latency_histogram = Histogram(
    "ai_api_latency_seconds",
    "Request latency",
    ["model", "stream"],
    buckets=[0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60],
)
 
# TTFT (Time To First Token)
ttft_histogram = Histogram(
    "ai_api_ttft_seconds",
    "Time to first token for streaming",
    ["model"],
    buckets=[0.05, 0.1, 0.25, 0.5, 1, 2, 5],
)
 
# 비용
cost_counter = Counter(
    "ai_api_cost_usd_total",
    "Total cost in USD",
    ["model", "provider", "user_tier"],
)
 
# 캐시 적중률
cache_hits = Counter(
    "ai_api_cache_hits_total",
    "Cache hit count",
    ["model"],
)
cache_misses = Counter(
    "ai_api_cache_misses_total",
    "Cache miss count",
    ["model"],
)