5장: 인간 평가와 어노테이션 설계

평가 방법별 역할:
 
자동 메트릭      --> 대규모 스크리닝, CI/CD 게이트
LLM-as-Judge    --> 중간 규모 품질 평가, 프롬프트 비교
인간 평가        --> 최종 품질 검증, 자동 메트릭 캘리브레이션, 문제 발견

# 절대 평가 가이드라인 구조
absolute_rating_schema = {
    "task": "응답 품질 평가",
    "scale": {
        1: "매우 나쁨 - 질문과 무관하거나, 완전히 부정확한 정보를 포함",
        2: "나쁨 - 부분적으로 관련되나, 주요 오류 또는 중요한 누락이 있음",
        3: "보통 - 질문에 답변하지만, 부정확성이나 불완전함이 있음",
        4: "좋음 - 정확하고 관련성 있는 답변이지만, 개선의 여지가 있음",
        5: "매우 좋음 - 정확하고 완전하며, 잘 구성된 답변",
    },
    "dimensions": [
        "정확성 - 사실적으로 올바른가",
        "완전성 - 질문의 모든 측면을 다루는가",
        "명확성 - 이해하기 쉬운가",
        "유용성 - 질문자에게 실질적 도움이 되는가",
    ]
}

# 상대 평가 가이드라인 구조
comparative_rating_schema = {
    "task": "두 응답 비교 평가",
    "options": [
        "A가 확실히 더 나음",
        "A가 약간 더 나음",
        "비슷함 (동점)",
        "B가 약간 더 나음",
        "B가 확실히 더 나음",
    ],
    "instruction": "두 응답을 전반적 품질 관점에서 비교하세요. "
                   "정확성, 완전성, 명확성을 종합적으로 고려하세요."
}

효과적인 어노테이션 가이드라인 구성:
 
1. 과제 설명
   - 평가의 목적과 맥락
   - 대상 시스템에 대한 배경 정보
 
2. 평가 기준 정의
   - 각 평가 차원의 명확한 정의
   - 점수 척도와 각 수준의 상세 설명
 
3. 예시 (Anchor Examples)
   - 각 점수 수준에 해당하는 구체적 예시
   - 경계 사례(Borderline Case)에 대한 판단 지침
 
4. 주의사항
   - 흔한 실수와 피해야 할 편향
   - 애매한 상황에서의 처리 방법
 
5. 프로세스 설명
   - 평가 도구 사용법
   - 작업 흐름과 소요 시간 안내

anchor_examples = {
    "dimension": "정확성",
    "examples": [
        {
            "score": 5,
            "question": "파이썬의 GIL이란 무엇인가요?",
            "answer": "GIL(Global Interpreter Lock)은 CPython 인터프리터에서 "
                      "한 번에 하나의 스레드만 파이썬 바이트코드를 실행하도록 "
                      "제한하는 뮤텍스입니다. 멀티스레딩 환경에서 메모리 관리의 "
                      "스레드 안전성을 보장하기 위해 도입되었습니다. "
                      "CPU 바운드 작업에서는 병렬 처리가 제한되지만, "
                      "I/O 바운드 작업에서는 GIL이 해제되므로 "
                      "멀티스레딩의 이점을 얻을 수 있습니다.",
            "rationale": "GIL의 정의, 존재 이유, 영향 범위를 정확하게 설명함",
        },
        {
            "score": 3,
            "question": "파이썬의 GIL이란 무엇인가요?",
            "answer": "GIL은 파이썬에서 멀티스레딩을 제한하는 잠금 장치입니다. "
                      "이 때문에 파이썬은 여러 스레드를 동시에 사용할 수 없습니다.",
            "rationale": "기본 개념은 맞지만, CPython 한정이라는 점을 빠뜨렸고, "
                         "I/O 바운드에서의 동작을 설명하지 않아 불완전하며 "
                         "다소 오해를 줄 수 있는 표현이 포함됨",
        },
        {
            "score": 1,
            "question": "파이썬의 GIL이란 무엇인가요?",
            "answer": "GIL은 파이썬의 가비지 컬렉션 시스템입니다. "
                      "메모리를 자동으로 관리하여 프로그래머가 "
                      "메모리 할당과 해제를 신경 쓰지 않아도 됩니다.",
            "rationale": "GIL을 가비지 컬렉션과 혼동하여 "
                         "근본적으로 부정확한 정보를 제공함",
        },
    ]
}

