Split View: 한국어 LLM 학습 데이터 제작 완전 가이드: Hugging Face 데이터셋, 전처리, 품질 관리까지

한국어 LLM 데이터셋 생태계
├── 사전학습
│   ├── Korean Wikipedia (~600K articles)
│   ├── Namuwiki Dump (~5GB)
│   ├── AI Hub 말뭉치 (국립국어원)
│   └── mC4-ko (Korean subset)
├── Instruction Tuning
│   ├── KoAlpaca (beomi) - 52K
│   ├── KoVicuna (melodysdreamj) - 40K+
│   ├── KOpen-platypus - 25K
│   ├── ko_wikidata_QA - 위키 기반 QA
│   └── kullm-v2 (고려대) - 152K
├── 선호도/정렬
│   ├── ko-rlhf (커뮤니티)
│   └── KoreanFeedback (자체 구축)
└── 평가
    ├── KLUE (8 tasks)
    ├── KoBBQ (편향)
    └── KorNAT (상식)

from datasets import load_dataset, Dataset, DatasetDict

# 기본 로딩
ds = load_dataset("beomi/KoAlpaca-v1.1a")
print(ds)
# DatasetDict({
#     train: Dataset({
#         features: ['instruction', 'output'],
#         num_rows: 21155
#     })
# })

# 특정 split 로딩
train_ds = load_dataset("beomi/KoAlpaca-v1.1a", split="train")

# 처음 5개 확인
for example in train_ds.select(range(5)):
    print(f"Instruction: {example['instruction'][:50]}...")
    print(f"Output: {example['output'][:50]}...")
    print("---")

# 길이 기반 필터링
filtered_ds = ds["train"].filter(
    lambda x: len(x["instruction"]) > 10 and len(x["output"]) > 20
)
print(f"필터링 후: {len(filtered_ds)} / {len(ds['train'])}")

# Alpaca 포맷으로 변환
def format_alpaca(example):
    text = f"""### Instruction:
{example['instruction']}

### Response:
{example['output']}"""
    return {"text": text}

formatted_ds = filtered_ds.map(format_alpaca)

# 토크나이저 적용
from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("beomi/llama-2-ko-7b")

def tokenize_function(examples):
    return tokenizer(
        examples["text"],
        truncation=True,
        max_length=2048,
        padding="max_length",
    )

tokenized_ds = formatted_ds.map(
    tokenize_function,
    batched=True,
    remove_columns=formatted_ds.column_names,
)

# 스트리밍으로 대용량 데이터 처리 (메모리 효율적)
streaming_ds = load_dataset(
    "allenai/c4",
    "ko",
    split="train",
    streaming=True,
)

# 처음 100개만 순회
for i, example in enumerate(streaming_ds):
    if i >= 100:
        break
    process(example["text"])

# 스트리밍 + 필터링 + 배치 처리
filtered_stream = streaming_ds.filter(
    lambda x: len(x["text"]) > 100
).take(10000)

# 배치 단위로 처리
batch = []
for example in filtered_stream:
    batch.append(example)
    if len(batch) == 32:
        process_batch(batch)
        batch = []

from datasets import Dataset
import pandas as pd

# 데이터프레임에서 데이터셋 생성
df = pd.DataFrame({
    "instruction": ["한국의 수도는?", "파이썬이란?"],
    "output": ["한국의 수도는 서울입니다.", "파이썬은 프로그래밍 언어입니다."],
})
my_dataset = Dataset.from_pandas(df)

# Hub에 업로드
my_dataset.push_to_hub(
    "my-org/my-korean-dataset",
    private=True,  # 비공개 설정
    token="hf_xxxxx",
)

# Dataset Card 자동 생성
from huggingface_hub import DatasetCard

card = DatasetCard.load("my-org/my-korean-dataset")
card.text = """
# My Korean Dataset
한국어 Instruction Tuning 데이터셋입니다.
## 데이터 구조
- instruction: 질문/지시문
- output: 응답
## 라이선스
CC-BY-4.0
"""
card.push_to_hub("my-org/my-korean-dataset")

# newspaper3k를 이용한 뉴스 크롤링
from newspaper import Article
import json

def crawl_article(url):
    """뉴스 기사 크롤링 (robots.txt 준수 필수!)"""
    article = Article(url, language="ko")
    article.download()
    article.parse()

    return {
        "title": article.title,
        "text": article.text,
        "publish_date": str(article.publish_date),
        "source_url": url,
    }

