본문 바로가기

LLM

LLM 사설 평가셋 50개 만들고 모델 비교하기 — 벤치마크를 믿지 마세요

반응형

48시간마다 새 모델이 나와요. 모두 "SWE-bench 1위", "GPQA 최고점"을 주장해요.

근데 그게 내 서비스에서도 최고일까요.

공개 벤치마크의 현실:
MMLU: 상위 모델 88% 이상 → 이미 포화
SWE-bench: 에이전트 코딩 특화 → 일반 서비스와 무관
GPQA: 박사급 과학 문제 → 실제 업무와 거리 멀어

데이터 오염 문제:
- 훈련 데이터에 이미 벤치마크 문제가 포함됨
- GSM8K 점수 vs GSM1K(새 문제) 점수 차이: 최대 16%p
- 모델이 실제로 못 풀어도 고점 가능

진짜 답은 내 서비스에 맞는 사설 평가셋을 직접 만드는 것이에요.

50개면 충분히 시작할 수 있어요. 오늘 만들어봅시다.


왜 50개인가

너무 적으면 (10개 미만):
→ 통계적으로 의미 없음
→ 우연에 의한 결과 가능

너무 많으면 (500개+):
→ 만드는 시간이 너무 오래 걸림
→ 처음엔 의욕만 앞서다 포기

50개의 현실:
→ 하루 안에 만들 수 있음
→ 모델 간 의미 있는 차이 감지 가능
→ 나중에 100개, 200개로 확장 용이

1단계 — 평가셋 설계

내 서비스 타입 파악

먼저 어떤 케이스를 테스트해야 하는지 정해야 해요.

# 평가 카테고리 템플릿
EVAL_CATEGORIES = {
    # 코딩 서비스라면
    "coding_service": {
        "bug_fix": 10,           # 버그 수정
        "code_review": 10,       # 코드 리뷰
        "refactoring": 10,       # 리팩토링
        "explanation": 10,       # 코드 설명
        "generation": 10,        # 코드 생성
    },

    # RAG/QA 서비스라면
    "rag_service": {
        "factual_qa": 15,        # 사실 기반 질답
        "multi_hop": 10,         # 여러 문서 조합
        "summarization": 10,     # 요약
        "edge_cases": 10,        # 엣지 케이스
        "refusal": 5,            # 거절해야 할 경우
    },

    # 고객 지원 서비스라면
    "support_service": {
        "complaint": 10,         # 불만 처리
        "technical": 15,         # 기술 지원
        "billing": 10,           # 결제 관련
        "escalation": 10,        # 에스컬레이션
        "polite_refusal": 5,     # 정중한 거절
    }
}

50개 구성 예시 — 코딩 서비스 기준

카테고리별 배분:
01~10: 버그 수정 (난이도 쉬움 4, 보통 4, 어려움 2)
11~20: 코드 리뷰 (Python 5, TypeScript 5)
21~30: 리팩토링 (레거시 코드 포함)
31~40: 코드 설명 (복잡한 알고리즘)
41~50: 엣지 케이스 (잘못된 요청, 불가능한 요구)

2단계 — 케이스 수집 방법

방법 A — 프로덕션 로그에서 추출 (가장 좋음)

import json
import random
from pathlib import Path

def extract_from_logs(log_file: str, n: int = 50) -> list:
    """프로덕션 로그에서 실제 사용 케이스 추출"""

    with open(log_file) as f:
        logs = [json.loads(line) for line in f]

    # 실제 사용자 요청만 필터
    real_requests = [
        log for log in logs
        if log.get("type") == "user_message"
        and len(log.get("content", "")) > 20  # 너무 짧은 거 제외
    ]

    # 카테고리별로 균등 샘플링
    sampled = random.sample(real_requests, min(n, len(real_requests)))

    return sampled

방법 B — 직접 작성 (프로덕션 로그 없을 때)

