LLM 성능 향상 시리즈 3편: 라우팅·앙상블로 비용 70% 절감

from sklearn.linear_model import LogisticRegression
import numpy as np

class QueryRouter:
    """쿼리 복잡도를 분류해 적합한 모델로 라우팅"""

    COMPLEXITY_LEVELS = {
        "simple": "claude-haiku-4-5",    # 단순 조회, FAQ
        "medium": "claude-sonnet-4-6",   # 일반 추론
        "complex": "claude-opus-4-7",    # 복잡한 분석, 코드
    }

    def __init__(self):
        self.classifier = LogisticRegression()
        # 특징: 토큰 수, 코드 포함 여부, 수치 복잡도, 도메인 키워드 수
        self.feature_extractors = [
            lambda q: len(q.split()),
            lambda q: int(any(kw in q for kw in ["코드", "함수", "구현", "sql"])),
            lambda q: q.count("?"),
            lambda q: int(len(q) > 200),
        ]

    def extract_features(self, query: str) -> np.ndarray:
        return np.array([f(query) for f in self.feature_extractors]).reshape(1, -1)

    def route(self, query: str) -> str:
        if not hasattr(self.classifier, "classes_"):
            return "medium"  # 미학습 시 기본값
        level = self.classifier.predict(self.extract_features(query))[0]
        return self.COMPLEXITY_LEVELS[level]

    async def query(self, user_input: str) -> str:
        model = self.route(user_input)
        return await llm_call(model, user_input)

def simple_router(query: str) -> str:
    tokens = len(query.split())
    has_code = any(kw in query.lower() for kw in ["코드", "sql", "함수", "알고리즘"])
    has_numbers = sum(c.isdigit() for c in query) > 5
    is_long = tokens > 100

    if has_code or (is_long and has_numbers):
        return "claude-opus-4-7"
    elif tokens > 30 or has_numbers:
        return "claude-sonnet-4-6"
    else:
        return "claude-haiku-4-5"

async def escalating_query(query: str) -> dict:
    models = [
        ("claude-haiku-4-5", 0.85),   # (모델, 신뢰도 임계값)
        ("claude-sonnet-4-6", 0.90),
        ("claude-opus-4-7", None),    # 최종 — 임계값 없음
    ]

    for model, threshold in models:
        response = await llm_call(model, query)
        confidence = await estimate_confidence(response)

        if threshold is None or confidence >= threshold:
            return {
                "answer": response,
                "model_used": model,
                "confidence": confidence
            }

        print(f"{model} 신뢰도 {confidence:.2f} < {threshold} → 상위 모델로 승격")

async def ensemble_query(query: str, models: list, weights: list = None) -> str:
    responses = await asyncio.gather(*[llm_call(m, query) for m in models])

    if weights is None:
        weights = [1.0] * len(models)

    # 응답을 임베딩으로 변환 후 가중 평균 클러스터링
    embeddings = await asyncio.gather(*[embed(r) for r in responses])
    centroid = np.average(embeddings, axis=0, weights=weights)

    # 클러스터 중심에 가장 가까운 응답 선택
    distances = [np.linalg.norm(e - centroid) for e in embeddings]
    best_idx = np.argmin(distances)
    return responses[best_idx]