7장: 커스텀 평가 하네스 설계와 구축

from dataclasses import dataclass
from enum import Enum
 
 
class EvalDimension(Enum):
    ACCURACY = "accuracy"           # 정보의 정확성
    COMPLETENESS = "completeness"   # 응답의 완전성
    TONE = "tone"                   # 톤앤매너 준수
    SAFETY = "safety"               # 안전성 (개인정보 노출 방지 등)
    ACTIONABILITY = "actionability" # 실행 가능한 안내 제공 여부
 
 
@dataclass
class SupportEvalCase:
    """고객 상담 평가 케이스."""
    
    case_id: str
    category: str                   # 문의 유형 (배송, 환불, 기술지원 등)
    customer_query: str             # 고객 질문
    context: list[str]              # 참조 문서/정책
    expected_answer: str            # 모범 답변
    required_elements: list[str]    # 반드시 포함해야 할 요소
    forbidden_elements: list[str]   # 포함되면 안 되는 요소
    eval_dimensions: list[EvalDimension]  # 평가할 차원
 
 
# 평가 케이스 정의
SUPPORT_EVAL_CASES = [
    SupportEvalCase(
        case_id="REFUND-001",
        category="환불",
        customer_query="주문한 지 2주가 넘었는데 상품이 오지 않습니다. 환불 받을 수 있나요?",
        context=[
            "환불 정책: 배송 지연 14일 초과 시 전액 환불 가능",
            "환불 처리 기간: 요청 후 3-5 영업일",
            "환불 방법: 원결제 수단으로 환불",
        ],
        expected_answer="14일 이상 배송 지연 시 전액 환불 가능하며, 환불 요청 시 3-5 영업일 내 원결제 수단으로 환불됩니다.",
        required_elements=["전액 환불", "3-5 영업일", "원결제 수단"],
        forbidden_elements=["확실하지 않", "모르겠", "다른 부서"],
        eval_dimensions=[
            EvalDimension.ACCURACY,
            EvalDimension.COMPLETENESS,
            EvalDimension.TONE,
            EvalDimension.ACTIONABILITY,
        ],
    ),
]

from dataclasses import dataclass
import re
 
 
@dataclass
class MetricResult:
    name: str
    score: float        # 0.0 - 1.0
    passed: bool
    details: str
 
 
class RuleBasedMetrics:
    """규칙 기반 평가 메트릭 모음."""
    
    @staticmethod
    def required_elements(response: str, elements: list[str]) -> MetricResult:
        """필수 요소 포함 여부를 검사합니다."""
        found = [elem for elem in elements if elem in response]
        score = len(found) / len(elements) if elements else 1.0
        missing = [elem for elem in elements if elem not in response]
        
        return MetricResult(
            name="required_elements",
            score=score,
            passed=score >= 1.0,
            details=f"포함: {found}, 누락: {missing}",
        )
    
    @staticmethod
    def forbidden_elements(response: str, elements: list[str]) -> MetricResult:
        """금지 요소 미포함 여부를 검사합니다."""
        violations = [elem for elem in elements if elem in response]
        score = 1.0 - (len(violations) / len(elements)) if elements else 1.0
        
        return MetricResult(
            name="forbidden_elements",
            score=score,
            passed=len(violations) == 0,
            details=f"위반: {violations}" if violations else "위반 없음",
        )
    
    @staticmethod
    def response_length(
        response: str,
        min_chars: int = 50,
        max_chars: int = 500,
    ) -> MetricResult:
        """응답 길이가 기준 범위 내인지 검사합니다."""
        length = len(response)
        in_range = min_chars <= length <= max_chars
        
        return MetricResult(
            name="response_length",
            score=1.0 if in_range else 0.0,
            passed=in_range,
            details=f"길이: {length}자 (기준: {min_chars}-{max_chars})",
        )
    
    @staticmethod
    def no_pii_leak(response: str) -> MetricResult:
        """개인정보 노출 여부를 검사합니다."""
        patterns = {
            "전화번호": r"01[016789]-?\d{3,4}-?\d{4}",
            "이메일": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
            "주민번호": r"\d{6}-?[1-4]\d{6}",
        }
        
        violations = []
        for name, pattern in patterns.items():
            if re.search(pattern, response):
                violations.append(name)
        
        return MetricResult(
            name="no_pii_leak",
            score=1.0 if not violations else 0.0,
            passed=len(violations) == 0,
            details=f"PII 감지: {violations}" if violations else "PII 미감지",
        )