# eval_dataset.jsonl 형식
eval_cases = [
    {
        "id": "code_001",
        "category": "bug_fix",
        "difficulty": "easy",
        "input": """
다음 Python 코드에서 버그를 찾아줘:

def calculate_average(numbers):
    total = 0
    for n in numbers:
        total += n
    return total / len(numbers)

calculate_average([])  # ZeroDivisionError 발생
""",
        "expected_elements": [
            "빈 리스트 체크",
            "ZeroDivisionError 언급",
            "수정된 코드 제공"
        ],
        "reference_answer": """
버그: 빈 리스트 입력 시 ZeroDivisionError 발생

수정 코드:
def calculate_average(numbers):
    if not numbers:
        return 0  # 또는 None, 또는 raise ValueError
    return sum(numbers) / len(numbers)
"""
    },
    {
        "id": "code_002",
        "category": "code_review",
        "difficulty": "medium",
        "input": """
다음 코드를 리뷰해줘:

def get_user(user_id):
    conn = sqlite3.connect('users.db')
    query = f"SELECT * FROM users WHERE id = {user_id}"
    result = conn.execute(query)
    return result.fetchone()
""",
        "expected_elements": [
            "SQL 인젝션 취약점",
            "커넥션 관리 문제",
            "파라미터화 쿼리 제안"
        ],
        "reference_answer": "SQL 인젝션 취약점 + with 구문으로 커넥션 관리"
    },
    # ... 계속
]

 

방법 C — AI로 케이스 생성 (빠른 방법)

import anthropic

client = anthropic.Anthropic()

def generate_eval_cases(
    service_type: str,
    category: str,
    n: int = 10
) -> list:
    """평가 케이스를 AI로 생성"""

    response = client.messages.create(
        model="claude-sonnet-4-6",  # 생성엔 Sonnet으로 절약
        max_tokens=4000,
        messages=[{
            "role": "user",
            "content": f"""
{service_type} 서비스를 위한 LLM 평가 케이스 {n}개를 만들어줘.
카테고리: {category}

각 케이스는 JSON 형식으로:
- id: 고유 ID
- input: 테스트할 프롬프트
- expected_elements: 좋은 답변에 포함되어야 할 요소들 (리스트)
- difficulty: easy/medium/hard

실제 사용자가 보낼 법한 현실적인 케이스로 만들어줘.
JSON 배열로만 응답해줘.
"""
        }]
    )

    import json
    cases = json.loads(response.content[0].text)
    return cases

# 카테고리별 생성
all_cases = []
categories = ["bug_fix", "code_review", "refactoring", "explanation", "edge_case"]

for cat in categories:
    cases = generate_eval_cases("코딩 어시스턴트", cat, n=10)
    all_cases.extend(cases)

# JSONL로 저장
with open("eval_dataset.jsonl", "w") as f:
    for case in all_cases:
        f.write(json.dumps(case, ensure_ascii=False) + "\n")

print(f"총 {len(all_cases)}개 케이스 생성 완료")

3단계 — 평가 방법 선택

방법 1 — LLM-as-a-Judge (빠르고 자동화 가능)

import anthropic
import json

client = anthropic.Anthropic()

JUDGE_PROMPT = """
당신은 LLM 응답 품질을 평가하는 전문가입니다.

[질문]
{input}

[참조 답변]
{reference}

[평가할 응답]
{response}

[포함되어야 할 요소]
{expected_elements}

다음 기준으로 0~10점 평가:
- 정확성 (0~4점): 내용이 맞는가
- 완전성 (0~3점): expected_elements를 얼마나 포함하는가
- 품질 (0~3점): 설명이 명확하고 실용적인가

JSON으로만 응답:
{{"score": 숫자, "reasoning": "이유", "missing": ["빠진 요소들"]}}
"""

def judge_response(
    input_text: str,
    response: str,
    reference: str,
    expected_elements: list
) -> dict:
    """LLM으로 응답 품질 평가"""

    result = client.messages.create(
        model="claude-haiku-4-5",  # 평가는 Haiku로 비용 절감
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": JUDGE_PROMPT.format(
                input=input_text,
                reference=reference,
                response=response,
                expected_elements="\n".join(f"- {e}" for e in expected_elements)
            )
        }]
    )

    return json.loads(result.content[0].text)

