3장: 데이터 품질 관리와 전처리 파이프라인

전처리 파이프라인 전체 흐름:
 
  원시 데이터 수집
       |
  1. 형식 표준화 (Format Normalization)
       |
  2. 데이터 정제 (Cleaning)
       |
  3. 중복 제거 (Deduplication)
       |
  4. 품질 필터링 (Quality Filtering)
       |
  5. 개인 정보 제거 (PII Removal)
       |
  6. 토큰화 및 패킹 (Tokenization & Packing)
       |
  최종 학습 데이터셋

import json
from typing import TypedDict
 
 
class Message(TypedDict):
    role: str
    content: str
 
 
class TrainingExample(TypedDict):
    messages: list[Message]
    source: str
    quality_score: float
 
 
def normalize_alpaca_format(data: dict) -> TrainingExample:
    """Alpaca 형식을 대화 형식으로 변환"""
    messages: list[Message] = []
 
    if data.get("system"):
        messages.append({
            "role": "system",
            "content": data["system"]
        })
 
    user_content = data["instruction"]
    if data.get("input"):
        user_content = user_content + "\n\n" + data["input"]
 
    messages.append({"role": "user", "content": user_content})
    messages.append({"role": "assistant", "content": data["output"]})
 
    return {
        "messages": messages,
        "source": "alpaca",
        "quality_score": 0.0
    }
 
 
def normalize_sharegpt_format(data: dict) -> TrainingExample:
    """ShareGPT 형식을 표준 대화 형식으로 변환"""
    role_map = {
        "human": "user",
        "gpt": "assistant",
        "system": "system"
    }
    messages: list[Message] = []
    for turn in data["conversations"]:
        role = role_map.get(turn["from"], turn["from"])
        messages.append({
            "role": role,
            "content": turn["value"]
        })
 
    return {
        "messages": messages,
        "source": "sharegpt",
        "quality_score": 0.0
    }

import re
import unicodedata
 
 
def clean_text(text: str) -> str:
    """텍스트 정제 파이프라인"""
 
    # 1. 유니코드 정규화 (NFC)
    text = unicodedata.normalize("NFC", text)
 
    # 2. 불필요한 HTML 태그 제거
    text = re.sub(r"<[^>]+>", "", text)
 
    # 3. 연속 공백 정리
    text = re.sub(r" +", " ", text)
 
    # 4. 연속 줄바꿈 정리 (3개 이상을 2개로)
    text = re.sub(r"\n{3,}", "\n\n", text)
 
    # 5. 제어 문자 제거 (줄바꿈, 탭은 유지)
    text = "".join(
        ch for ch in text
        if ch in ("\n", "\t") or not unicodedata.category(ch).startswith("C")
    )
 
    # 6. 앞뒤 공백 제거
    text = text.strip()
 
    return text
 
 
def clean_example(example: TrainingExample) -> TrainingExample:
    """학습 예제의 모든 메시지를 정제"""
    cleaned_messages = []
    for msg in example["messages"]:
        cleaned_messages.append({
            "role": msg["role"],
            "content": clean_text(msg["content"])
        })
    example["messages"] = cleaned_messages
    return example

def clean_korean_text(text: str) -> str:
    """한국어 텍스트 전용 정제"""
 
    # 1. 자모 분리 오류 수정 (예: ㅎㅏㄴㄱㅜㄱㅇㅓ -> 한국어)
    # jamo 라이브러리 사용 가능
 
    # 2. 불필요한 이모지 제거
    emoji_pattern = re.compile(
        "["
        "\U0001F600-\U0001F64F"
        "\U0001F300-\U0001F5FF"
        "\U0001F680-\U0001F6FF"
        "\U0001F1E0-\U0001F1FF"
        "]+",
        flags=re.UNICODE
    )
    text = emoji_pattern.sub("", text)
 
    # 3. 반복 문자 정규화 (ㅋㅋㅋㅋㅋ -> ㅋㅋ)
    text = re.sub(r"(.)\1{4,}", r"\1\1", text)
 
    return text

import hashlib
 
 
def exact_dedup(dataset: list[TrainingExample]) -> list[TrainingExample]:
    """정확한 중복 제거 (해시 기반)"""
    seen_hashes: set[str] = set()
    unique_data: list[TrainingExample] = []
 
    for example in dataset:
        # 메시지 내용을 연결하여 해시 생성
        content = "||".join(
            msg["role"] + ":" + msg["content"]
            for msg in example["messages"]
        )
        content_hash = hashlib.sha256(content.encode()).hexdigest()
 
        if content_hash not in seen_hashes:
            seen_hashes.add(content_hash)
            unique_data.append(example)
 
    removed = len(dataset) - len(unique_data)
    print("정확한 중복 제거: " + str(removed) + "개 제거됨")
    return unique_data

