LLM 라우팅·캐스케이드 전략: 멀티 모델 오케스트레이션으로 비용 최적화

# RouteLLM 서버 실행 및 클라이언트 사용 예시
# pip install routellm

# 1. 서버 실행 (OpenAI API 호환)
# python -m routellm.openai_server \
#   --routers mf \
#   --strong-model gpt-4o \
#   --weak-model gpt-4o-mini

# 2. 클라이언트에서 사용
import openai

client = openai.OpenAI(
    base_url="http://localhost:6060/v1",
    api_key="not-needed",  # RouteLLM 서버에서 관리
)

# 라우터가 자동으로 strong/weak 모델 선택
# threshold 값으로 라우팅 민감도 조절 (0.0~1.0)
response = client.chat.completions.create(
    model="router-mf-0.11593",  # router-{라우터명}-{threshold}
    messages=[
        {"role": "user", "content": "양자 컴퓨팅의 큐비트 오류 정정 방법을 설명해줘"}
    ],
)

# threshold가 낮을수록 strong 모델 사용 비율 증가
# 0.5 → ~50% strong 모델 사용
# 0.1 → ~90% strong 모델 사용 (품질 우선)
# 0.9 → ~10% strong 모델 사용 (비용 우선)
print(response.choices[0].message.content)
print(f"사용된 모델: {response.model}")

# FrugalGPT 스타일 캐스케이드 구현
from openai import OpenAI
from anthropic import Anthropic
import time
from dataclasses import dataclass
from typing import Optional

@dataclass
class ModelTier:
    name: str
    provider: str
    cost_per_1k_tokens: float
    quality_threshold: float  # 이 점수 이상이면 응답 반환

# 비용 오름차순으로 모델 티어 정의
MODEL_CASCADE = [
    ModelTier("gpt-4o-mini", "openai", 0.00015, 0.7),
    ModelTier("claude-3-5-haiku-20241022", "anthropic", 0.001, 0.8),
    ModelTier("claude-sonnet-4-20250514", "anthropic", 0.003, 0.85),
    ModelTier("gpt-4o", "openai", 0.005, 0.0),  # 최종 단계: 항상 반환
]

openai_client = OpenAI()
anthropic_client = Anthropic()


def call_model(model: ModelTier, query: str) -> str:
    """모델 프로바이더에 따라 적절한 API 호출"""
    if model.provider == "openai":
        resp = openai_client.chat.completions.create(
            model=model.name,
            messages=[{"role": "user", "content": query}],
            temperature=0.3,
        )
        return resp.choices[0].message.content
    elif model.provider == "anthropic":
        resp = anthropic_client.messages.create(
            model=model.name,
            max_tokens=2048,
            messages=[{"role": "user", "content": query}],
        )
        return resp.content[0].text


def estimate_quality(query: str, response: str) -> float:
    """응답 품질 판정기 - 경량 모델로 품질 점수 반환"""
    judge_prompt = f"""다음 질문과 답변의 품질을 0.0~1.0 점수로 평가하세요.
평가 기준: 정확성, 완전성, 관련성, 명확성
질문: {query}
답변: {response}
점수만 숫자로 반환하세요 (예: 0.85)"""

    resp = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": judge_prompt}],
        temperature=0.0,
        max_tokens=10,
    )
    try:
        return float(resp.choices[0].message.content.strip())
    except ValueError:
        return 0.0  # 파싱 실패 시 낮은 점수 → 상위 모델로 에스컬레이션


def frugal_cascade(query: str) -> dict:
    """FrugalGPT 캐스케이드 실행"""
    results = []

    for tier in MODEL_CASCADE:
        start = time.time()
        response = call_model(tier, query)
        latency = time.time() - start

        # 최종 티어는 품질 판정 생략
        if tier.quality_threshold == 0.0:
            return {
                "response": response,
                "model": tier.name,
                "latency": latency,
                "cascade_depth": len(results) + 1,
                "total_cost_ratio": sum(r["cost"] for r in results) + tier.cost_per_1k_tokens,
            }

        quality = estimate_quality(query, response)

        results.append({
            "model": tier.name,
            "quality": quality,
            "cost": tier.cost_per_1k_tokens,
            "latency": latency,
        })

        if quality >= tier.quality_threshold:
            return {
                "response": response,
                "model": tier.name,
                "quality_score": quality,
                "latency": latency,
                "cascade_depth": len(results),
            }

    # 이론적으로 도달하지 않음 (마지막 티어에서 무조건 반환)
    return {"response": response, "model": MODEL_CASCADE[-1].name}