방법 2 — 규칙 기반 평가 (객관적, 빠름)

def rule_based_eval(
    response: str,
    expected_elements: list,
    forbidden_elements: list = None
) -> dict:
    """키워드 기반 자동 평가"""

    response_lower = response.lower()
    scores = {}

    # 포함 체크
    found = []
    missing = []
    for element in expected_elements:
        if element.lower() in response_lower:
            found.append(element)
        else:
            missing.append(element)

    completeness = len(found) / len(expected_elements) * 10

    # 금지 요소 체크 (할루시네이션 등)
    violations = []
    if forbidden_elements:
        for elem in forbidden_elements:
            if elem.lower() in response_lower:
                violations.append(elem)

    return {
        "completeness_score": completeness,
        "found_elements": found,
        "missing_elements": missing,
        "violations": violations,
        "pass": completeness >= 7.0 and len(violations) == 0
    }

4단계 — 여러 모델 동시 평가

import asyncio
import anthropic
from openai import AsyncOpenAI

anthropic_client = anthropic.AsyncAnthropic()
openai_client = AsyncOpenAI()

MODELS_TO_TEST = [
    {"provider": "anthropic", "model": "claude-opus-4-7",    "name": "Opus 4.7"},
    {"provider": "anthropic", "model": "claude-sonnet-4-6",  "name": "Sonnet 4.6"},
    {"provider": "openai",    "model": "gpt-5.4",            "name": "GPT-5.4"},
    {"provider": "openai",    "model": "gpt-5.4-mini",       "name": "GPT-5.4 mini"},
]