# Scrapy를 이용한 대규모 크롤링
# scrapy_spider.py
"""
import scrapy

class KoreanTextSpider(scrapy.Spider):
    name = 'korean_text'
    custom_settings = {
        'ROBOTSTXT_OBEY': True,  # robots.txt 준수 필수
        'DOWNLOAD_DELAY': 2,      # 2초 간격
        'CONCURRENT_REQUESTS': 4, # 동시 요청 제한
    }

    def parse(self, response):
        text = response.css('article::text').getall()
        yield {
            'url': response.url,
            'text': ' '.join(text),
        }
"""

# AI Hub 데이터 로딩 예시
import json
import glob

def load_aihub_data(data_dir):
    """AI Hub JSON 포맷 데이터 로딩"""
    all_data = []
    for filepath in glob.glob(f"{data_dir}/**/*.json", recursive=True):
        with open(filepath, "r", encoding="utf-8") as f:
            data = json.load(f)
            # AI Hub 형식에 따라 파싱
            if "document" in data:
                for doc in data["document"]:
                    for sent in doc.get("sentence", []):
                        all_data.append({
                            "text": sent.get("form", ""),
                            "source": "aihub",
                        })
    return all_data

# NLLB (No Language Left Behind)를 이용한 번역
from transformers import AutoModelForSeq2SeqLM, AutoTokenizer

model_name = "facebook/nllb-200-distilled-600M"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSeq2SeqLM.from_pretrained(model_name)

def translate_en_to_ko(text):
    """영어 -> 한국어 번역"""
    tokenizer.src_lang = "eng_Latn"
    inputs = tokenizer(text, return_tensors="pt", max_length=512, truncation=True)
    translated = model.generate(
        **inputs,
        forced_bos_token_id=tokenizer.convert_tokens_to_ids("kor_Hang"),
        max_length=512,
    )
    return tokenizer.decode(translated[0], skip_special_tokens=True)

# 번역 품질 검증
def validate_translation(original, translated):
    """번역 품질 자동 검증"""
    checks = {
        "not_empty": len(translated.strip()) > 0,
        "not_too_short": len(translated) > len(original) * 0.3,
        "not_too_long": len(translated) < len(original) * 3,
        "no_english_majority": sum(1 for c in translated if c.isascii()) / max(len(translated), 1) < 0.5,
    }
    return all(checks.values()), checks

import openai
import json
import random

# Self-Instruct: 시드 데이터에서 새로운 지시문 생성
SEED_INSTRUCTIONS = [
    "한국의 4계절에 대해 설명해주세요.",
    "파이썬에서 리스트 컴프리헨션의 장점은?",
    "이메일 작성 시 주의사항을 알려주세요.",
]

def generate_new_instructions(seed_instructions, num_generate=10):
    """GPT-4로 새로운 instruction 생성"""
    prompt = f"""다음은 한국어 지시문 예시입니다:

{chr(10).join(f'{i+1}. {inst}' for i, inst in enumerate(seed_instructions))}

위 예시와 비슷한 스타일이지만 완전히 새로운 한국어 지시문을 {num_generate}개 생성하세요.
다양한 주제(과학, 역사, 기술, 일상생활 등)를 포함하세요.
각 지시문은 번호와 함께 한 줄로 작성하세요."""

    client = openai.OpenAI()
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.8,
    )
    return parse_instructions(response.choices[0].message.content)

def generate_response(instruction):
    """지시문에 대한 응답 생성"""
    client = openai.OpenAI()
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "당신은 도움이 되는 한국어 AI 어시스턴트입니다."},
            {"role": "user", "content": instruction},
        ],
        temperature=0.7,
    )
    return response.choices[0].message.content

def evolve_instruction(instruction, evolution_type="deepen"):
    """WizardLM의 Evol-Instruct: 지시문을 점진적으로 복잡하게 만들기"""

    evolution_prompts = {
        "deepen": f"""다음 지시문을 더 깊이 있고 구체적으로 만들어주세요.
원본: {instruction}
진화된 버전:""",
        "broaden": f"""다음 지시문의 범위를 넓혀서 더 포괄적으로 만들어주세요.
원본: {instruction}
진화된 버전:""",
        "concretize": f"""다음 지시문에 구체적인 조건이나 제약을 추가해주세요.
원본: {instruction}
진화된 버전:""",
        "reasoning": f"""다음 지시문을 단계적 추론이 필요한 형태로 변환해주세요.
원본: {instruction}
진화된 버전:""",
    }

    client = openai.OpenAI()
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": evolution_prompts[evolution_type]}],
        temperature=0.7,
    )
    return response.choices[0].message.content

원본 데이터
    │
    ▼
┌─────────────────┐
│ 1. 텍스트 정제   │  HTML 태그 제거, 인코딩 정리
└────────┬────────┘
         ▼
