2장: 평가 하네스 아키텍처와 핵심 개념

from dataclasses import dataclass, field
from typing import Callable, Any
 
 
@dataclass
class TaskConfig:
    """평가 태스크의 설정을 정의합니다."""
    
    task_name: str
    dataset_path: str
    dataset_name: str | None = None
    dataset_split: str = "test"
    
    # 프롬프트 구성
    prompt_template: str = ""
    system_prompt: str | None = None
    num_fewshot: int = 0
    fewshot_split: str = "validation"
    
    # 메트릭 구성
    metrics: list[str] = field(default_factory=lambda: ["accuracy"])
    
    # 출력 처리
    output_type: str = "generate"  # "generate" | "loglikelihood" | "loglikelihood_rolling"
    max_gen_tokens: int = 256
    stop_sequences: list[str] = field(default_factory=list)
 
 
@dataclass
class TaskInstance:
    """하나의 평가 인스턴스를 나타냅니다."""
    
    task_name: str
    instance_id: str
    prompt: str
    reference: str | list[str]
    metadata: dict[str, Any] = field(default_factory=dict)

from string import Template
 
 
class PromptBuilder:
    """태스크 인스턴스로부터 모델 입력 프롬프트를 생성합니다."""
    
    def __init__(self, config: TaskConfig):
        self.config = config
        self.template = Template(config.prompt_template)
        self._fewshot_examples: list[dict] | None = None
    
    def build_prompt(self, instance: dict, fewshot_examples: list[dict] | None = None) -> str:
        """단일 인스턴스에 대한 전체 프롬프트를 조립합니다."""
        parts: list[str] = []
        
        # few-shot 예제 추가
        if fewshot_examples:
            for example in fewshot_examples:
                parts.append(self._format_example(example, include_answer=True))
            parts.append("")  # 구분자
        
        # 실제 질문 추가
        parts.append(self._format_example(instance, include_answer=False))
        
        return "\n".join(parts)
    
    def _format_example(self, data: dict, include_answer: bool) -> str:
        """단일 예제를 템플릿에 맞게 포맷합니다."""
        text = self.template.safe_substitute(data)
        if include_answer and "answer" in data:
            text += f" {data['answer']}"
        return text

from abc import ABC, abstractmethod
 
 
class ModelBackend(ABC):
    """모든 모델 백엔드가 구현해야 하는 인터페이스입니다."""
    
    @abstractmethod
    def generate(
        self,
        prompts: list[str],
        max_tokens: int = 256,
        temperature: float = 0.0,
        stop: list[str] | None = None,
    ) -> list[str]:
        """프롬프트 목록에 대한 생성 결과를 반환합니다."""
        ...
    
    @abstractmethod
    def loglikelihood(
        self,
        contexts: list[str],
        continuations: list[str],
    ) -> list[float]:
        """각 (context, continuation) 쌍의 로그 확률을 반환합니다."""
        ...
    
    @abstractmethod
    def model_info(self) -> dict:
        """모델 메타데이터를 반환합니다."""
        ...
 
 
class HuggingFaceBackend(ModelBackend):
    """HuggingFace Transformers 모델 백엔드."""
    
    def __init__(self, model_name: str, device: str = "auto", dtype: str = "float16"):
        from transformers import AutoModelForCausalLM, AutoTokenizer
        
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForCausalLM.from_pretrained(
            model_name,
            device_map=device,
            torch_dtype=dtype,
        )
    
    def generate(self, prompts, max_tokens=256, temperature=0.0, stop=None):
        # 구현 생략 - 토크나이저와 모델을 사용한 생성
        ...
    
    def loglikelihood(self, contexts, continuations):
        # 구현 생략 - 토큰별 로그 확률 계산
        ...
    
    def model_info(self):
        return {"type": "huggingface", "name": self.model.config.name_or_path}
 
 
class OpenAIBackend(ModelBackend):
    """OpenAI API 모델 백엔드."""
    
    def __init__(self, model_name: str = "gpt-4o", api_key: str | None = None):
        from openai import OpenAI
        self.client = OpenAI(api_key=api_key)
        self.model_name = model_name
    
    def generate(self, prompts, max_tokens=256, temperature=0.0, stop=None):
        results = []
        for prompt in prompts:
            response = self.client.chat.completions.create(
                model=self.model_name,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=max_tokens,
                temperature=temperature,
                stop=stop,
            )
            results.append(response.choices[0].message.content or "")
        return results
    
    def loglikelihood(self, contexts, continuations):
        raise NotImplementedError("OpenAI API는 토큰 로그 확률 직접 접근을 제한합니다.")
    
    def model_info(self):
        return {"type": "openai", "name": self.model_name}