from sklearn.metrics import cohen_kappa_score
 
def compute_agreement(
    annotator_a: list,
    annotator_b: list
) -> dict:
    """두 평가자 간의 일치도를 계산합니다."""
    # 정확 일치율
    raw_agreement = sum(
        1 for a, b in zip(annotator_a, annotator_b) if a == b
    ) / len(annotator_a)
 
    # Cohen's Kappa
    kappa = cohen_kappa_score(annotator_a, annotator_b)
 
    # 가중 Kappa (순서형 척도에 적합)
    weighted_kappa = cohen_kappa_score(
        annotator_a, annotator_b, weights="quadratic"
    )
 
    return {
        "raw_agreement": round(raw_agreement, 3),
        "cohen_kappa": round(kappa, 3),
        "weighted_kappa": round(weighted_kappa, 3),
    }

def fleiss_kappa(ratings_matrix) -> float:
    """Fleiss' Kappa를 계산합니다.
 
    ratings_matrix: N x K 행렬
    N = 평가 대상 수, K = 카테고리 수
    각 셀은 해당 카테고리를 선택한 평가자 수
    """
    import numpy as np
    matrix = np.array(ratings_matrix)
    N, K = matrix.shape
    n = matrix.sum(axis=1)[0]  # 각 항목의 평가자 수
 
    # 항목별 일치도
    p_i = (np.sum(matrix ** 2, axis=1) - n) / (n * (n - 1))
    P_bar = np.mean(p_i)
 
    # 우연 일치도
    p_j = np.sum(matrix, axis=0) / (N * n)
    P_e = np.sum(p_j ** 2)
 
    if P_e == 1:
        return 1.0
 
    kappa = (P_bar - P_e) / (1 - P_e)
    return round(float(kappa), 3)

def run_pilot_evaluation(
    sample_cases: list,
    annotators: list,
    guideline: dict
) -> dict:
    """파일럿 평가를 실시하고 결과를 분석합니다."""
    # 1. 소규모 샘플 준비 (20-50건)
    pilot_size = min(50, len(sample_cases))
    pilot_cases = sample_cases[:pilot_size]
 
    # 2. 모든 평가자가 동일 케이스 평가 (중복 평가)
    all_ratings = {}
    for annotator in annotators:
        ratings = annotator.evaluate(pilot_cases, guideline)
        all_ratings[annotator.id] = ratings
 
    # 3. 일치도 분석
    agreement = compute_pairwise_agreement(all_ratings)
 
    # 4. 불일치 케이스 분석
    disagreements = find_disagreements(all_ratings, threshold=2)
 
    # 5. 가이드라인 개선점 도출
    improvements = analyze_disagreement_patterns(disagreements)
 
    return {
        "agreement_scores": agreement,
        "disagreement_cases": disagreements,
        "suggested_improvements": improvements,
        "ready_for_main": agreement["average_kappa"] >= 0.6,
    }

캘리브레이션 세션 프로세스:
 
1. 가이드라인 설명 (30분)
   - 평가 목적, 기준, 척도 설명
   - 앵커 예시 검토
 
2. 독립 평가 실습 (30분)
   - 10-20건의 예시를 각자 평가
   - 다른 평가자의 결과를 보지 않고 독립적으로 수행
 
3. 결과 비교 토론 (30분)
   - 불일치 케이스에 대해 토론
   - 각자의 판단 근거를 공유
   - 합의점 도출
 
4. 가이드라인 업데이트 (15분)
   - 토론 결과를 가이드라인에 반영
   - 새로운 앵커 예시 추가
 
5. 재평가 확인 (15분)
   - 추가 예시로 일치도 확인