┌─────────────────┐
│ 2. 언어 감지     │  한국어 텍스트만 필터링
└────────┬────────┘
         ▼
┌─────────────────┐
│ 3. 중복 제거     │  MinHash LSH, Exact Match
└────────┬────────┘
         ▼
┌─────────────────┐
│ 4. 품질 필터링   │  Perplexity, 길이, 독성
└────────┬────────┘
         ▼
┌─────────────────┐
│ 5. PII 제거     │  개인정보 마스킹
└────────┬────────┘
         ▼
┌─────────────────┐
│ 6. 토크나이징    │  SentencePiece / BPE
└────────┬────────┘
         ▼
    정제된 데이터

import re
import html
import unicodedata

def clean_text(text):
    """한국어 텍스트 기본 정제"""
    # HTML 엔티티 디코딩
    text = html.unescape(text)

    # HTML 태그 제거
    text = re.sub(r'<[^>]+>', '', text)

    # URL 제거
    text = re.sub(r'https?://\S+|www\.\S+', '', text)

    # 이메일 제거
    text = re.sub(r'\S+@\S+\.\S+', '[EMAIL]', text)

    # 전화번호 마스킹
    text = re.sub(r'\d{2,3}-\d{3,4}-\d{4}', '[PHONE]', text)

    # 연속 공백 정리
    text = re.sub(r'\s+', ' ', text)

    # Unicode 정규화 (NFC)
    text = unicodedata.normalize('NFC', text)

    # 한국어에 불필요한 특수문자 제거 (기본 문장부호 유지)
    text = re.sub(r'[^\w\s가-힣ㄱ-ㅎㅏ-ㅣa-zA-Z0-9.,!?;:\'\"()\-]', '', text)

    return text.strip()

def clean_korean_specific(text):
    """한국어 특화 정제"""
    # 자음/모음만 있는 경우 제거 (ㅋㅋㅋ, ㅎㅎㅎ 등은 상황에 따라)
    # 광고성 텍스트 패턴 제거
    ad_patterns = [
        r'지금\s*바로\s*클릭',
        r'무료\s*상담',
        r'카카오톡?\s*문의',
        r'전화\s*주세요',
    ]
    for pattern in ad_patterns:
        if re.search(pattern, text):
            return None  # 광고성으로 판단되면 제거

    return text

# 배치 처리
def clean_batch(texts):
    """배치 단위 정제"""
    cleaned = []
    for text in texts:
        result = clean_text(text)
        result = clean_korean_specific(result)
        if result and len(result) > 20:
            cleaned.append(result)
    return cleaned

from datasketch import MinHash, MinHashLSH
import hashlib

class TextDeduplicator:
    """MinHash LSH 기반 근사 중복 제거"""

    def __init__(self, threshold=0.8, num_perm=128):
        self.threshold = threshold
        self.num_perm = num_perm
        self.lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)
        self.seen_exact = set()

    def get_minhash(self, text):
        """텍스트의 MinHash 생성"""
        m = MinHash(num_perm=self.num_perm)
        # 3-gram 단위로 분할
        for i in range(len(text) - 2):
            m.update(text[i:i+3].encode('utf-8'))
        return m

    def is_duplicate(self, text, doc_id):
        """중복 여부 확인"""
        # 1. 정확 매칭 (해시 기반)
        text_hash = hashlib.md5(text.encode('utf-8')).hexdigest()
        if text_hash in self.seen_exact:
            return True
        self.seen_exact.add(text_hash)

        # 2. 근사 매칭 (MinHash LSH)
        minhash = self.get_minhash(text)
        result = self.lsh.query(minhash)
        if result:
            return True

        self.lsh.insert(doc_id, minhash)
        return False

    def deduplicate(self, documents):
        """문서 리스트 중복 제거"""
        unique_docs = []
        for i, doc in enumerate(documents):
            if not self.is_duplicate(doc["text"], f"doc_{i}"):
                unique_docs.append(doc)

        print(f"중복 제거: {len(documents)} -> {len(unique_docs)} "
              f"({len(documents) - len(unique_docs)}개 제거)")
        return unique_docs

import fasttext

# fasttext 언어 감지 모델 로드
model_path = "lid.176.bin"  # 사전 다운로드 필요
lang_model = fasttext.load_model(model_path)

def detect_language(text):
    """텍스트 언어 감지"""
    # 줄바꿈 제거 (fasttext는 한 줄 입력)
    text_clean = text.replace('\n', ' ')[:200]
    predictions = lang_model.predict(text_clean)
    lang = predictions[0][0].replace('__label__', '')
    confidence = predictions[1][0]
    return lang, confidence

