9장: 자동화된 모델 비교 파이프라인

import math
from dataclasses import dataclass, field
 
 
@dataclass
class ModelRating:
    model_name: str
    rating: float = 1500.0
    matches: int = 0
    wins: int = 0
    losses: int = 0
    draws: int = 0
 
 
class ELOSystem:
    """모델 비교를 위한 ELO 레이팅 시스템."""
    
    def __init__(self, k_factor: float = 32.0, initial_rating: float = 1500.0):
        self.k_factor = k_factor
        self.initial_rating = initial_rating
        self.ratings: dict[str, ModelRating] = {}
    
    def register_model(self, model_name: str) -> None:
        """새 모델을 등록합니다."""
        if model_name not in self.ratings:
            self.ratings[model_name] = ModelRating(
                model_name=model_name,
                rating=self.initial_rating,
            )
    
    def expected_score(self, rating_a: float, rating_b: float) -> float:
        """모델 A가 모델 B를 이길 기대 확률을 계산합니다."""
        return 1.0 / (1.0 + math.pow(10, (rating_b - rating_a) / 400.0))
    
    def record_match(
        self,
        model_a: str,
        model_b: str,
        outcome: str,  # "a_wins", "b_wins", "draw"
    ) -> tuple[float, float]:
        """대결 결과를 기록하고 레이팅을 업데이트합니다."""
        
        ra = self.ratings[model_a]
        rb = self.ratings[model_b]
        
        expected_a = self.expected_score(ra.rating, rb.rating)
        expected_b = 1.0 - expected_a
        
        if outcome == "a_wins":
            actual_a, actual_b = 1.0, 0.0
            ra.wins += 1
            rb.losses += 1
        elif outcome == "b_wins":
            actual_a, actual_b = 0.0, 1.0
            ra.losses += 1
            rb.wins += 1
        else:  # draw
            actual_a, actual_b = 0.5, 0.5
            ra.draws += 1
            rb.draws += 1
        
        ra.rating += self.k_factor * (actual_a - expected_a)
        rb.rating += self.k_factor * (actual_b - expected_b)
        ra.matches += 1
        rb.matches += 1
        
        return ra.rating, rb.rating
    
    def leaderboard(self) -> list[ModelRating]:
        """레이팅 순으로 정렬된 리더보드를 반환합니다."""
        return sorted(
            self.ratings.values(),
            key=lambda r: r.rating,
            reverse=True,
        )

import random
from itertools import combinations
 
 
class AutoMatchmaker:
    """모델 간 자동 대결을 관리합니다."""
    
    def __init__(self, elo_system: ELOSystem, judge: LLMJudge):
        self.elo = elo_system
        self.judge = judge
    
    async def run_tournament(
        self,
        models: dict[str, callable],
        eval_cases: list[dict],
        rounds: int = 3,
    ) -> list[ModelRating]:
        """라운드 로빈 토너먼트를 실행합니다."""
        
        model_pairs = list(combinations(models.keys(), 2))
        
        for round_num in range(rounds):
            random.shuffle(model_pairs)
            
            for model_a, model_b in model_pairs:
                # 무작위 평가 케이스 선택
                case = random.choice(eval_cases)
                
                # 두 모델의 응답 생성
                response_a = await models[model_a](case["query"])
                response_b = await models[model_b](case["query"])
                
                # LLM 심판의 판정
                outcome = await self._judge_pair(
                    query=case["query"],
                    response_a=response_a,
                    response_b=response_b,
                )
                
                self.elo.record_match(model_a, model_b, outcome)
        
        return self.elo.leaderboard()
    
    async def _judge_pair(self, query: str, response_a: str, response_b: str) -> str:
        """두 응답을 비교하여 승자를 판정합니다."""
        result = self.judge.evaluate(
            query=query,
            response=f"응답 A: {response_a}\n\n응답 B: {response_b}",
            criteria="두 응답 중 질문에 더 정확하고 유용하게 답변한 것을 선택하세요. "
                     "A가 더 좋으면 'A', B가 더 좋으면 'B', 비슷하면 'DRAW'로 답하세요.",
        )
        
        detail = result.details.upper()
        if "A" in detail and "B" not in detail:
            return "a_wins"
        elif "B" in detail and "A" not in detail:
            return "b_wins"
        return "draw"