from openai import OpenAI
 
 
class LLMJudge:
    """LLM을 사용한 평가 메트릭."""
    
    def __init__(self, model: str = "gpt-4o", api_key: str | None = None):
        self.client = OpenAI(api_key=api_key)
        self.model = model
    
    def evaluate(
        self,
        query: str,
        response: str,
        criteria: str,
        reference: str | None = None,
    ) -> MetricResult:
        """주어진 기준에 따라 응답을 평가합니다."""
        
        system_prompt = """당신은 AI 응답의 품질을 평가하는 전문 심판입니다.
주어진 기준에 따라 응답을 1-5점으로 평가하세요.
 
평가 결과를 다음 형식으로 출력하세요:
점수: [1-5]
근거: [평가 근거를 2-3문장으로 작성]"""
 
        user_prompt = f"""## 질문
{query}
 
## 응답
{response}
 
## 평가 기준
{criteria}"""
 
        if reference:
            user_prompt += f"\n\n## 참고 모범 답변\n{reference}"
        
        completion = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt},
            ],
            temperature=0.0,
        )
        
        result_text = completion.choices[0].message.content or ""
        score = self._parse_score(result_text)
        
        return MetricResult(
            name=f"llm_judge_{criteria[:20]}",
            score=score / 5.0,  # 0-1 스케일로 정규화
            passed=score >= 3,
            details=result_text,
        )
    
    @staticmethod
    def _parse_score(text: str) -> int:
        """LLM 출력에서 점수를 추출합니다."""
        import re
        match = re.search(r"점수:\s*(\d)", text)
        return int(match.group(1)) if match else 3