def filter_korean(documents, min_confidence=0.7):
    """한국어 텍스트만 필터링"""
    korean_docs = []
    for doc in documents:
        lang, conf = detect_language(doc["text"])
        if lang == "ko" and conf >= min_confidence:
            korean_docs.append(doc)
    return korean_docs

import numpy as np
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

class QualityFilter:
    """텍스트 품질 필터링"""

    def __init__(self):
        self.criteria = {
            "min_length": 50,
            "max_length": 10000,
            "min_words": 10,
            "max_repetition_ratio": 0.3,
            "max_special_char_ratio": 0.1,
        }

    def check_length(self, text):
        """길이 기반 필터"""
        return self.criteria["min_length"] <= len(text) <= self.criteria["max_length"]

    def check_repetition(self, text):
        """반복 텍스트 감지"""
        words = text.split()
        if len(words) == 0:
            return False
        unique_ratio = len(set(words)) / len(words)
        return unique_ratio >= (1 - self.criteria["max_repetition_ratio"])

    def check_special_chars(self, text):
        """특수 문자 비율 확인"""
        special = sum(1 for c in text if not c.isalnum() and not c.isspace()
                     and c not in '.,!?;:')
        return special / max(len(text), 1) < self.criteria["max_special_char_ratio"]

    def compute_perplexity(self, text, model, tokenizer, device="cuda"):
        """Perplexity 기반 품질 평가 (낮을수록 자연스러운 텍스트)"""
        inputs = tokenizer(text, return_tensors="pt", truncation=True,
                          max_length=512).to(device)
        with torch.no_grad():
            outputs = model(**inputs, labels=inputs["input_ids"])
        return torch.exp(outputs.loss).item()

    def filter(self, text):
        """종합 품질 필터링"""
        return (
            self.check_length(text)
            and self.check_repetition(text)
            and self.check_special_chars(text)
        )

import re
from typing import Dict, List

class PIIRemover:
    """개인식별정보(PII) 제거"""

    PATTERNS = {
        "주민등록번호": r'\d{6}[-]?\d{7}',
        "전화번호": r'0\d{1,2}[-.]?\d{3,4}[-.]?\d{4}',
        "이메일": r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',
        "카드번호": r'\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}',
        "계좌번호": r'\d{3,6}[-]?\d{2,6}[-]?\d{2,6}[-]?\d{0,3}',
        "IP주소": r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}',
    }

    def remove_pii(self, text: str) -> str:
        """PII를 마스킹 토큰으로 대체"""
        for pii_type, pattern in self.PATTERNS.items():
            mask_token = f"[{pii_type}]"
            text = re.sub(pattern, mask_token, text)
        return text

    def detect_pii(self, text: str) -> Dict[str, List[str]]:
        """PII 감지 (제거 전 확인용)"""
        found = {}
        for pii_type, pattern in self.PATTERNS.items():
            matches = re.findall(pattern, text)
            if matches:
                found[pii_type] = matches
        return found

from tokenizers import Tokenizer, models, trainers, pre_tokenizers

# SentencePiece 한국어 토크나이저 학습
def train_korean_tokenizer(text_files, vocab_size=32000):
    """한국어 특화 BPE 토크나이저 학습"""
    tokenizer = Tokenizer(models.BPE(unk_token="[UNK]"))

    # 한국어는 pre-tokenization이 중요
    tokenizer.pre_tokenizer = pre_tokenizers.Sequence([
        pre_tokenizers.ByteLevel(add_prefix_space=False),
    ])

    trainer = trainers.BpeTrainer(
        vocab_size=vocab_size,
        special_tokens=["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]"],
        min_frequency=2,
    )

    tokenizer.train(text_files, trainer)
    return tokenizer

# 토크나이저 효율성 비교
def compare_tokenizer_efficiency(text, tokenizers_dict):
    """여러 토크나이저의 한국어 토큰 효율성 비교"""
    print(f"원문 ({len(text)}자): {text[:50]}...")
    print("-" * 60)
    for name, tok in tokenizers_dict.items():
        tokens = tok.encode(text)
        token_strs = tok.convert_ids_to_tokens(tokens)
        fertility = len(tokens) / len(text.split())
        print(f"{name}: {len(tokens)} tokens, fertility={fertility:.2f}")
        print(f"  처음 10토큰: {token_strs[:10]}")