# Semantic Router를 활용한 도메인별 모델 라우팅
# pip install semantic-router

from semantic_router import Route, RouteLayer
from semantic_router.encoders import OpenAIEncoder

# 라우트 정의: 각 라우트에 대표 발화(utterance) 지정
simple_route = Route(
    name="simple",
    utterances=[
        "오늘 날씨 어때?",
        "서울의 인구는?",
        "안녕하세요",
        "이 단어의 뜻이 뭐야?",
        "1+1은?",
        "현재 시간 알려줘",
    ],
)

coding_route = Route(
    name="coding",
    utterances=[
        "Python으로 퀵소트 구현해줘",
        "React 컴포넌트에서 useEffect 메모리 누수 해결",
        "Kubernetes Pod OOMKilled 디버깅",
        "SQL 쿼리 최적화 방법",
        "gRPC와 REST API 성능 비교 코드",
    ],
)

reasoning_route = Route(
    name="reasoning",
    utterances=[
        "이 논문의 방법론을 비판적으로 분석해줘",
        "GDP 성장률과 실업률의 상관관계를 경제학적으로 설명해",
        "양자 얽힘과 양자 텔레포테이션의 차이를 수학적으로 유도해",
        "RLHF와 DPO의 이론적 한계를 비교 분석해줘",
    ],
)

# 인코더 및 라우트 레이어 초기화
encoder = OpenAIEncoder(name="text-embedding-3-small")
route_layer = RouteLayer(
    encoder=encoder,
    routes=[simple_route, coding_route, reasoning_route],
)

# 라우트에 따른 모델 매핑
MODEL_MAP = {
    "simple": "gpt-4o-mini",        # 저비용 모델
    "coding": "claude-sonnet-4-20250514",  # 코딩 특화
    "reasoning": "gpt-4o",          # 고성능 추론
    None: "claude-sonnet-4-20250514",      # 기본 폴백
}


def route_query(query: str) -> dict:
    """쿼리를 분석하여 최적 모델 결정"""
    route_result = route_layer(query)
    selected_model = MODEL_MAP.get(route_result.name, MODEL_MAP[None])

    return {
        "query": query,
        "route": route_result.name,
        "confidence": route_result.similarity_score,
        "model": selected_model,
    }


# 사용 예시
queries = [
    "파이썬에서 리스트 정렬하는 법",
    "트랜스포머 아키텍처의 어텐션 메커니즘을 수학적으로 설명해",
    "내일 부산 날씨",
]

for q in queries:
    result = route_query(q)
    print(f"쿼리: {q}")
    print(f"  라우트: {result['route']} → 모델: {result['model']}")
    print(f"  신뢰도: {result['confidence']:.3f}")

사용자 요청 → API Gateway → 시맨틱 캐시 검색
                              ↓ (캐시 미스)
                         쿼리 분류기 (복잡도/도메인 분석)
                              ↓
                    ┌─────────┼─────────┐
                    ↓         ↓         ↓
               소형 모델   중형 모델   대형 모델
              (GPT-4o-mini) (Sonnet)  (GPT-4o/Opus)
                    ↓         ↓         ↓
                    └─────────┼─────────┘
                              ↓
                         품질 게이트 (캐스케이드 판단)
                              ↓
                    ┌── 품질 충족 → 응답 반환 + 캐시 저장
                    └── 품질 미달 → 상위 모델 에스컬레이션
                              ↓
                         메트릭 수집 (비용/레이턴시/품질)

// multi-model-orchestrator.ts
import OpenAI from 'openai'
import Anthropic from '@anthropic-ai/sdk'

interface ModelConfig {
  id: string
  provider: 'openai' | 'anthropic'
  costPer1kInput: number
  costPer1kOutput: number
  maxTokens: number
  tier: 'small' | 'medium' | 'large'
}

interface RoutingDecision {
  model: ModelConfig
  reason: string
  confidence: number
}

interface OrchestratorResult {
  response: string
  model: string
  tier: string
  latencyMs: number
  estimatedCost: number
  cascadeDepth: number
  cacheHit: boolean
}