import random
import hashlib
from dataclasses import dataclass
from datetime import datetime
 
 
@dataclass
class ABTestConfig:
    """A/B 테스트 설정."""
    test_name: str
    model_a: str             # 대조군 (현재 모델)
    model_b: str             # 실험군 (새 모델)
    traffic_split: float     # B 모델로 보낼 트래픽 비율 (0.0 - 1.0)
    min_samples: int         # 최소 샘플 수
    confidence_level: float  # 신뢰 수준 (기본 0.95)
    metrics: list[str]       # 추적할 메트릭
    start_date: str
    max_duration_days: int
 
 
class ABTestRouter:
    """요청을 A/B 그룹으로 라우팅합니다."""
    
    def __init__(self, config: ABTestConfig):
        self.config = config
    
    def assign_group(self, user_id: str) -> str:
        """사용자를 일관되게 A/B 그룹에 할당합니다."""
        # 해시 기반 할당으로 동일 사용자는 항상 같은 그룹
        hash_val = int(hashlib.md5(
            f"{self.config.test_name}:{user_id}".encode()
        ).hexdigest(), 16)
        
        if (hash_val % 100) / 100 < self.config.traffic_split:
            return "B"
        return "A"
 
 
class ABTestAnalyzer:
    """A/B 테스트 결과를 분석합니다."""
    
    def analyze(
        self,
        group_a_metrics: dict[str, list[float]],
        group_b_metrics: dict[str, list[float]],
        confidence_level: float = 0.95,
    ) -> dict:
        """두 그룹의 메트릭을 비교 분석합니다."""
        import numpy as np
        from scipy import stats
        
        results = {}
        for metric_name in group_a_metrics:
            a_values = np.array(group_a_metrics[metric_name])
            b_values = np.array(group_b_metrics[metric_name])
            
            # 기본 통계
            mean_a = np.mean(a_values)
            mean_b = np.mean(b_values)
            relative_change = (mean_b - mean_a) / mean_a if mean_a != 0 else 0
            
            # 통계 검정
            t_stat, p_value = stats.ttest_ind(a_values, b_values)
            significant = p_value < (1 - confidence_level)
            
            results[metric_name] = {
                "mean_a": mean_a,
                "mean_b": mean_b,
                "relative_change": relative_change,
                "p_value": p_value,
                "significant": significant,
                "recommendation": self._get_recommendation(relative_change, significant),
            }
        
        return results
    
    @staticmethod
    def _get_recommendation(change: float, significant: bool) -> str:
        if not significant:
            return "유의미한 차이 없음 - 추가 데이터 수집 필요"
        if change > 0.05:
            return "B 모델로 교체 권장 (5% 이상 개선)"
        if change > 0:
            return "B 모델 소폭 개선 - 비용/복잡성 고려하여 결정"
        return "A 모델 유지 권장 (B 모델이 열등)"

from dataclasses import dataclass
 
 
@dataclass
class ModelProfile:
    """모델의 성능 프로파일."""
    name: str
    quality_score: float        # 0-1 (평가 점수)
    avg_latency_ms: float       # 평균 지연시간 (밀리초)
    p99_latency_ms: float       # 99퍼센타일 지연시간
    cost_per_1k_tokens: float   # 1000 토큰당 비용 (USD)
    avg_output_tokens: float    # 평균 출력 토큰 수
 
 