{
  "instruction": "다음 텍스트를 요약해주세요.",
  "input": "인공지능(AI)은 인간의 학습능력, 추론능력, 지각능력을 인공적으로 구현한 컴퓨터 과학의 세부분야입니다...",
  "output": "인공지능은 인간의 지능을 컴퓨터로 구현하는 기술로, 머신러닝과 딥러닝의 발전으로 다양한 분야에서 활용되고 있습니다."
}

def format_alpaca(instruction, input_text="", output=""):
    """Alpaca 포맷 생성"""
    if input_text:
        return {
            "instruction": instruction,
            "input": input_text,
            "output": output,
        }
    return {
        "instruction": instruction,
        "input": "",
        "output": output,
    }

# Alpaca 프롬프트 템플릿
ALPACA_TEMPLATE = """Below is an instruction that describes a task. Write a response that appropriately completes the request.

### Instruction:
{instruction}

### Input:
{input}

### Response:
{output}"""

ALPACA_TEMPLATE_NO_INPUT = """Below is an instruction that describes a task. Write a response that appropriately completes the request.

### Instruction:
{instruction}

### Response:
{output}"""

{
  "conversations": [
    {
      "from": "human",
      "value": "파이썬에서 리스트와 튜플의 차이점이 뭐야?"
    },
    {
      "from": "gpt",
      "value": "파이썬에서 리스트와 튜플의 주요 차이점은...\n\n1. **가변성**: 리스트는 변경 가능(mutable), 튜플은 변경 불가(immutable)\n2. **성능**: 튜플이 메모리 효율적\n3. **문법**: 리스트는 [], 튜플은 ()"
    },
    {
      "from": "human",
      "value": "그러면 언제 튜플을 쓰는 게 좋아?"
    },
    {
      "from": "gpt",
      "value": "튜플은 다음과 같은 경우에 사용하면 좋습니다:\n\n1. 데이터가 변경되면 안 되는 경우 (좌표, RGB 값)\n2. 딕셔너리의 키로 사용할 때\n3. 함수에서 여러 값을 반환할 때"
    }
  ]
}

{
  "messages": [
    {
      "role": "system",
      "content": "당신은 한국어에 능통한 AI 어시스턴트입니다. 정확하고 도움이 되는 답변을 제공하세요."
    },
    {
      "role": "user",
      "content": "머신러닝과 딥러닝의 차이를 설명해줘."
    },
    {
      "role": "assistant",
      "content": "머신러닝과 딥러닝의 핵심 차이를 설명하겠습니다..."
    }
  ]
}

# Llama 3 Chat Template
LLAMA3_TEMPLATE = """<|begin_of_text|><|start_header_id|>system<|end_header_id|>

{system_message}<|eot_id|><|start_header_id|>user<|end_header_id|>

{user_message}<|eot_id|><|start_header_id|>assistant<|end_header_id|>

{assistant_message}<|eot_id|>"""

# Mistral Chat Template
MISTRAL_TEMPLATE = """<s>[INST] {system_message}

{user_message} [/INST]{assistant_message}</s>"""

# Qwen 2 Chat Template
QWEN2_TEMPLATE = """<|im_start|>system
{system_message}<|im_end|>
<|im_start|>user
{user_message}<|im_end|>
<|im_start|>assistant
{assistant_message}<|im_end|>"""

def sharegpt_to_openai(sharegpt_data):
    """ShareGPT -> OpenAI Messages 변환"""
    messages = []
    role_map = {"human": "user", "gpt": "assistant", "system": "system"}

    for conv in sharegpt_data["conversations"]:
        messages.append({
            "role": role_map.get(conv["from"], conv["from"]),
            "content": conv["value"],
        })
    return {"messages": messages}

def alpaca_to_openai(alpaca_data, system_prompt=""):
    """Alpaca -> OpenAI Messages 변환"""
    messages = []
    if system_prompt:
        messages.append({"role": "system", "content": system_prompt})

    user_content = alpaca_data["instruction"]
    if alpaca_data.get("input"):
        user_content += f"\n\n{alpaca_data['input']}"

    messages.append({"role": "user", "content": user_content})
    messages.append({"role": "assistant", "content": alpaca_data["output"]})

    return {"messages": messages}

def openai_to_sharegpt(openai_data):
    """OpenAI Messages -> ShareGPT 변환"""
    role_map = {"user": "human", "assistant": "gpt", "system": "system"}
    conversations = []

    for msg in openai_data["messages"]:
        conversations.append({
            "from": role_map.get(msg["role"], msg["role"]),
            "value": msg["content"],
        })
    return {"conversations": conversations}