// 모델 카탈로그 정의
const MODEL_CATALOG: ModelConfig[] = [
  {
    id: 'gpt-4o-mini',
    provider: 'openai',
    costPer1kInput: 0.00015,
    costPer1kOutput: 0.0006,
    maxTokens: 16384,
    tier: 'small',
  },
  {
    id: 'claude-3-5-haiku-20241022',
    provider: 'anthropic',
    costPer1kInput: 0.001,
    costPer1kOutput: 0.005,
    maxTokens: 8192,
    tier: 'small',
  },
  {
    id: 'claude-sonnet-4-20250514',
    provider: 'anthropic',
    costPer1kInput: 0.003,
    costPer1kOutput: 0.015,
    maxTokens: 8192,
    tier: 'medium',
  },
  {
    id: 'gpt-4o',
    provider: 'openai',
    costPer1kInput: 0.005,
    costPer1kOutput: 0.015,
    maxTokens: 16384,
    tier: 'large',
  },
]

class QueryClassifier {
  private openai: OpenAI

  constructor(openai: OpenAI) {
    this.openai = openai
  }

  async classify(query: string): Promise<RoutingDecision> {
    // 규칙 기반 1차 분류 (LLM 호출 없이 즉시 판단)
    const ruleResult = this.ruleBasedClassify(query)
    if (ruleResult) return ruleResult

    // 경량 모델로 2차 분류
    const resp = await this.openai.chat.completions.create({
      model: 'gpt-4o-mini',
      messages: [
        {
          role: 'system',
          content: `쿼리 복잡도를 분류하세요.
JSON으로 응답: {"tier": "small|medium|large", "reason": "...", "confidence": 0.0~1.0}
- small: 단순 질의, 인사, 번역, 요약
- medium: 코딩, 분석, 비교, 구조화된 설명
- large: 복합 추론, 수학 증명, 멀티스텝 분석, 창의적 장문`,
        },
        { role: 'user', content: query },
      ],
      response_format: { type: 'json_object' },
      temperature: 0.0,
      max_tokens: 100,
    })

    const parsed = JSON.parse(resp.choices[0].message.content || '{}')
    const tier = parsed.tier || 'medium'
    const model = MODEL_CATALOG.find((m) => m.tier === tier) || MODEL_CATALOG[2]

    return {
      model,
      reason: parsed.reason || 'LLM 분류기 판단',
      confidence: parsed.confidence || 0.5,
    }
  }

  private ruleBasedClassify(query: string): RoutingDecision | null {
    const len = query.length

    // 매우 짧은 쿼리 → small
    if (len < 30) {
      return {
        model: MODEL_CATALOG[0],
        reason: '짧은 쿼리 (규칙 기반)',
        confidence: 0.9,
      }
    }

    // 코드 관련 키워드 → medium
    const codeKeywords = /\b(코드|구현|함수|클래스|디버그|에러|API|SQL|React|Python)\b/i
    if (codeKeywords.test(query)) {
      return {
        model: MODEL_CATALOG[2], // claude-sonnet
        reason: '코딩 관련 쿼리 (규칙 기반)',
        confidence: 0.8,
      }
    }

    // 복잡한 추론 키워드 → large
    const reasoningKeywords = /\b(증명|분석|비교.*차이|수학적|논리적|전략.*수립|아키텍처.*설계)\b/
    if (reasoningKeywords.test(query)) {
      return {
        model: MODEL_CATALOG[3], // gpt-4o
        reason: '복합 추론 쿼리 (규칙 기반)',
        confidence: 0.75,
      }
    }

    return null // 규칙으로 판단 불가 → LLM 분류기 사용
  }
}

# 시맨틱 캐시 구현 (Redis + 벡터 유사도)
import hashlib
import json
import time
import numpy as np
from openai import OpenAI
from redis import Redis

client = OpenAI()
redis_client = Redis(host="localhost", port=6379, db=0)

CACHE_TTL = 3600  # 1시간
SIMILARITY_THRESHOLD = 0.92  # 유사도 임계값


def get_embedding(text: str) -> list[float]:
    """텍스트의 임베딩 벡터 생성"""
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=text,
    )
    return resp.data[0].embedding


def cosine_similarity(a: list[float], b: list[float]) -> float:
    """코사인 유사도 계산"""
    a_np, b_np = np.array(a), np.array(b)
    return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))


