LLM Training Data Pipeline

from datasketch import MinHash, MinHashLSH
import re

class DocumentDeduplicator:
    """Near-duplicate detection using MinHash + LSH.
    Processes billions of documents efficiently."""

    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)

    def get_shingles(self, text, k=5):
        text = re.sub(r'\s+', ' ', text.lower())
        return [text[i:i+k] for i in range(len(text) - k + 1)]

    def compute_minhash(self, text):
        m = MinHash(num_perm=self.num_perm)
        for shingle in self.get_shingles(text):
            m.update(shingle.encode('utf8'))
        return m

    def add_document(self, doc_id, text):
        self.lsh.insert(doc_id, self.compute_minhash(text))

    def find_duplicates(self, doc_id, text):
        minhash = self.compute_minhash(text)
        similar = self.lsh.query(minhash)
        return [d for d in similar if d != doc_id]

# Result on Common Crawl: 40% are duplicates

from transformers import GPT2LMHeadModel, GPT2Tokenizer
import torch

class QualityScorer:
    """Score document quality with GPT-2 perplexity.
    Low perplexity = coherent, natural text."""

    def __init__(self):
        self.model = GPT2LMHeadModel.from_pretrained('gpt2')
        self.tokenizer = GPT2Tokenizer.from_pretrained('gpt2')
        self.model.eval()

    def calculate_perplexity(self, text):
        encodings = self.tokenizer(text, return_tensors='pt')
        max_length = self.model.config.n_positions
        stride = 512

        nlls = []
        for i in range(0, encodings.input_ids.size(1), stride):
            begin_loc = max(i + stride - max_length, 0)
            end_loc = min(i + stride, encodings.input_ids.size(1))
            trg_len = end_loc - i

            input_ids = encodings.input_ids[:, begin_loc:end_loc]
            target_ids = input_ids.clone()
            target_ids[:, :-trg_len] = -100  # ignore prefix

            with torch.no_grad():
                outputs = self.model(input_ids, labels=target_ids)
                neg_log_likelihood = outputs.loss * trg_len

            nlls.append(neg_log_likelihood)

        return torch.exp(torch.stack(nlls).sum() / end_loc).item()

    def is_high_quality(self, text, threshold=500):
        """Filter by perplexity. threshold=500 keeps ~60% of web data."""
        return self.calculate_perplexity(text) < threshold

# Wikipedia ~100 · spam >1000

from transformers import pipeline

class ConstitutionalAIFilter:
    """Applies HHH (Helpful, Harmless, Honest) principles at data level."""

    def __init__(self):
        self.harmless_classifier = pipeline(
            "text-classification",
            model="anthropic/harmlessness-classifier",
        )

    def check_helpfulness(self, text):
        helpful_indicators = [
            'how to', 'step-by-step', 'tutorial',
            'explanation', 'guide', 'tips',
        ]
        return any(ind in text.lower() for ind in helpful_indicators)

    def check_harmlessness(self, text):
        result = self.harmless_classifier(text)[0]
        harmful_categories = [
            'violence', 'hate_speech', 'self_harm',
            'illegal_activity', 'sexual_content',
            'misinformation', 'manipulation',
        ]
        for category in harmful_categories:
            if result['label'] == category and result['score'] > 0.7:
                return False, category
        return True, None

    def check_honesty(self, text):
        import re
        dishonest_patterns = [
            r'guaranteed to', r'100% effective',
            r'miracle cure', r'secret',
            r'doctors hate', r'one weird trick',
        ]
        return not any(
            re.search(p, text, re.IGNORECASE) for p in dishonest_patterns
        )

    def filter_document(self, text):
        helpful = self.check_helpfulness(text)
        harmless, harm_type = self.check_harmlessness(text)
        honest = self.check_honesty(text)

        if not harmless:
            return {'include': False, 'reason': f'Harmful: {harm_type}'}

        hhh_score = sum([helpful, harmless, honest])
        return {
            'include': hhh_score >= 2,
            'hhh_score': hhh_score,
            'helpful': helpful, 'harmless': harmless, 'honest': honest,
        }

# 90% reduction in harmful outputs

from sentence_transformers import SentenceTransformer
import numpy as np

class BiasDetector:
    """Detect and mitigate demographic bias in training data."""

    def __init__(self):
        self.model = SentenceTransformer('all-MiniLM-L6-v2')

    def detect_demographic_bias(self, documents, demographics):
        counts = {demo: 0 for demo in demographics}
        for doc in documents:
            for demo in demographics:
                if demo.lower() in doc.lower():
                    counts[demo] += 1

        total = sum(counts.values())
        ratios = {demo: c / total for demo, c in counts.items()}

        expected_ratio = 1.0 / len(demographics)
        return {
            demo: ratio
            for demo, ratio in ratios.items()
            if ratio > 2 * expected_ratio or ratio < 0.5 * expected_ratio
        }

    def balance_dataset(self, documents, target_demo, current_ratio, target_ratio):
        if current_ratio < target_ratio:
            gap = target_ratio - current_ratio
            synthetic_count = int(len(documents) * gap)
            print(f"Generating {synthetic_count} synthetic examples for {target_demo}")
            synthetic = self.generate_synthetic_examples(
                topic=target_demo, count=synthetic_count,
            )
            return documents + synthetic
        return documents