async def get_response(provider: str, model: str, prompt: str) -> str:
    """모델별 응답 수집"""
    if provider == "anthropic":
        response = await anthropic_client.messages.create(
            model=model,
            max_tokens=2048,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text

    elif provider == "openai":
        response = await openai_client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content

async def run_eval(eval_cases: list) -> dict:
    """전체 평가셋으로 모든 모델 평가"""

    results = {m["name"]: [] for m in MODELS_TO_TEST}

    for i, case in enumerate(eval_cases):
        print(f"케이스 {i+1}/{len(eval_cases)}: {case['id']}")

        # 모든 모델 동시 요청
        tasks = [
            get_response(m["provider"], m["model"], case["input"])
            for m in MODELS_TO_TEST
        ]
        responses = await asyncio.gather(*tasks, return_exceptions=True)

        # 각 응답 평가
        for model_info, response in zip(MODELS_TO_TEST, responses):
            if isinstance(response, Exception):
                score = {"score": 0, "error": str(response)}
            else:
                score = judge_response(
                    input_text=case["input"],
                    response=response,
                    reference=case.get("reference_answer", ""),
                    expected_elements=case.get("expected_elements", [])
                )
                score["response"] = response

            score["case_id"] = case["id"]
            score["category"] = case["category"]
            results[model_info["name"]].append(score)

    return results

# 실행
async def main():
    # 평가셋 로드
    eval_cases = []
    with open("eval_dataset.jsonl") as f:
        for line in f:
            eval_cases.append(json.loads(line))

    print(f"총 {len(eval_cases)}개 케이스로 평가 시작...")
    results = await run_eval(eval_cases)
    return results

results = asyncio.run(main())

5단계 — 결과 분석 및 리포트

import pandas as pd

def analyze_results(results: dict) -> None:
    """평가 결과 분석 및 출력"""

    print("\n" + "="*60)
    print("📊 LLM 평가 결과 리포트")
    print("="*60)

    summary = []

    for model_name, cases in results.items():
        valid_cases = [c for c in cases if "score" in c and "error" not in c]

        if not valid_cases:
            continue

        avg_score = sum(c["score"] for c in valid_cases) / len(valid_cases)

        # 카테고리별 점수
        category_scores = {}
        for case in valid_cases:
            cat = case.get("category", "unknown")
            if cat not in category_scores:
                category_scores[cat] = []
            category_scores[cat].append(case["score"])

        cat_avgs = {
            cat: sum(scores) / len(scores)
            for cat, scores in category_scores.items()
        }

        summary.append({
            "모델": model_name,
            "전체 평균": round(avg_score, 2),
            "평가 케이스": len(valid_cases),
            **{f"{cat}": round(avg, 2) for cat, avg in cat_avgs.items()}
        })

    # 표로 출력
    df = pd.DataFrame(summary).sort_values("전체 평균", ascending=False)
    print(df.to_string(index=False))

    # 1위 모델
    best = df.iloc[0]
    print(f"\n🏆 최고 모델: {best['모델']} (평균 {best['전체 평균']}점)")

    # 카테고리별 강자
    print("\n📌 카테고리별 최고 모델:")
    categories = [col for col in df.columns
                  if col not in ["모델", "전체 평균", "평가 케이스"]]

    for cat in categories:
        if cat in df.columns:
            best_cat = df.loc[df[cat].idxmax(), "모델"]
            best_score = df[cat].max()
            print(f"  {cat}: {best_cat} ({best_score}점)")

analyze_results(results)

출력 예시:

============================================================
📊 LLM 평가 결과 리포트
============================================================
     모델  전체 평균  평가 케이스  bug_fix  code_review  refactoring
 Opus 4.7       8.24          50     8.50         8.10         8.10
  GPT-5.4       7.91          50     7.80         8.20         7.74
Sonnet 4.6      7.65          50     7.60         7.90         7.44
GPT-5.4 mini    6.82          50     6.90         7.10         6.46

🏆 최고 모델: Opus 4.7 (평균 8.24점)

📌 카테고리별 최고 모델:
  bug_fix: Opus 4.7 (8.50점)
  code_review: GPT-5.4 (8.20점)  ← 이거 의외!
  refactoring: Opus 4.7 (8.10점)

6단계 — 평가셋 유지 관리

import hashlib
from datetime import datetime

def add_case(
    case: dict,
    dataset_file: str = "eval_dataset.jsonl"
) -> None:
    """새 케이스 추가 (중복 방지)"""

    # 기존 케이스 로드
    existing_ids = set()
    try:
        with open(dataset_file) as f:
            for line in f:
                existing_ids.add(json.loads(line)["id"])
    except FileNotFoundError:
        pass

    # 새 케이스에 메타데이터 추가
    case["created_at"] = datetime.now().isoformat()
    case["id"] = case.get("id") or hashlib.md5(
        case["input"].encode()
    ).hexdigest()[:8]

    if case["id"] in existing_ids:
        print(f"중복 케이스 스킵: {case['id']}")
        return

    with open(dataset_file, "a") as f:
        f.write(json.dumps(case, ensure_ascii=False) + "\n")

    print(f"케이스 추가: {case['id']}")


def rotate_dataset(
    dataset_file: str,
    max_age_days: int = 180
) -> None:
    """6개월 넘은 케이스 제거 (오버피팅 방지)"""

    cutoff = datetime.now().timestamp() - (max_age_days * 86400)

    fresh_cases = []
    with open(dataset_file) as f:
        for line in f:
            case = json.loads(line)
            created = datetime.fromisoformat(
                case.get("created_at", datetime.now().isoformat())
            ).timestamp()

            if created > cutoff:
                fresh_cases.append(case)

    with open(dataset_file, "w") as f:
        for case in fresh_cases:
            f.write(json.dumps(case, ensure_ascii=False) + "\n")

    print(f"케이스 정리 완료: {len(fresh_cases)}개 유지")

전체 실행 스크립트

# 1. 평가셋 생성
python generate_eval.py --service coding --n 50

# 2. 모델 비교 실행
python run_eval.py --dataset eval_dataset.jsonl

# 3. 결과 분석
python analyze_results.py --results results.json

# 4. 리포트 출력
# → 어떤 모델이 내 서비스에 맞는지 데이터로 확인

 

반응형