class SemanticCache:
    def __init__(self, namespace: str = "llm_cache"):
        self.namespace = namespace

    def _cache_key(self, idx: int) -> str:
        return f"{self.namespace}:entry:{idx}"

    def _counter_key(self) -> str:
        return f"{self.namespace}:counter"

    def get(self, query: str) -> dict | None:
        """유사 쿼리의 캐시된 응답 검색"""
        query_embedding = get_embedding(query)
        counter = int(redis_client.get(self._counter_key()) or 0)

        best_match = None
        best_similarity = 0.0

        for i in range(counter):
            entry_raw = redis_client.get(self._cache_key(i))
            if not entry_raw:
                continue

            entry = json.loads(entry_raw)
            similarity = cosine_similarity(query_embedding, entry["embedding"])

            if similarity > best_similarity and similarity >= SIMILARITY_THRESHOLD:
                best_similarity = similarity
                best_match = entry

        if best_match:
            return {
                "response": best_match["response"],
                "model": best_match["model"],
                "similarity": best_similarity,
                "cached_at": best_match["timestamp"],
            }

        return None

    def put(self, query: str, response: str, model: str):
        """응답을 캐시에 저장"""
        embedding = get_embedding(query)
        counter = int(redis_client.get(self._counter_key()) or 0)

        entry = {
            "query": query,
            "response": response,
            "model": model,
            "embedding": embedding,
            "timestamp": time.time(),
        }

        redis_client.setex(
            self._cache_key(counter),
            CACHE_TTL,
            json.dumps(entry),
        )
        redis_client.incr(self._counter_key())

# 비용 추적 및 분석 시스템
from dataclasses import dataclass, field
from collections import defaultdict
from datetime import datetime, timedelta
import json


@dataclass
class TokenUsage:
    model: str
    input_tokens: int
    output_tokens: int
    timestamp: datetime
    route: str  # 어떤 라우트로 분류되었는지
    cascade_depth: int = 1
    quality_score: float = 0.0


class CostAnalyzer:
    # 2026년 3월 기준 주요 모델 가격 (USD per 1K tokens)
    PRICING = {
        "gpt-4o": {"input": 0.0025, "output": 0.01},
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
        "claude-opus-4-20250514": {"input": 0.015, "output": 0.075},
        "claude-sonnet-4-20250514": {"input": 0.003, "output": 0.015},
        "claude-3-5-haiku-20241022": {"input": 0.001, "output": 0.005},
    }

    def __init__(self):
        self.usage_log: list[TokenUsage] = []

    def log(self, usage: TokenUsage):
        self.usage_log.append(usage)

    def calculate_cost(self, usage: TokenUsage) -> float:
        """단건 비용 계산"""
        pricing = self.PRICING.get(usage.model)
        if not pricing:
            return 0.0

        input_cost = (usage.input_tokens / 1000) * pricing["input"]
        output_cost = (usage.output_tokens / 1000) * pricing["output"]
        return input_cost + output_cost

    def daily_report(self, date: datetime = None) -> dict:
        """일별 비용 리포트 생성"""
        date = date or datetime.now()
        day_start = date.replace(hour=0, minute=0, second=0)
        day_end = day_start + timedelta(days=1)

        day_logs = [
            u for u in self.usage_log
            if day_start <= u.timestamp < day_end
        ]

        model_costs = defaultdict(float)
        route_costs = defaultdict(float)
        total_cost = 0.0
        total_requests = len(day_logs)

        for usage in day_logs:
            cost = self.calculate_cost(usage)
            model_costs[usage.model] += cost
            route_costs[usage.route] += cost
            total_cost += cost

        # 라우팅 없이 모든 요청을 GPT-4o로 보냈을 때 비용 추정
        baseline_cost = sum(
            (u.input_tokens / 1000) * self.PRICING["gpt-4o"]["input"]
            + (u.output_tokens / 1000) * self.PRICING["gpt-4o"]["output"]
            for u in day_logs
        )

        return {
            "date": date.strftime("%Y-%m-%d"),
            "total_requests": total_requests,
            "total_cost_usd": round(total_cost, 4),
            "baseline_cost_usd": round(baseline_cost, 4),
            "savings_pct": round((1 - total_cost / baseline_cost) * 100, 1) if baseline_cost > 0 else 0,
            "cost_by_model": dict(model_costs),
            "cost_by_route": dict(route_costs),
            "avg_cost_per_request": round(total_cost / total_requests, 6) if total_requests > 0 else 0,
        }

# 프로바이더별 서킷 브레이커 구현
import time
from enum import Enum
from threading import Lock


class CircuitState(Enum):
    CLOSED = "closed"      # 정상 상태
    OPEN = "open"          # 차단 상태
    HALF_OPEN = "half_open"  # 시험 상태