class TradeoffAnalyzer:
    """비용/지연시간/품질 트레이드오프를 분석합니다."""
    
    def __init__(self, models: list[ModelProfile]):
        self.models = models
    
    def pareto_frontier(self) -> list[ModelProfile]:
        """파레토 최적 모델을 찾습니다.
        
        품질이 높으면서 비용이 낮은 모델이 파레토 최적입니다.
        어떤 모델보다 모든 차원에서 열등한 모델은 제외됩니다.
        """
        frontier = []
        for model in self.models:
            dominated = False
            for other in self.models:
                if other.name == model.name:
                    continue
                # other가 모든 차원에서 model보다 좋거나 같으면 model은 지배됨
                if (other.quality_score >= model.quality_score and
                    other.cost_per_1k_tokens <= model.cost_per_1k_tokens and
                    other.avg_latency_ms <= model.avg_latency_ms and
                    (other.quality_score > model.quality_score or
                     other.cost_per_1k_tokens < model.cost_per_1k_tokens or
                     other.avg_latency_ms < model.avg_latency_ms)):
                    dominated = True
                    break
            if not dominated:
                frontier.append(model)
        return frontier
    
    def recommend(
        self,
        min_quality: float = 0.0,
        max_latency_ms: float = float("inf"),
        max_cost_per_1k: float = float("inf"),
    ) -> list[ModelProfile]:
        """제약 조건을 만족하는 최적 모델을 추천합니다."""
        candidates = [
            m for m in self.models
            if (m.quality_score >= min_quality and
                m.avg_latency_ms <= max_latency_ms and
                m.cost_per_1k_tokens <= max_cost_per_1k)
        ]
        
        # 품질 기준 내림차순 정렬
        return sorted(candidates, key=lambda m: m.quality_score, reverse=True)
    
    def cost_projection(
        self,
        model: ModelProfile,
        monthly_requests: int,
    ) -> dict:
        """월간 비용을 추정합니다."""
        total_tokens = monthly_requests * model.avg_output_tokens
        monthly_cost = (total_tokens / 1000) * model.cost_per_1k_tokens
        
        return {
            "model": model.name,
            "monthly_requests": monthly_requests,
            "estimated_tokens": total_tokens,
            "estimated_cost_usd": round(monthly_cost, 2),
        }

from datetime import datetime
from pathlib import Path
 
 
class ComparisonReportGenerator:
    """모델 비교 리포트를 자동 생성합니다."""
    
    def generate_markdown(
        self,
        models: list[ModelProfile],
        eval_results: dict,
        tradeoff: TradeoffAnalyzer,
        output_path: str,
    ) -> str:
        """마크다운 형식의 비교 리포트를 생성합니다."""
        
        lines = [
            f"# 모델 비교 리포트",
            f"",
            f"생성일: {datetime.now().strftime('%Y-%m-%d %H:%M')}",
            f"",
            f"## 요약",
            f"",
            f"| 모델 | 품질 점수 | 지연시간 (ms) | 비용 ($/1K tok) |",
            f"|------|----------|-------------|----------------|",
        ]
        
        for m in sorted(models, key=lambda x: x.quality_score, reverse=True):
            lines.append(
                f"| {m.name} | {m.quality_score:.3f} | "
                f"{m.avg_latency_ms:.0f} | ${m.cost_per_1k_tokens:.4f} |"
            )
        
        # 파레토 최적 모델
        frontier = tradeoff.pareto_frontier()
        lines.extend([
            f"",
            f"## 파레토 최적 모델",
            f"",
        ])
        for m in frontier:
            lines.append(f"- **{m.name}**: 품질 {m.quality_score:.3f}, "
                        f"비용 ${m.cost_per_1k_tokens:.4f}/1K tok")
        
        # 추천
        lines.extend([
            f"",
            f"## 제약 조건별 추천",
            f"",
            f"### 최고 품질 우선",
        ])
        best_quality = tradeoff.recommend(min_quality=0.85)
        if best_quality:
            lines.append(f"추천: **{best_quality[0].name}**")
        
        lines.extend([f"", f"### 비용 효율 우선"])
        cost_efficient = tradeoff.recommend(max_cost_per_1k=0.005)
        if cost_efficient:
            lines.append(f"추천: **{cost_efficient[0].name}**")
        
        content = "\n".join(lines)
        Path(output_path).write_text(content, encoding="utf-8")
        return content