LLM 평가(Evaluation) 체계 가이드 - 벤치마크부터 자체 평가 구축까지

import anthropic
 
client = anthropic.Anthropic()
 
def llm_judge(question: str, response: str, criteria: list) -> dict:
    """LLM을 평가자로 활용"""
    eval_prompt = f"""다음 응답을 평가하세요.
 
## 질문
{question}
 
## 응답
{response}
 
## 평가 기준 (각 1~5점)
{chr(10).join(f"- {c}" for c in criteria)}
 
## 채점 규칙
- 1점: 매우 부족
- 2점: 부족
- 3점: 보통
- 4점: 우수
- 5점: 매우 우수
 
JSON 형식으로 반환:
{{"scores": {{"기준명": 점수}}, "average": 평균, "reasoning": "평가 근거", "improvements": "개선 사항"}}
"""
    
    result = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        temperature=0.0,
        messages=[{"role": "user", "content": eval_prompt}]
    )
    return json.loads(result.content[0].text)
 
# 사용
scores = llm_judge(
    question="Kafka에서 컨슈머 랙이 증가하는 원인은?",
    response="...(평가 대상 응답)...",
    criteria=["정확성", "완전성", "실용성", "코드 예시 품질"]
)

def pairwise_compare(question: str, response_a: str, response_b: str) -> str:
    """두 응답 중 더 나은 것을 선택"""
    prompt = f"""두 응답을 비교하여 더 나은 것을 선택하세요.
 
질문: {question}
 
응답 A: {response_a}
 
응답 B: {response_b}
 
더 나은 응답을 선택하고 이유를 설명하세요.
반환 형식: {{"winner": "A" 또는 "B" 또는 "tie", "reason": "이유"}}
"""
    result = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=512,
        temperature=0.0,
        messages=[{"role": "user", "content": prompt}]
    )
    return json.loads(result.content[0].text)

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    context_utilization
)
from datasets import Dataset
 
# 평가 데이터 구성
eval_data = {
    "question": [
        "Spark에서 셔플 파티션을 조정하는 방법은?",
        "Kafka 컨슈머 그룹의 리밸런싱이란?"
    ],
    "answer": [
        "spark.sql.shuffle.partitions 설정을 변경합니다...",
        "컨슈머 그룹 내에서 파티션 할당을 재조정하는 과정입니다..."
    ],
    "contexts": [
        ["Spark 셔플 파티션은 spark.sql.shuffle.partitions로 설정..."],
        ["Kafka 리밸런싱은 컨슈머 그룹의 멤버 변경 시 발생..."]
    ],
    "ground_truth": [
        "spark.sql.shuffle.partitions 설정으로 셔플 파티션 수를 조정합니다.",
        "리밸런싱은 컨슈머 그룹 내 파티션 재할당 과정입니다."
    ]
}
 
dataset = Dataset.from_dict(eval_data)
 
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)
print(results)
# {'faithfulness': 0.92, 'answer_relevancy': 0.88, 
#  'context_precision': 0.85, 'context_recall': 0.90}

# 평가 데이터셋 구조
eval_dataset = [
    {
        "id": "eval-001",
        "category": "troubleshooting",
        "difficulty": "medium",
        "question": "Spark executor OOM 에러가 발생합니다. 원인과 해결 방법은?",
        "reference_answer": "executor.memory를 늘리거나, 파티션 수를 증가시키거나...",
        "required_elements": ["executor.memory", "파티션", "데이터 스큐"],
        "evaluation_criteria": {
            "accuracy": "기술적으로 정확한 원인 분석",
            "completeness": "주요 원인 3가지 이상 포함",
            "actionability": "구체적 설정값 또는 코드 포함",
            "format": "구조화된 형식 (번호 목록 등)"
        }
    }
]

class EvaluationPipeline:
    def __init__(self, judge_model: str = "claude-sonnet-4-6"):
        self.judge = anthropic.Anthropic()
        self.judge_model = judge_model
        self.results = []
    
    def evaluate_batch(self, model_fn, eval_dataset: list) -> dict:
        """배치 평가 실행"""
        for item in eval_dataset:
            # 1. 모델 응답 생성
            response = model_fn(item["question"])
            
            # 2. 자동 지표 평가
            auto_scores = self.auto_evaluate(item, response)
            
            # 3. LLM-as-Judge 평가
            judge_scores = self.judge_evaluate(item, response)
            
            self.results.append({
                "id": item["id"],
                "category": item["category"],
                "auto_scores": auto_scores,
                "judge_scores": judge_scores,
                "response": response
            })
        
        return self.aggregate_results()
    
    def auto_evaluate(self, item: dict, response: str) -> dict:
        """규칙 기반 자동 평가"""
        scores = {}
        # 필수 요소 포함 여부
        required = item.get("required_elements", [])
        found = sum(1 for e in required if e.lower() in response.lower())
        scores["element_coverage"] = found / max(len(required), 1)
        
        # 응답 길이 적정성
        word_count = len(response.split())
        scores["length_appropriate"] = 1.0 if 50 < word_count < 500 else 0.5
        
        return scores
    
    def aggregate_results(self) -> dict:
        """결과 집계"""
        categories = {}
        for r in self.results:
            cat = r["category"]
            if cat not in categories:
                categories[cat] = []
            categories[cat].append(r["judge_scores"]["average"])
        
        return {
            "overall_average": sum(r["judge_scores"]["average"] for r in self.results) / len(self.results),
            "by_category": {k: sum(v)/len(v) for k, v in categories.items()},
            "total_evaluated": len(self.results)
        }

import random
from datetime import datetime
 
class ABTestManager:
    def __init__(self):
        self.experiments = {}
    
    def create_experiment(self, name: str, variants: dict, traffic_split: dict):
        """A/B 테스트 실험 생성"""
        self.experiments[name] = {
            "variants": variants,  # {"A": model_a, "B": model_b}
            "traffic_split": traffic_split,  # {"A": 0.5, "B": 0.5}
            "results": {"A": [], "B": []},
            "created_at": datetime.now()
        }
    
    def get_variant(self, experiment_name: str, user_id: str) -> str:
        """사용자별 일관된 변형 할당"""
        # 사용자 ID 기반 결정적 할당 (같은 사용자는 항상 같은 변형)
        hash_val = hash(f"{experiment_name}:{user_id}") % 100
        exp = self.experiments[experiment_name]
        cumulative = 0
        for variant, ratio in exp["traffic_split"].items():
            cumulative += ratio * 100
            if hash_val < cumulative:
                return variant
        return list(exp["variants"].keys())[-1]
    
    def record_feedback(self, experiment_name: str, variant: str, score: float):
        """사용자 피드백 기록"""
        self.experiments[experiment_name]["results"][variant].append(score)
    
    def get_statistics(self, experiment_name: str) -> dict:
        """통계적 유의성 분석"""
        exp = self.experiments[experiment_name]
        stats = {}
        for variant, scores in exp["results"].items():
            if scores:
                stats[variant] = {
                    "mean": sum(scores) / len(scores),
                    "count": len(scores),
                    "positive_rate": sum(1 for s in scores if s > 0) / len(scores)
                }
        return stats