def stratified_sample(
    data: list,
    sample_size: int,
    stratify_key: str,
    priority_filter: str = None
) -> list:
    """계층화 샘플링으로 대표성 있는 평가 샘플을 추출합니다."""
    from collections import defaultdict
    import random
 
    # 계층별 분류
    strata = defaultdict(list)
    for item in data:
        key = item.get("metadata", {}).get(stratify_key, "unknown")
        strata[key].append(item)
 
    # 우선 평가 대상 필터링
    if priority_filter:
        priority_items = [
            item for item in data
            if item.get("auto_eval", {}).get("flag") == priority_filter
        ]
        # 우선 대상에서 50% 할당
        priority_sample = random.sample(
            priority_items,
            min(sample_size // 2, len(priority_items))
        )
    else:
        priority_sample = []
 
    # 나머지는 계층별 비례 배분
    remaining = sample_size - len(priority_sample)
    proportional_sample = []
    total = sum(len(v) for v in strata.values())
 
    for key, items in strata.items():
        n = max(1, int(remaining * len(items) / total))
        proportional_sample.extend(random.sample(items, min(n, len(items))))
 
    return priority_sample + proportional_sample[:remaining]

class HybridEvaluator:
    """인간 평가와 LLM 평가를 결합한 하이브리드 평가 시스템입니다."""
 
    def __init__(self, llm_judge, human_budget: int):
        self.llm_judge = llm_judge
        self.human_budget = human_budget
 
    async def evaluate(self, cases: list) -> list:
        """하이브리드 평가를 실행합니다."""
        # 1단계: 전체를 LLM으로 평가
        llm_results = await self.llm_judge.batch_evaluate(cases)
 
        # 2단계: 인간 평가 대상 선별
        human_candidates = self._select_for_human_review(llm_results)
 
        # 3단계: 선별된 케이스를 인간이 평가
        human_results = await self._request_human_evaluation(
            human_candidates[:self.human_budget]
        )
 
        # 4단계: 결과 통합
        return self._merge_results(llm_results, human_results)
 
    def _select_for_human_review(self, llm_results: list) -> list:
        """인간 리뷰가 필요한 케이스를 선별합니다."""
        candidates = []
        for result in llm_results:
            overall = result.get("overall", 0)
            confidence = result.get("confidence", 1.0)
 
            # 경계선 점수 (2.5-3.5)
            if 2.5 <= overall <= 3.5:
                candidates.append(("borderline", result))
            # 낮은 확신도
            elif confidence < 0.6:
                candidates.append(("low_confidence", result))
            # 기준 차원 간 큰 편차
            elif self._has_high_variance(result):
                candidates.append(("high_variance", result))
 
        # 우선순위에 따라 정렬
        priority = {"low_confidence": 0, "borderline": 1, "high_variance": 2}
        candidates.sort(key=lambda x: priority.get(x[0], 3))
        return [c[1] for c in candidates]
 
    def _has_high_variance(self, result: dict) -> bool:
        scores = list(result.get("scores", {}).values())
        if len(scores) < 2:
            return False
        return max(scores) - min(scores) >= 3

def inject_gold_standards(
    eval_cases: list,
    gold_cases: list,
    injection_rate: float = 0.1
) -> list:
    """평가 케이스에 골드 스탠다드를 삽입합니다."""
    import random
    import math
 
    num_gold = max(1, math.ceil(len(eval_cases) * injection_rate))
    selected_gold = random.sample(gold_cases, min(num_gold, len(gold_cases)))
 
    # 골드 케이스에 표식 (평가자에게는 보이지 않음)
    for g in selected_gold:
        g["_is_gold"] = True
        g["_expected_score"] = g.pop("known_score")
 
    # 랜덤 위치에 삽입
    combined = eval_cases.copy()
    for gold in selected_gold:
        pos = random.randint(0, len(combined))
        combined.insert(pos, gold)
 
    return combined
 
def check_annotator_quality(
    annotations: list,
    tolerance: float = 1.0
) -> dict:
    """골드 스탠다드 결과로 평가자 품질을 확인합니다."""
    gold_items = [a for a in annotations if a.get("_is_gold")]
    if not gold_items:
        return {"status": "no_gold_items"}
 
    correct = 0
    for item in gold_items:
        expected = item["_expected_score"]
        actual = item["annotated_score"]
        if abs(expected - actual) <= tolerance:
            correct += 1
 
    accuracy = correct / len(gold_items)
    return {
        "gold_accuracy": round(accuracy, 3),
        "total_gold": len(gold_items),
        "correct": correct,
        "acceptable": accuracy >= 0.8,
    }