# 대량 변환
def batch_convert(dataset, source_format, target_format):
    """데이터셋 전체 포맷 변환"""
    converters = {
        ("sharegpt", "openai"): sharegpt_to_openai,
        ("alpaca", "openai"): alpaca_to_openai,
        ("openai", "sharegpt"): openai_to_sharegpt,
    }
    converter = converters.get((source_format, target_format))
    if not converter:
        raise ValueError(f"지원하지 않는 변환: {source_format} -> {target_format}")

    return [converter(item) for item in dataset]

# DPO (Direct Preference Optimization) 데이터 구조
dpo_example = {
    "prompt": "한국의 전통 음식 중 건강에 좋은 것을 추천해주세요.",
    "chosen": "한국 전통 음식 중 건강에 좋은 대표적인 음식들을 소개하겠습니다.\n\n1. **김치**: 유산균이 풍부하고 비타민C, 식이섬유가 많습니다.\n2. **된장찌개**: 발효 식품으로 항암 효과가 있으며 단백질이 풍부합니다.\n3. **잡곡밥**: 다양한 영양소를 고르게 섭취할 수 있습니다.\n4. **나물**: 시금치, 콩나물 등 다양한 비타민과 미네랄을 제공합니다.",
    "rejected": "음... 비빔밥이요. 맛있으니까요. 불고기도요.",
}

ANNOTATION_GUIDELINES = """
## 선호도 평가 가이드라인

### 평가 기준 (1-5점)
1. **도움됨 (Helpfulness)**: 질문에 얼마나 잘 답했는가
2. **정확성 (Accuracy)**: 사실적으로 맞는 정보인가
3. **안전성 (Safety)**: 유해하거나 편향된 내용은 없는가
4. **유창성 (Fluency)**: 한국어가 자연스러운가

### 비교 평가 시 주의사항
- 두 응답 모두 읽은 후 비교할 것
- 길이가 아닌 품질 기준으로 판단
- 확신이 없으면 'tie'로 표시
- 개인 의견이 아닌 객관적 품질 기준으로 판단
"""

# 평가 데이터 수집 도구
class PreferenceCollector:
    def __init__(self):
        self.annotations = []

    def add_comparison(self, prompt, response_a, response_b, preference, annotator_id):
        """선호도 비교 결과 저장"""
        self.annotations.append({
            "prompt": prompt,
            "response_a": response_a,
            "response_b": response_b,
            "preference": preference,  # "a", "b", "tie"
            "annotator_id": annotator_id,
            "timestamp": datetime.now().isoformat(),
        })

    def compute_agreement(self):
        """평가자 간 일치도 계산"""
        from collections import Counter
        # 같은 prompt에 대한 평가 비교
        prompt_votes = {}
        for ann in self.annotations:
            key = (ann["prompt"], ann["response_a"][:50])
            if key not in prompt_votes:
                prompt_votes[key] = []
            prompt_votes[key].append(ann["preference"])

        agreements = []
        for key, votes in prompt_votes.items():
            if len(votes) >= 2:
                most_common = Counter(votes).most_common(1)[0][1]
                agreements.append(most_common / len(votes))
        return np.mean(agreements) if agreements else 0

def ai_rank_responses(prompt, responses, model="gpt-4"):
    """Constitutional AI 방식의 자동 순위 매기기"""
    ranking_prompt = f"""다음 질문에 대한 여러 응답을 평가해주세요.

질문: {prompt}

"""
    for i, resp in enumerate(responses):
        ranking_prompt += f"응답 {i+1}: {resp}\n\n"

    ranking_prompt += """각 응답을 다음 기준으로 1-5점 평가하고, 최종 순위를 매겨주세요:
1. 도움됨 (Helpfulness)
2. 정확성 (Accuracy)
3. 안전성 (Safety)
4. 유창성 (Fluency)

JSON 형식으로 결과를 반환하세요."""

    client = openai.OpenAI()
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": ranking_prompt}],
        response_format={"type": "json_object"},
    )
    return json.loads(response.choices[0].message.content)

def create_ultrafeedback_data(prompts, models_to_evaluate):
    """UltraFeedback 스타일의 다중 모델 응답 수집 및 평가"""
    dataset = []

    for prompt in prompts:
        responses = {}
        # 여러 모델에서 응답 수집
        for model_name in models_to_evaluate:
            responses[model_name] = generate_response(prompt, model_name)

        # GPT-4로 각 응답 평가 (1-10점)
        evaluations = {}
        for model_name, response in responses.items():
            score = evaluate_single_response(prompt, response)
            evaluations[model_name] = score

        # 최고/최저 점수 응답 선택 (DPO용)
        best_model = max(evaluations, key=evaluations.get)
        worst_model = min(evaluations, key=evaluations.get)

        dataset.append({
            "prompt": prompt,
            "chosen": responses[best_model],
            "rejected": responses[worst_model],
            "chosen_model": best_model,
            "rejected_model": worst_model,
            "scores": evaluations,
        })

    return dataset