from dataclasses import dataclass
from enum import Enum
 
 
class ReviewStatus(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"
 
 
@dataclass
class HumanReviewItem:
    case_id: str
    query: str
    response: str
    auto_score: float
    auto_metrics: dict
    status: ReviewStatus = ReviewStatus.PENDING
    reviewer: str | None = None
    human_score: float | None = None
    comments: str | None = None
 
 
class HumanEvalQueue:
    """인간 평가 큐를 관리합니다."""
    
    def __init__(self, auto_pass_threshold: float = 0.9, auto_fail_threshold: float = 0.3):
        self.queue: list[HumanReviewItem] = []
        self.auto_pass_threshold = auto_pass_threshold
        self.auto_fail_threshold = auto_fail_threshold
    
    def triage(self, case_id: str, query: str, response: str, auto_score: float, auto_metrics: dict) -> str:
        """자동 평가 결과를 기반으로 인간 평가 필요 여부를 결정합니다."""
        
        if auto_score >= self.auto_pass_threshold:
            return "auto_pass"
        
        if auto_score <= self.auto_fail_threshold:
            return "auto_fail"
        
        # 경계 영역: 인간 평가 필요
        self.queue.append(HumanReviewItem(
            case_id=case_id,
            query=query,
            response=response,
            auto_score=auto_score,
            auto_metrics=auto_metrics,
        ))
        return "human_review"
    
    def compute_agreement(self) -> float:
        """인간 평가와 자동 평가의 일치율을 계산합니다."""
        reviewed = [item for item in self.queue if item.human_score is not None]
        if not reviewed:
            return 0.0
        
        agreements = sum(
            1 for item in reviewed
            if (item.auto_score >= 0.5) == (item.human_score >= 0.5)
        )
        return agreements / len(reviewed)

import json
import hashlib
from datetime import datetime
from pathlib import Path
 
 
class GoldenDatasetManager:
    """버전 관리되는 골든 데이터셋을 관리합니다."""
    
    def __init__(self, base_dir: str):
        self.base_dir = Path(base_dir)
        self.base_dir.mkdir(parents=True, exist_ok=True)
    
    def save_version(self, dataset: list[dict], version: str, description: str) -> str:
        """데이터셋의 새 버전을 저장합니다."""
        
        # 데이터 해시 계산 (무결성 검증용)
        data_hash = hashlib.sha256(
            json.dumps(dataset, sort_keys=True, ensure_ascii=False).encode()
        ).hexdigest()[:12]
        
        metadata = {
            "version": version,
            "description": description,
            "created_at": datetime.now().isoformat(),
            "num_samples": len(dataset),
            "data_hash": data_hash,
        }
        
        version_dir = self.base_dir / version
        version_dir.mkdir(exist_ok=True)
        
        with open(version_dir / "data.json", "w", encoding="utf-8") as f:
            json.dump(dataset, f, ensure_ascii=False, indent=2)
        
        with open(version_dir / "metadata.json", "w", encoding="utf-8") as f:
            json.dump(metadata, f, ensure_ascii=False, indent=2)
        
        return data_hash
    
    def load_version(self, version: str) -> list[dict]:
        """특정 버전의 데이터셋을 로드합니다."""
        data_path = self.base_dir / version / "data.json"
        with open(data_path, encoding="utf-8") as f:
            return json.load(f)
    
    def list_versions(self) -> list[dict]:
        """모든 버전의 메타데이터를 반환합니다."""
        versions = []
        for version_dir in sorted(self.base_dir.iterdir()):
            if version_dir.is_dir():
                meta_path = version_dir / "metadata.json"
                if meta_path.exists():
                    with open(meta_path) as f:
                        versions.append(json.load(f))
        return versions

import asyncio
import json
from dataclasses import dataclass, field, asdict
from datetime import datetime
from pathlib import Path
 
 
@dataclass
class EvalConfig:
    """평가 실행 설정."""
    model_name: str
    tasks: list[str]
    output_dir: str = "./eval_results"
    max_concurrent: int = 5
 
 
@dataclass
class EvalReport:
    """평가 결과 리포트."""
    model_name: str
    timestamp: str
    task_results: dict = field(default_factory=dict)
    summary: dict = field(default_factory=dict)
 
 
class MiniEvalHarness:
    """도메인 특화 미니 평가 하네스."""
    
    def __init__(self, config: EvalConfig):
        self.config = config
        self.rule_metrics = RuleBasedMetrics()
        self.llm_judge = LLMJudge()
        self.semaphore = asyncio.Semaphore(config.max_concurrent)
    
    async def run(self, eval_cases: list[SupportEvalCase], model_fn) -> EvalReport:
        """전체 평가를 실행합니다."""
        
        report = EvalReport(
            model_name=self.config.model_name,
            timestamp=datetime.now().isoformat(),
        )
        
        # 모든 케이스를 병렬로 평가
        tasks = [
            self._evaluate_case(case, model_fn)
            for case in eval_cases
        ]
        results = await asyncio.gather(*tasks)
        
        # 결과 집계
        for case, result in zip(eval_cases, results):
            report.task_results[case.case_id] = result
        
        report.summary = self._compute_summary(results)
        
        # 결과 저장
        self._save_report(report)
        
        return report
    
    async def _evaluate_case(self, case: SupportEvalCase, model_fn) -> dict:
        """단일 케이스를 평가합니다."""
        async with self.semaphore:
            # 모델 응답 생성
            response = await model_fn(
                query=case.customer_query,
                context=case.context,
            )
            
            # 규칙 기반 메트릭
            metrics = {}
            metrics["required_elements"] = asdict(
                self.rule_metrics.required_elements(response, case.required_elements)
            )
            metrics["forbidden_elements"] = asdict(
                self.rule_metrics.forbidden_elements(response, case.forbidden_elements)
            )
            metrics["no_pii_leak"] = asdict(
                self.rule_metrics.no_pii_leak(response)
            )
            
            # LLM-as-Judge 메트릭
            if EvalDimension.TONE in case.eval_dimensions:
                metrics["tone"] = asdict(
                    self.llm_judge.evaluate(
                        query=case.customer_query,
                        response=response,
                        criteria="응답이 친절하고 전문적인 톤을 유지하는지 평가",
                        reference=case.expected_answer,
                    )
                )
            
            return {
                "response": response,
                "metrics": metrics,
                "overall_pass": all(
                    m.get("passed", True) for m in metrics.values()
                ),
            }
    
    def _compute_summary(self, results: list[dict]) -> dict:
        """전체 결과를 요약합니다."""
        total = len(results)
        passed = sum(1 for r in results if r["overall_pass"])
        
        return {
            "total_cases": total,
            "passed": passed,
            "failed": total - passed,
            "pass_rate": passed / total if total > 0 else 0,
        }
    
    def _save_report(self, report: EvalReport) -> None:
        """결과를 JSON 파일로 저장합니다."""
        output_dir = Path(self.config.output_dir)
        output_dir.mkdir(parents=True, exist_ok=True)
        
        filename = f"eval_{report.model_name}_{report.timestamp.replace(':', '-')}.json"
        with open(output_dir / filename, "w", encoding="utf-8") as f:
            json.dump(asdict(report), f, ensure_ascii=False, indent=2)