import asyncio
from dataclasses import dataclass
 
 
@dataclass
class EvalRequest:
    instance_id: str
    prompt: str
    task_name: str
 
 
@dataclass
class EvalResult:
    instance_id: str
    output: str
    metadata: dict
 
 
class ExecutionEngine:
    """평가 요청의 배칭, 병렬 실행, 캐싱을 관리합니다."""
    
    def __init__(
        self,
        backend: ModelBackend,
        batch_size: int = 32,
        max_concurrent: int = 8,
        cache_dir: str | None = None,
    ):
        self.backend = backend
        self.batch_size = batch_size
        self.max_concurrent = max_concurrent
        self.cache = EvalCache(cache_dir) if cache_dir else None
        self._semaphore = asyncio.Semaphore(max_concurrent)
    
    async def execute_all(self, requests: list[EvalRequest]) -> list[EvalResult]:
        """모든 평가 요청을 배치 단위로 실행합니다."""
        # 캐시 히트 확인
        cached, uncached = self._partition_cached(requests)
        
        # 미처리 요청을 배치로 분할
        batches = self._create_batches(uncached, self.batch_size)
        
        # 배치를 병렬로 실행
        results = list(cached)
        batch_results = await asyncio.gather(
            *(self._execute_batch(batch) for batch in batches)
        )
        
        for batch_result in batch_results:
            results.extend(batch_result)
            if self.cache:
                for result in batch_result:
                    self.cache.store(result)
        
        return results
    
    async def _execute_batch(self, batch: list[EvalRequest]) -> list[EvalResult]:
        """단일 배치를 실행합니다."""
        async with self._semaphore:
            prompts = [req.prompt for req in batch]
            outputs = self.backend.generate(prompts)
            return [
                EvalResult(
                    instance_id=req.instance_id,
                    output=output,
                    metadata={"task": req.task_name},
                )
                for req, output in zip(batch, outputs)
            ]
    
    @staticmethod
    def _create_batches(items: list, size: int) -> list[list]:
        return [items[i:i + size] for i in range(0, len(items), size)]

import numpy as np
from scipy import stats
 
 
class MetricsAggregator:
    """개별 인스턴스 결과를 태스크 수준 메트릭으로 집계합니다."""
    
    METRIC_FUNCTIONS = {
        "accuracy": lambda preds, refs: np.mean([p == r for p, r in zip(preds, refs)]),
        "exact_match": lambda preds, refs: np.mean([p.strip() == r.strip() for p, r in zip(preds, refs)]),
    }
    
    def aggregate(
        self,
        predictions: list[str],
        references: list[str],
        metric_names: list[str],
    ) -> dict[str, float]:
        """지정된 메트릭들을 계산합니다."""
        results = {}
        for name in metric_names:
            func = self.METRIC_FUNCTIONS.get(name)
            if func is None:
                raise ValueError(f"지원하지 않는 메트릭: {name}")
            results[name] = func(predictions, references)
        return results
    
    def compute_confidence_interval(
        self,
        scores: list[float],
        confidence: float = 0.95,
        method: str = "bootstrap",
    ) -> tuple[float, float]:
        """신뢰 구간을 계산합니다."""
        if method == "bootstrap":
            n_bootstrap = 1000
            bootstrap_means = []
            for _ in range(n_bootstrap):
                sample = np.random.choice(scores, size=len(scores), replace=True)
                bootstrap_means.append(np.mean(sample))
            lower = np.percentile(bootstrap_means, (1 - confidence) / 2 * 100)
            upper = np.percentile(bootstrap_means, (1 + confidence) / 2 * 100)
            return (lower, upper)
        
        # Wilson score interval
        n = len(scores)
        mean = np.mean(scores)
        z = stats.norm.ppf((1 + confidence) / 2)
        se = z * np.sqrt(mean * (1 - mean) / n)
        return (mean - se, mean + se)

def run_sequential(tasks: list[TaskConfig], backend: ModelBackend) -> dict:
    results = {}
    for task in tasks:
        dataset = load_dataset(task.dataset_path, task.dataset_name, split=task.dataset_split)
        prompts = build_prompts(task, dataset)
        outputs = backend.generate(prompts)
        metrics = compute_metrics(task.metrics, outputs, dataset["answer"])
        results[task.task_name] = metrics
    return results