from collections import Counter
import numpy as np

def vocabulary_diversity(texts):
    """어휘 다양성 측정 (Type-Token Ratio)"""
    all_tokens = []
    for text in texts:
        all_tokens.extend(text.split())

    types = len(set(all_tokens))
    tokens = len(all_tokens)
    ttr = types / tokens if tokens > 0 else 0
    return {"type_token_ratio": ttr, "unique_words": types, "total_words": tokens}

def topic_diversity(texts, n_topics=10):
    """토픽 다양성 (LDA 기반)"""
    from sklearn.decomposition import LatentDirichletAllocation
    from sklearn.feature_extraction.text import CountVectorizer

    vectorizer = CountVectorizer(max_features=5000)
    dtm = vectorizer.fit_transform(texts)

    lda = LatentDirichletAllocation(n_components=n_topics, random_state=42)
    topic_dist = lda.fit_transform(dtm)

    # 토픽 엔트로피 (높을수록 균등 분포)
    avg_dist = topic_dist.mean(axis=0)
    entropy = -np.sum(avg_dist * np.log(avg_dist + 1e-10))
    return {"topic_entropy": entropy, "max_entropy": np.log(n_topics)}

def instruction_diversity(instructions):
    """지시문 시작 동사 다양성"""
    first_words = [inst.split()[0] if inst.split() else "" for inst in instructions]
    counter = Counter(first_words)
    return {
        "unique_starters": len(counter),
        "top_10": counter.most_common(10),
        "starter_entropy": -sum(
            (c/len(first_words)) * np.log(c/len(first_words))
            for c in counter.values()
        ),
    }

import matplotlib.pyplot as plt

def analyze_length_distribution(dataset, text_field="text"):
    """데이터셋 길이 분포 분석"""
    lengths = [len(item[text_field]) for item in dataset]

    stats = {
        "count": len(lengths),
        "mean": np.mean(lengths),
        "median": np.median(lengths),
        "std": np.std(lengths),
        "min": np.min(lengths),
        "max": np.max(lengths),
        "p25": np.percentile(lengths, 25),
        "p75": np.percentile(lengths, 75),
        "p95": np.percentile(lengths, 95),
    }

    print("=== 길이 분포 통계 ===")
    for k, v in stats.items():
        print(f"  {k}: {v:.1f}")

    return stats

def check_contamination(train_data, benchmark_data, n_gram=13):
    """학습 데이터와 벤치마크 간 오염(contamination) 검사"""
    # 벤치마크 n-gram 집합 생성
    benchmark_ngrams = set()
    for item in benchmark_data:
        text = item["question"] if "question" in item else item["text"]
        words = text.split()
        for i in range(len(words) - n_gram + 1):
            ngram = " ".join(words[i:i+n_gram])
            benchmark_ngrams.add(ngram)

    # 학습 데이터에서 겹치는 n-gram 검색
    contaminated = []
    for i, item in enumerate(train_data):
        text = item.get("instruction", "") + " " + item.get("output", "")
        words = text.split()
        for j in range(len(words) - n_gram + 1):
            ngram = " ".join(words[j:j+n_gram])
            if ngram in benchmark_ngrams:
                contaminated.append({
                    "train_idx": i,
                    "matched_ngram": ngram,
                })
                break

    contamination_rate = len(contaminated) / max(len(train_data), 1)
    print(f"오염률: {contamination_rate:.4%} ({len(contaminated)}/{len(train_data)})")
    return contaminated

"""
한국어 SFT 데이터셋 구축 전체 파이프라인
수집 -> 정제 -> 포맷 변환 -> 검증 -> 업로드
"""

import json
import os
from datasets import Dataset
from tqdm import tqdm

# ===== Step 1: 데이터 수집 =====
def collect_data():
    """다양한 소스에서 데이터 수집"""
    all_data = []

    # 1-1. 기존 데이터셋 로드
    from datasets import load_dataset
    koalpaca = load_dataset("beomi/KoAlpaca-v1.1a", split="train")
    for item in koalpaca:
        all_data.append({
            "instruction": item["instruction"],
            "input": "",
            "output": item["output"],
            "source": "koalpaca",
        })

    # 1-2. 합성 데이터 추가
    synthetic = generate_synthetic_data(num_samples=1000)
    all_data.extend(synthetic)

    print(f"총 수집: {len(all_data)}개")
    return all_data