from datasketch import MinHash, MinHashLSH
 
 
def near_dedup(
    dataset: list[TrainingExample],
    threshold: float = 0.8
) -> list[TrainingExample]:
    """유사 중복 제거 (MinHash LSH)"""
    lsh = MinHashLSH(threshold=threshold, num_perm=128)
    minhashes: list[MinHash] = []
 
    # MinHash 생성
    for i, example in enumerate(dataset):
        content = " ".join(
            msg["content"] for msg in example["messages"]
        )
        mh = MinHash(num_perm=128)
        for word in content.split():
            mh.update(word.encode("utf-8"))
        minhashes.append(mh)
 
        try:
            lsh.insert(str(i), mh)
        except ValueError:
            pass  # 이미 유사한 항목이 존재
 
    # 중복 그룹 탐지
    duplicates: set[int] = set()
    for i, mh in enumerate(minhashes):
        if i in duplicates:
            continue
        results = lsh.query(mh)
        for r in results:
            idx = int(r)
            if idx != i:
                duplicates.add(idx)
 
    unique_data = [
        ex for i, ex in enumerate(dataset)
        if i not in duplicates
    ]
 
    removed = len(dataset) - len(unique_data)
    print("유사 중복 제거: " + str(removed) + "개 제거됨")
    return unique_data

def rule_based_filter(example: TrainingExample) -> bool:
    """규칙 기반 품질 필터 (True면 유지, False면 제거)"""
    messages = example["messages"]
 
    # 어시스턴트 메시지 추출
    assistant_msgs = [
        m["content"] for m in messages if m["role"] == "assistant"
    ]
    if not assistant_msgs:
        return False
 
    response = " ".join(assistant_msgs)
 
    # 1. 최소 길이 (너무 짧은 응답 제거)
    if len(response) < 50:
        return False
 
    # 2. 최대 길이 (비정상적으로 긴 응답 제거)
    if len(response) > 10000:
        return False
 
    # 3. 반복 패턴 감지
    words = response.split()
    if len(words) > 10:
        unique_ratio = len(set(words)) / len(words)
        if unique_ratio < 0.3:
            return False
 
    # 4. 불완전한 응답 감지
    if response.rstrip().endswith("...") and len(response) < 200:
        return False
 
    # 5. 거부 응답 제거 ("할 수 없습니다" 류)
    refusal_patterns = [
        "죄송합니다만",
        "답변을 드리기 어렵",
        "도움을 드리기 어렵",
    ]
    if any(p in response[:100] for p in refusal_patterns):
        return False
 
    return True

import anthropic
 
client = anthropic.Anthropic()
 
 
def llm_quality_score(example: TrainingExample) -> float:
    """LLM을 사용한 데이터 품질 점수 (0~1)"""
    messages_text = "\n".join(
        "[" + m["role"] + "]: " + m["content"]
        for m in example["messages"]
    )
 
    prompt = (
        "다음 대화 데이터의 품질을 0.0에서 1.0 사이의 "
        "점수로 평가해 주세요.\n\n"
        "평가 기준:\n"
        "1. 정확성: 정보가 사실에 부합하는가\n"
        "2. 완전성: 질문에 충분히 답변했는가\n"
        "3. 명확성: 설명이 이해하기 쉬운가\n"
        "4. 유용성: 실질적으로 도움이 되는가\n\n"
        "대화:\n" + messages_text + "\n\n"
        "점수만 숫자로 응답해 주세요 (예: 0.85)"
    )
 
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=10,
        messages=[{"role": "user", "content": prompt}]
    )
 
    try:
        score = float(response.content[0].text.strip())
        return min(max(score, 0.0), 1.0)
    except ValueError:
        return 0.5

import re
 
 
def remove_pii(text: str) -> str:
    """개인 정보 패턴 감지 및 마스킹"""
 
    # 이메일 주소
    text = re.sub(
        r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
        "[EMAIL]",
        text
    )
 
    # 전화번호 (한국 형식)
    text = re.sub(
        r"0\d{1,2}[-.\s]?\d{3,4}[-.\s]?\d{4}",
        "[PHONE]",
        text
    )
 
    # 주민등록번호
    text = re.sub(
        r"\d{6}[-\s]?\d{7}",
        "[RRN]",
        text
    )
 
    # IP 주소
    text = re.sub(
        r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b",
        "[IP]",
        text
    )
 
    # 신용카드 번호
    text = re.sub(
        r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b",
        "[CARD]",
        text
    )
 
    return text

from transformers import AutoTokenizer
 
tokenizer = AutoTokenizer.from_pretrained(
    "meta-llama/Llama-3.1-8B-Instruct"
)
 
# 패딩 토큰 설정 (없는 경우)
if tokenizer.pad_token is None:
    tokenizer.pad_token = tokenizer.eos_token
 
 
