韓国語LLM学習データ制作完全ガイド：Hugging Faceデータセット、前処理、品質管理まで

韓国語LLMデータセットエコシステム
├── 事前学習
│   ├── Korean Wikipedia (~600K articles)
│   ├── Namuwiki Dump (~5GB)
│   ├── AI Hub コーパス (NIKL)
│   └── mC4-ko (Korean subset)
├── Instruction Tuning
│   ├── KoAlpaca (beomi) - 52K
│   ├── KoVicuna (melodysdreamj) - 40K+
│   ├── KOpen-platypus - 25K
│   ├── ko_wikidata_QA - Wiki 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

# DataFrameからデータセット作成
df = pd.DataFrame({
    "instruction": ["韓国の首都は？", "Pythonとは？"],
    "output": ["韓国の首都はソウルです。", "Pythonはプログラミング言語です。"],
})
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),
        }
"""

# 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 = [
    "韓国の四季について説明してください。",
    "Pythonでリスト内包表記の利点は？",
    "メール作成時の注意事項を教えてください。",
]

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}個生成してください。
多様なトピック（科学、歴史、技術、日常生活など）を含めてください。
各指示文は番号付きで1行で記述してください。"""

    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

元データ
    |
    v
+-------------------+
| 1. テキスト整形    |  HTMLタグ除去、エンコーディング整理
+--------+----------+
         v
+-------------------+
| 2. 言語検出       |  韓国語テキストのみフィルタリング
+--------+----------+
         v
+-------------------+
| 3. 重複除去       |  MinHash LSH, Exact Match
+--------+----------+
         v
+-------------------+
| 4. 品質フィルタ    |  Perplexity, 長さ, 毒性
+--------+----------+
         v
+-------------------+
| 5. PII除去       |  個人情報マスキング
+--------+----------+
         v
+-------------------+
| 6. トークン化     |  SentencePiece / BPE
+--------+----------+
         v
    整形済みデータ

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)

    return text.strip()

# バッチ処理
def clean_batch(texts):
    """バッチ単位のクリーニング"""
    cleaned = []
    for text in texts:
        result = clean_text(text)
        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):
    """テキストの言語検出"""
    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 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)

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}',
        "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

{
  "instruction": "次のテキストを要約してください。",
  "input": "人工知能（AI）は、人間の学習能力、推論能力、知覚能力を人工的に実現したコンピュータ科学の分野です...",
  "output": "AIは人間の知能をコンピュータで実現する技術で、機械学習とディープラーニングの発展により様々な分野で活用されています。"
}

{
  "conversations": [
    {
      "from": "human",
      "value": "Pythonでリストとタプルの違いは何ですか？"
    },
    {
      "from": "gpt",
      "value": "Pythonでのリストとタプルの主な違いは...\n\n1. **可変性**: リストは変更可能(mutable)、タプルは変更不可(immutable)\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 batch_convert(dataset, source_format, target_format):
    """データセット全体のフォーマット変換"""
    converters = {
        ("sharegpt", "openai"): sharegpt_to_openai,
        ("alpaca", "openai"): alpaca_to_openai,
    }
    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. **雑穀ご飯**: 多様な栄養素をバランスよく摂取できます。",
    "rejected": "うーん...ビビンバかな。美味しいから。焼肉もね。",
}

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)

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 check_contamination(train_data, benchmark_data, n_gram=13):
    """学習データとベンチマーク間の汚染(contamination)検査"""
    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)

    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 = []

    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",
        })

    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"])

        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-6: 重複除去 -> フォーマット変換 -> 検証 -> アップロード =====
if __name__ == "__main__":
    raw_data = collect_data()
    cleaned_data = clean_data(raw_data)
    unique_data = remove_duplicates(cleaned_data)
    formatted_data = format_data(unique_data, "openai")
    is_valid = validate_data(formatted_data, "openai")
    if is_valid:
        upload_to_hub(formatted_data, "my-org/korean-sft-v1")