# ===== Step 2: 데이터 정제 =====
def clean_data(raw_data):
    """데이터 정제 파이프라인"""
    cleaned = []
    pii_remover = PIIRemover()
    quality_filter = QualityFilter()

    for item in tqdm(raw_data, desc="정제 중"):
        # 텍스트 정제
        instruction = clean_text(item["instruction"])
        output = clean_text(item["output"])

        # PII 제거
        instruction = pii_remover.remove_pii(instruction)
        output = pii_remover.remove_pii(output)

        # 품질 필터링
        if not quality_filter.filter(instruction) or not quality_filter.filter(output):
            continue

        cleaned.append({
            "instruction": instruction,
            "input": item.get("input", ""),
            "output": output,
            "source": item["source"],
        })

    print(f"정제 후: {len(cleaned)}/{len(raw_data)}")
    return cleaned

# ===== Step 3: 중복 제거 =====
def remove_duplicates(data):
    """중복 제거"""
    dedup = TextDeduplicator(threshold=0.85)
    docs = [{"text": item["instruction"] + " " + item["output"], **item} for item in data]
    unique = dedup.deduplicate(docs)
    return [{"instruction": d["instruction"], "input": d.get("input", ""),
             "output": d["output"], "source": d["source"]} for d in unique]

# ===== Step 4: 포맷 변환 =====
def format_data(data, target_format="openai"):
    """목표 포맷으로 변환"""
    formatted = []
    system_prompt = "당신은 도움이 되는 한국어 AI 어시스턴트입니다."

    for item in data:
        if target_format == "openai":
            formatted.append(alpaca_to_openai(item, system_prompt))
        elif target_format == "sharegpt":
            formatted.append({
                "conversations": [
                    {"from": "human", "value": item["instruction"]},
                    {"from": "gpt", "value": item["output"]},
                ],
            })
    return formatted

# ===== Step 5: 검증 =====
def validate_data(data, format_type="openai"):
    """데이터 품질 검증"""
    errors = []
    for i, item in enumerate(data):
        if format_type == "openai":
            if "messages" not in item:
                errors.append(f"[{i}] messages 필드 없음")
            elif len(item["messages"]) < 2:
                errors.append(f"[{i}] 메시지 수 부족")
            for msg in item.get("messages", []):
                if msg["role"] not in ("system", "user", "assistant"):
                    errors.append(f"[{i}] 잘못된 role: {msg['role']}")
                if not msg["content"].strip():
                    errors.append(f"[{i}] 빈 content")

    if errors:
        print(f"검증 오류 {len(errors)}개:")
        for e in errors[:10]:
            print(f"  {e}")
    else:
        print("검증 통과!")
    return len(errors) == 0

# ===== Step 6: 업로드 =====
def upload_to_hub(data, repo_name):
    """Hugging Face Hub에 업로드"""
    ds = Dataset.from_list(data)

    # Train/Validation 분할
    split_ds = ds.train_test_split(test_size=0.05, seed=42)

    split_ds.push_to_hub(
        repo_name,
        private=True,
    )
    print(f"업로드 완료: {repo_name}")
    print(f"  Train: {len(split_ds['train'])}, Validation: {len(split_ds['test'])}")

# ===== 실행 =====
if __name__ == "__main__":
    # 1. 수집
    raw_data = collect_data()

    # 2. 정제
    cleaned_data = clean_data(raw_data)

    # 3. 중복 제거
    unique_data = remove_duplicates(cleaned_data)

    # 4. 포맷 변환
    formatted_data = format_data(unique_data, "openai")

    # 5. 검증
    is_valid = validate_data(formatted_data, "openai")

    # 6. 업로드
    if is_valid:
        upload_to_hub(formatted_data, "my-org/korean-sft-v1")

def generate_quality_report(dataset):
    """데이터셋 품질 보고서 생성"""
    report = {
        "total_samples": len(dataset),
        "length_stats": analyze_length_distribution(dataset, "instruction"),
        "diversity": vocabulary_diversity([d["instruction"] for d in dataset]),
        "source_distribution": Counter(d["source"] for d in dataset),
    }

    print("=" * 60)
    print("데이터셋 품질 보고서")
    print("=" * 60)
    print(f"총 샘플 수: {report['total_samples']}")
    print(f"\n길이 통계:")
    for k, v in report['length_stats'].items():
        print(f"  {k}: {v:.1f}")
    print(f"\n어휘 다양성: TTR = {report['diversity']['type_token_ratio']:.4f}")
    print(f"\n소스 분포:")
    for source, count in report['source_distribution'].most_common():
        print(f"  {source}: {count} ({count/len(dataset)*100:.1f}%)")
    print("=" * 60)

    return report