def tokenize_example(example, max_length=2048):
    """대화 형식 데이터를 토큰화"""
    formatted = tokenizer.apply_chat_template(
        example["messages"],
        tokenize=False,
        add_generation_prompt=False
    )
 
    tokenized = tokenizer(
        formatted,
        max_length=max_length,
        truncation=True,
        padding=False,  # 동적 패딩 사용 시
        return_tensors="pt"
    )
 
    return tokenized

import torch
 
IGNORE_INDEX = -100
 
 
def create_labels_with_masking(
    input_ids: torch.Tensor,
    assistant_start_token_id: int,
    assistant_end_token_id: int
) -> torch.Tensor:
    """사용자 입력 부분을 마스킹한 레이블 생성"""
    labels = input_ids.clone()
    in_assistant = False
 
    for i in range(len(labels)):
        token_id = labels[i].item()
 
        if token_id == assistant_start_token_id:
            in_assistant = True
            labels[i] = IGNORE_INDEX  # 시작 토큰 자체는 마스킹
            continue
 
        if token_id == assistant_end_token_id:
            in_assistant = False
            # 종료 토큰은 학습 대상에 포함
 
        if not in_assistant:
            labels[i] = IGNORE_INDEX
 
    return labels

패킹 전 (패딩 낭비 발생):
  [예제1][PAD][PAD][PAD][PAD][PAD]  -> 50% 낭비
  [예제2][PAD][PAD][PAD]            -> 30% 낭비
  [예제3][PAD][PAD][PAD][PAD]       -> 40% 낭비
 
패킹 후 (효율적):
  [예제1][SEP][예제2][SEP][예제3][PAD]  -> 10% 낭비

def pack_sequences(
    examples: list[list[int]],
    max_length: int = 2048,
    sep_token_id: int = 0
) -> list[list[int]]:
    """여러 시퀀스를 하나로 패킹"""
    packed: list[list[int]] = []
    current_pack: list[int] = []
 
    for seq in examples:
        # 현재 팩에 추가할 수 있는지 확인
        needed = len(seq) + (1 if current_pack else 0)
        if len(current_pack) + needed <= max_length:
            if current_pack:
                current_pack.append(sep_token_id)
            current_pack.extend(seq)
        else:
            if current_pack:
                packed.append(current_pack)
            current_pack = list(seq)
 
    if current_pack:
        packed.append(current_pack)
 
    return packed

import json
from pathlib import Path
 
 
def run_preprocessing_pipeline(
    input_path: str,
    output_path: str,
    max_length: int = 2048,
    quality_threshold: float = 0.6
):
    """전체 전처리 파이프라인 실행"""
 
    # 1. 데이터 로드
    print("1. 데이터 로드 중...")
    with open(input_path) as f:
        raw_data = [json.loads(line) for line in f]
    print("   원본: " + str(len(raw_data)) + "개")
 
    # 2. 형식 표준화
    print("2. 형식 표준화 중...")
    normalized = [normalize_alpaca_format(d) for d in raw_data]
 
    # 3. 텍스트 정제
    print("3. 텍스트 정제 중...")
    cleaned = [clean_example(ex) for ex in normalized]
 
    # 4. 정확한 중복 제거
    print("4. 중복 제거 중...")
    deduped = exact_dedup(cleaned)
 
    # 5. 품질 필터링
    print("5. 품질 필터링 중...")
    filtered = [ex for ex in deduped if rule_based_filter(ex)]
    print("   필터링 후: " + str(len(filtered)) + "개")
 
    # 6. PII 제거
    print("6. 개인 정보 제거 중...")
    for ex in filtered:
        for msg in ex["messages"]:
            msg["content"] = remove_pii(msg["content"])
 
    # 7. 저장
    print("7. 저장 중...")
    output = Path(output_path)
    output.parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, "w") as f:
        for ex in filtered:
            f.write(json.dumps(ex, ensure_ascii=False) + "\n")
 
    print("완료: " + str(len(filtered)) + "개 저장됨")
    return filtered

데이터 검증 체크리스트:
 
  형식 검증:
    - 모든 예제가 동일한 스키마를 따르는가
    - 필수 필드 (messages, role, content)가 빠진 예제가 없는가
    - role 값이 system/user/assistant 중 하나인가
 
  내용 검증:
    - 빈 문자열인 메시지가 없는가
    - 토큰 수가 max_length를 초과하는 예제가 없는가
    - 사용자 메시지 없이 어시스턴트 응답만 있는 예제가 없는가
 
  품질 검증:
    - 무작위 50개 샘플링하여 수동 검토
    - 입력-출력의 관련성 확인
    - 응답의 정확성과 완전성 확인
 
  통계 검증:
    - 토큰 길이 분포가 합리적인가
    - 주제 분포가 편향되지 않았는가
    - train/val/test 분할이 적절한가