Подготовка датасета для дообучения LLM

Подготовка датасета для дообучения LLM

Качество датасета — главный фактор успеха fine-tuning. "Мусор на входе — мусор на выходе" работает вдвойне для LLM: плохо структурированные или нерелевантные примеры не просто не помогают — они активно деградируют модель. 1000 качественных примеров лучше 100,000 плохих.

Форматы датасетов для fine-tuning

Instruction following (Alpaca формат):

{"instruction": "Переведи на английский", "input": "Привет мир", "output": "Hello world"}
{"instruction": "Напиши SQL запрос", "input": "Выбери всех пользователей старше 30", "output": "SELECT * FROM users WHERE age > 30;"}

Chat format (ShareGPT/ChatML):

{
  "conversations": [
    {"from": "system", "value": "Ты помощник по SQL"},
    {"from": "human", "value": "Как выбрать уникальные значения?"},
    {"from": "gpt", "value": "Используй SELECT DISTINCT: `SELECT DISTINCT column FROM table;`"}
  ]
}

Completion format (простой):

{"text": "### Вопрос: Что такое RLHF?\n### Ответ: RLHF (Reinforcement Learning from Human Feedback) — метод..."}

Требования к объёму датасета

Структура хорошего примера

from dataclasses import dataclass

@dataclass
class FineTuningExample:
    instruction: str    # Чёткая задача без двусмысленности
    input: str          # Конкретный контекст/данные (опционально)
    output: str         # Идеальный ответ модели

    def validate(self) -> list[str]:
        issues = []
        if len(self.output) < 10:
            issues.append("Output too short")
        if len(self.output) > 2000:
            issues.append("Output may be too long for this task")
        if self.output in ["I don't know", "N/A", ""]:
            issues.append("Uninformative output")
        # Проверка на утечку промпта в ответ
        if self.instruction.lower()[:20] in self.output.lower():
            issues.append("Output contains instruction text")
        return issues

Разделение на train/eval

from sklearn.model_selection import train_test_split

def split_dataset(examples: list, eval_ratio: float = 0.1) -> tuple:
    # Стратифицированное разделение по длине output
    short = [e for e in examples if len(e.output) < 200]
    medium = [e for e in examples if 200 <= len(e.output) < 500]
    long = [e for e in examples if len(e.output) >= 500]

    train, eval = [], []
    for group in [short, medium, long]:
        if len(group) > 1:
            tr, ev = train_test_split(group, test_size=eval_ratio)
            train.extend(tr); eval.extend(ev)

    return train, eval

Дедупликация

import hashlib
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity

def deduplicate_exact(examples: list) -> list:
    """Точная дедупликация по хэшу"""
    seen = set()
    unique = []
    for ex in examples:
        h = hashlib.md5(f"{ex.instruction}{ex.input}".encode()).hexdigest()
        if h not in seen:
            seen.add(h)
            unique.append(ex)
    return unique

def deduplicate_semantic(examples: list, threshold: float = 0.95) -> list:
    """Семантическая дедупликация (убирает near-duplicates)"""
    model = SentenceTransformer('all-MiniLM-L6-v2')
    texts = [f"{e.instruction} {e.input}" for e in examples]
    embeddings = model.encode(texts, batch_size=512, show_progress_bar=True)

    keep = [True] * len(examples)
    for i in range(len(examples)):
        if not keep[i]:
            continue
        for j in range(i+1, len(examples)):
            sim = cosine_similarity([embeddings[i]], [embeddings[j]])[0][0]
            if sim > threshold:
                keep[j] = False

    return [ex for ex, k in zip(examples, keep) if k]

Финальный чеклист перед обучением

• Нет дублей (exact и near-duplicate)
• Нет PII в датасете (имена, emails, телефоны)
• Output не содержит ссылок на даты/версии ("по состоянию на 2023")
• Равномерное распределение по типам задач
• Eval set не пересекается с train
• Токенизированные примеры не превышают max_length модели (без truncation)