Очистка и предобработка данных для дообучения LLM

Очистка и предобработка данных для дообучения LLM

Очистка данных для LLM fine-tuning имеет свою специфику: нужно не только удалить технический мусор (HTML-теги, дубли), но и отфильтровать токсичный контент, исправить encoding проблемы и убедиться, что примеры действительно соответствуют целевой задаче.

Пайплайн очистки

import re
import unicodedata
from dataclasses import dataclass

@dataclass
class CleaningResult:
    original: str
    cleaned: str
    removed: bool
    removal_reason: str = None

class TextCleaner:
    def clean(self, text: str) -> CleaningResult:
        cleaned = text

        # 1. Нормализация Unicode
        cleaned = unicodedata.normalize('NFKC', cleaned)

        # 2. Удаление HTML/XML тегов
        cleaned = re.sub(r'<[^>]+>', ' ', cleaned)

        # 3. Очистка URL (опционально — заменяем на placeholder)
        cleaned = re.sub(
            r'https?://[^\s]+', '[URL]', cleaned
        )

        # 4. Нормализация пробелов
        cleaned = re.sub(r'\s+', ' ', cleaned).strip()

        # 5. Удаление повторяющихся символов (ааааааа → а)
        cleaned = re.sub(r'(.)\1{4,}', r'\1\1', cleaned)

        # Проверка на минимальную длину
        if len(cleaned.split()) < 3:
            return CleaningResult(text, cleaned, True, "too_short")

        return CleaningResult(text, cleaned, False)

class DataFilter:
    def __init__(self):
        # Токсичность (можно использовать detoxify или fasttext)
        from detoxify import Detoxify
        self.toxicity_model = Detoxify('multilingual')

    def is_toxic(self, text: str, threshold: float = 0.7) -> bool:
        result = self.toxicity_model.predict(text)
        return result['toxicity'] > threshold

    def has_pii(self, text: str) -> bool:
        """Простая эвристика для PII детекции"""
        patterns = [
            r'\b\d{3}-\d{2}-\d{4}\b',           # SSN
            r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',  # Email
            r'\b(?:\+7|8)?[\s-]?\(?\d{3}\)?[\s-]?\d{3}[\s-]?\d{2}[\s-]?\d{2}\b',  # RU phone
            r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',  # Credit card
        ]
        for pattern in patterns:
            if re.search(pattern, text):
                return True
        return False

Очистка output полей

class OutputCleaner:
    def clean_output(self, output: str, task_type: str) -> tuple[str, bool]:
        cleaned = output.strip()

        # Удаление нежелательных фраз модели
        unwanted_starts = [
            "As an AI language model",
            "As a helpful assistant",
            "I don't have access to real-time",
            "I cannot browse the internet",
            "Certainly! Here",
            "Of course! I'd be happy to",
        ]

        for phrase in unwanted_starts:
            if cleaned.lower().startswith(phrase.lower()):
                # Удаляем вступительную фразу
                cleaned = cleaned[len(phrase):].lstrip('.,! ')

        # Проверка: output не должен содержать meta-комментарии
        meta_indicators = [
            "Note: This is a fictional",
            "[This response was",
            "Disclaimer:",
        ]
        for indicator in meta_indicators:
            if indicator in cleaned:
                idx = cleaned.find(indicator)
                cleaned = cleaned[:idx].strip()

        # Минимальная длина
        if len(cleaned.split()) < 5:
            return cleaned, True  # Пометить для удаления

        return cleaned, False

Детекция дублей разных уровней

from datasketch import MinHash, MinHashLSH

def find_near_duplicates(texts: list[str],
                          threshold: float = 0.8) -> list[tuple]:
    """MinHash LSH для эффективного поиска near-duplicates O(n log n)"""
    lsh = MinHashLSH(threshold=threshold, num_perm=128)
    minhashes = {}

    for i, text in enumerate(texts):
        m = MinHash(num_perm=128)
        for word in text.lower().split():
            m.update(word.encode('utf8'))
        lsh.insert(f"doc_{i}", m)
        minhashes[f"doc_{i}"] = m

    duplicates = []
    for i, text in enumerate(texts):
        key = f"doc_{i}"
        result = lsh.query(minhashes[key])
        result.remove(key)
        if result:
            duplicates.append((i, [int(r.split('_')[1]) for r in result]))

    return duplicates

Статистика после очистки

После очистки датасета важно проверить:

• % удалённых примеров по каждой причине (too_short, toxic, pii, duplicate)
• Распределение длин output (histogram)
• Словарное разнообразие (type-token ratio)
• Покрытие целевого домена (насколько примеры покрывают задачи)

Типичный результат: из 50,000 сырых примеров после очистки остаётся 35,000-42,000 высококачественных. Снижение объёма на 15-30% — норма, и итоговое качество модели от этого только улучшается.