class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        half_open_max_calls: int = 3,
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = 0.0
        self.half_open_calls = 0
        self._lock = Lock()

    def can_execute(self) -> bool:
        """현재 상태에서 요청 실행 가능 여부"""
        with self._lock:
            if self.state == CircuitState.CLOSED:
                return True

            if self.state == CircuitState.OPEN:
                # 복구 타임아웃 경과 시 HALF_OPEN으로 전환
                if time.time() - self.last_failure_time >= self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.half_open_calls = 0
                    return True
                return False

            if self.state == CircuitState.HALF_OPEN:
                return self.half_open_calls < self.half_open_max_calls

            return False

    def record_success(self):
        """성공 기록"""
        with self._lock:
            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                if self.success_count >= self.half_open_max_calls:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0
                    self.success_count = 0
            elif self.state == CircuitState.CLOSED:
                self.failure_count = 0

    def record_failure(self):
        """실패 기록"""
        with self._lock:
            self.failure_count += 1
            self.last_failure_time = time.time()

            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.OPEN  # 다시 차단
            elif self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN


# 프로바이더별 서킷 브레이커 관리
provider_circuits: dict[str, CircuitBreaker] = {
    "openai": CircuitBreaker(failure_threshold=5, recovery_timeout=60),
    "anthropic": CircuitBreaker(failure_threshold=5, recovery_timeout=60),
    "google": CircuitBreaker(failure_threshold=3, recovery_timeout=120),
}

# 동일 티어 폴백 매핑
FALLBACK_MAP = {
    "gpt-4o-mini": ["claude-3-5-haiku-20241022"],
    "claude-3-5-haiku-20241022": ["gpt-4o-mini"],
    "claude-sonnet-4-20250514": ["gpt-4o"],
    "gpt-4o": ["claude-sonnet-4-20250514"],
}


def get_available_model(
    primary_model: str,
    primary_provider: str,
) -> tuple[str, str]:
    """사용 가능한 모델 반환 (서킷 브레이커 확인 포함)"""
    # 1차: 기본 모델의 프로바이더 확인
    if provider_circuits[primary_provider].can_execute():
        return primary_model, primary_provider

    # 2차: 폴백 모델 탐색
    fallbacks = FALLBACK_MAP.get(primary_model, [])
    for fb_model in fallbacks:
        fb_provider = "anthropic" if "claude" in fb_model else "openai"
        if provider_circuits[fb_provider].can_execute():
            return fb_model, fb_provider

    # 3차: 모든 프로바이더 장애 시 에러
    raise RuntimeError(
        f"모든 프로바이더 서킷 OPEN 상태. "
        f"primary={primary_model}, fallbacks={fallbacks}"
    )

// 안전한 멀티 프로바이더 API 키 관리
// provider-config.ts
import { SecretManagerServiceClient } from '@google-cloud/secret-manager'

interface ProviderCredentials {
  apiKey: string
  orgId?: string
  rateLimit: number // RPM
  lastRotated: Date
}

class ProviderKeyManager {
  private secretClient: SecretManagerServiceClient
  private cache: Map<string, ProviderCredentials> = new Map()
  private cacheTTL = 300_000 // 5분
  private lastFetch: Map<string, number> = new Map()

  constructor() {
    this.secretClient = new SecretManagerServiceClient()
  }

  async getCredentials(provider: string): Promise<ProviderCredentials> {
    const now = Date.now()
    const lastFetched = this.lastFetch.get(provider) || 0

    // 캐시가 유효하면 캐시에서 반환
    if (now - lastFetched < this.cacheTTL && this.cache.has(provider)) {
      return this.cache.get(provider)!
    }

    // Secret Manager에서 키 조회
    const secretName = `projects/my-project/secrets/llm-${provider}-key/versions/latest`
    const [version] = await this.secretClient.accessSecretVersion({
      name: secretName,
    })

    const apiKey = version.payload?.data?.toString() || ''
    const credentials: ProviderCredentials = {
      apiKey,
      rateLimit: this.getDefaultRateLimit(provider),
      lastRotated: new Date(),
    }

    this.cache.set(provider, credentials)
    this.lastFetch.set(provider, now)

    return credentials
  }

  private getDefaultRateLimit(provider: string): number {
    const limits: Record<string, number> = {
      openai: 500,
      anthropic: 300,
      google: 200,
    }
    return limits[provider] || 100
  }
}