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

Очищення та попередня обробка даних для fine-tuning LLM

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

Pipeline очищення

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_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% нормально, і кінцева якість моделі від цього тільки покращується.