5장: 평가 하네스 — 모델 성능 측정 파이프라인

# MMLU 벤치마크 실행 (5-shot)
lm_eval --model openai-completions \
    --model_args model=gpt-4o \
    --tasks mmlu \
    --num_fewshot 5 \
    --output_path results/
 
# 여러 벤치마크 동시 실행
lm_eval --model anthropic \
    --model_args model=claude-sonnet-4-20250514 \
    --tasks mmlu,hellaswag,arc_challenge,gsm8k \
    --output_path results/

# 한국어 도메인 특화 평가 태스크
task: korean_medical_qa
dataset_path: my_org/korean_medical_qa
dataset_name: default
output_type: multiple_choice
training_split: train
test_split: test
doc_to_text: "질문: {{question}}\n선택지:\nA. {{choices[0]}}\nB. {{choices[1]}}\nC. {{choices[2]}}\nD. {{choices[3]}}\n정답:"
doc_to_target: "{{['A', 'B', 'C', 'D'][answer]}}"
metric_list:
  - metric: acc
    aggregation: mean
    higher_is_better: true
  - metric: acc_norm
    aggregation: mean
    higher_is_better: true
num_fewshot: 3

from inspect_ai import Task, task
from inspect_ai.dataset import json_dataset
from inspect_ai.scorer import model_graded_fact
from inspect_ai.solver import generate, system_message
 
 
@task
def korean_qa_eval():
    """한국어 질의응답 평가 태스크"""
    return Task(
        dataset=json_dataset("data/korean_qa.json"),
        solver=[
            system_message(
                "당신은 한국어에 능통한 AI 어시스턴트입니다. "
                "질문에 정확하고 간결하게 답변하세요."
            ),
            generate(),
        ],
        scorer=model_graded_fact(),
    )
 
 
@task
def agent_coding_eval():
    """코딩 에이전트 평가 태스크"""
    return Task(
        dataset=json_dataset("data/coding_tasks.json"),
        solver=[
            system_message("코딩 문제를 해결하세요."),
            use_tools([
                read_file(),
                write_file(),
                execute_code(),
            ]),
            generate(),
        ],
        scorer=code_execution_scorer(),
    )

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any
 
 
@dataclass
class EvalExample:
    """평가 데이터 항목"""
    id: str
    input_text: str
    expected: Any
    metadata: dict = field(default_factory=dict)
 
 
@dataclass
class EvalResult:
    """개별 평가 결과"""
    example_id: str
    model_output: str
    scores: dict[str, float]
    latency_ms: float
    token_usage: dict[str, int]
 
 
class Metric(ABC):
    """메트릭 인터페이스"""
 
    @abstractmethod
    def name(self) -> str:
        ...
 
    @abstractmethod
    def compute(
        self, output: str, expected: Any
    ) -> float:
        ...
 
 
class ExactMatchMetric(Metric):
    def name(self) -> str:
        return "exact_match"
 
    def compute(self, output: str, expected: Any) -> float:
        return 1.0 if output.strip() == str(expected).strip() else 0.0
 
 
class ContainsMetric(Metric):
    def name(self) -> str:
        return "contains"
 
    def compute(self, output: str, expected: Any) -> float:
        return 1.0 if str(expected) in output else 0.0
 
 
class SemanticSimilarityMetric(Metric):
    def __init__(self, embedding_model):
        self.embedding_model = embedding_model
 
    def name(self) -> str:
        return "semantic_similarity"
 
    def compute(self, output: str, expected: Any) -> float:
        emb_output = self.embedding_model.encode(output)
        emb_expected = self.embedding_model.encode(str(expected))
        return cosine_similarity(emb_output, emb_expected)
 
 
class EvalHarness:
    """커스텀 평가 하네스"""
 
    def __init__(self, model_fn, metrics: list[Metric]):
        self.model_fn = model_fn
        self.metrics = metrics
 
    async def evaluate(
        self,
        examples: list[EvalExample],
    ) -> dict:
        results = []
 
        for example in examples:
            import time
            start = time.monotonic()
            output = await self.model_fn(example.input_text)
            elapsed = (time.monotonic() - start) * 1000
 
            scores = {}
            for metric in self.metrics:
                scores[metric.name()] = metric.compute(
                    output, example.expected
                )
 
            results.append(EvalResult(
                example_id=example.id,
                model_output=output,
                scores=scores,
                latency_ms=elapsed,
                token_usage={},  # 실제로는 모델 응답에서 추출
            ))
 
        return self._aggregate(results)
 
    def _aggregate(self, results: list[EvalResult]) -> dict:
        """결과 집계"""
        summary = {"total": len(results)}
 
        for metric in self.metrics:
            scores = [r.scores[metric.name()] for r in results]
            summary[metric.name()] = {
                "mean": sum(scores) / len(scores),
                "min": min(scores),
                "max": max(scores),
            }
 
        avg_latency = sum(r.latency_ms for r in results) / len(results)
        summary["avg_latency_ms"] = avg_latency
 
        return summary

from dataclasses import dataclass
 
 
@dataclass
class ComparisonResult:
    model_a: str
    model_b: str
    metrics: dict[str, dict[str, float]]
    winner: str
    recommendation: str
 
 
async def compare_models(
    model_a_fn,
    model_b_fn,
    model_a_name: str,
    model_b_name: str,
    eval_examples: list[EvalExample],
    metrics: list[Metric],
) -> ComparisonResult:
    """두 모델의 체계적 비교"""
 
    harness_a = EvalHarness(model_a_fn, metrics)
    harness_b = EvalHarness(model_b_fn, metrics)
 
    results_a = await harness_a.evaluate(eval_examples)
    results_b = await harness_b.evaluate(eval_examples)
 
    # 메트릭별 비교
    comparison = {}
    a_wins = 0
    b_wins = 0
 
    for metric in metrics:
        name = metric.name()
        score_a = results_a[name]["mean"]
        score_b = results_b[name]["mean"]
        comparison[name] = {
            model_a_name: score_a,
            model_b_name: score_b,
            "difference": score_b - score_a,
        }
        if score_b > score_a:
            b_wins += 1
        elif score_a > score_b:
            a_wins += 1
 
    winner = model_a_name if a_wins > b_wins else model_b_name
 
    # 지연시간 및 비용 비교
    latency_a = results_a["avg_latency_ms"]
    latency_b = results_b["avg_latency_ms"]
 
    recommendation = (
        f"{winner}이(가) {max(a_wins, b_wins)}/{len(metrics)} "
        f"메트릭에서 우수합니다. "
        f"지연시간: {model_a_name}={latency_a:.0f}ms, "
        f"{model_b_name}={latency_b:.0f}ms"
    )
 
    return ComparisonResult(
        model_a=model_a_name,
        model_b=model_b_name,
        metrics=comparison,
        winner=winner,
        recommendation=recommendation,
    )