Text Classifier in Machine Learning

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.model_selection import cross_val_score
from sklearn.metrics import classification_report
import numpy as np

# Sample data
texts = [
    "Get rich quick! Buy now!",
    "Meeting scheduled for tomorrow at 3 PM",
    "Congratulations! You won a free iPhone",
    "Please review the attached quarterly report",
    "Limited time offer - 90% discount!!!",
    "Team standup moved to 10 AM",
]
labels = ["spam", "ham", "spam", "ham", "spam", "ham"]

# Build pipeline
pipeline = Pipeline([
    ('tfidf', TfidfVectorizer(
        max_features=50000,
        ngram_range=(1, 3),       # unigrams, bigrams, trigrams
        sublinear_tf=True,         # apply log normalization
        min_df=2,                  # ignore very rare terms
        strip_accents='unicode',
    )),
    ('clf', LogisticRegression(
        C=1.0,
        class_weight='balanced',   # handle class imbalance
        max_iter=1000,
        solver='lbfgs',
    ))
])

# Cross-validation
scores = cross_val_score(pipeline, texts, labels, cv=3, scoring='f1_macro')
print(f"Macro F1: {np.mean(scores):.3f} (+/- {np.std(scores):.3f})")

# Train and evaluate
pipeline.fit(texts, labels)
preds = pipeline.predict(texts)
print(classification_report(labels, preds))

from transformers import (
    AutoTokenizer,
    AutoModelForSequenceClassification,
    TrainingArguments,
    Trainer,
)
from datasets import load_dataset
import numpy as np
from sklearn.metrics import f1_score, accuracy_score

# Load dataset (example: AG News - 4 topic classes)
dataset = load_dataset("ag_news")
label_names = dataset["train"].features["label"].names  # World, Sports, Business, Sci/Tech

# Load tokenizer and model
model_name = "bert-base-uncased"  # or "answerdotai/ModernBERT-base" for ModernBERT
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(
    model_name,
    num_labels=len(label_names),
    problem_type="single_label_classification",
)

# Tokenize
def tokenize_fn(examples):
    return tokenizer(
        examples["text"],
        padding="max_length",
        truncation=True,
        max_length=256,
    )

tokenized = dataset.map(tokenize_fn, batched=True, remove_columns=["text"])

# Metrics
def compute_metrics(eval_pred):
    logits, labels = eval_pred
    preds = np.argmax(logits, axis=-1)
    return {
        "accuracy": accuracy_score(labels, preds),
        "f1_macro": f1_score(labels, preds, average="macro"),
        "f1_weighted": f1_score(labels, preds, average="weighted"),
    }

# Training arguments
training_args = TrainingArguments(
    output_dir="./results",
    num_train_epochs=3,
    per_device_train_batch_size=32,
    per_device_eval_batch_size=64,
    learning_rate=2e-5,             # standard for BERT fine-tuning
    weight_decay=0.01,
    warmup_ratio=0.1,               # 10% warmup
    eval_strategy="epoch",
    save_strategy="epoch",
    load_best_model_at_end=True,
    metric_for_best_model="f1_macro",
    fp16=True,                       # mixed precision for speed
    logging_steps=100,
    report_to="wandb",               # optional: W&B logging
)

# Train
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=tokenized["train"],
    eval_dataset=tokenized["test"],
    compute_metrics=compute_metrics,
)
trainer.train()

# Evaluate
results = trainer.evaluate()
print(f"Test F1 (macro): {results['eval_f1_macro']:.4f}")

from setfit import SetFitModel, Trainer, TrainingArguments, sample_dataset
from datasets import load_dataset
from sklearn.metrics import classification_report

# Load dataset and sample few-shot examples
dataset = load_dataset("SetFit/sst2")  # sentiment: positive/negative
train_dataset = sample_dataset(
    dataset["train"],
    label_column="label",
    num_samples=8,  # only 8 examples per class!
)
eval_dataset = dataset["validation"]

# Initialize SetFit model
model = SetFitModel.from_pretrained(
    "sentence-transformers/paraphrase-MiniLM-L6-v2",
    labels=["negative", "positive"],
)

# Training arguments
args = TrainingArguments(
    batch_size=16,
    num_epochs=1,            # contrastive learning phase
    num_iterations=20,       # number of text pairs per step
    body_learning_rate=1e-5,
    head_learning_rate=1e-2,
)

# Train
trainer = Trainer(
    model=model,
    args=args,
    train_dataset=train_dataset,
    eval_dataset=eval_dataset,
)
trainer.train()

# Evaluate
metrics = trainer.evaluate(eval_dataset)
print(f"Accuracy: {metrics['accuracy']:.4f}")

# Inference
predictions = model.predict([
    "This movie was absolutely fantastic!",
    "Terrible waste of time.",
    "The food was okay, nothing special.",
])
print(predictions)  # ["positive", "negative", "negative"]

from transformers import pipeline

# Initialize zero-shot classifier
classifier = pipeline(
    "zero-shot-classification",
    model="facebook/bart-large-mnli",  # or "MoritzLaurer/deberta-v3-large-zeroshot-v2.0"
    device=0,  # GPU
)

# Single-label classification
result = classifier(
    "The new iPhone 16 Pro has an amazing camera system",
    candidate_labels=["technology", "sports", "politics", "entertainment", "business"],
)
print(f"Label: {result['labels'][0]}, Score: {result['scores'][0]:.3f}")
# Label: technology, Score: 0.912

# Multi-label classification
result = classifier(
    "The CEO announced record profits while laying off 500 workers",
    candidate_labels=["business", "employment", "ethics", "finance"],
    multi_label=True,
)
for label, score in zip(result['labels'], result['scores']):
    print(f"{label}: {score:.3f}")
# business: 0.934
# finance: 0.821
# employment: 0.789
# ethics: 0.612

# Multilingual example (Hindi)
result = classifier(
    "प्रधानमंत्री ने नई शिक्षा नीति की घोषणा की",
    candidate_labels=["politics", "education", "sports", "technology"],
)
print(f"Label: {result['labels'][0]}")
# Label: education

import torch
import torch.nn as nn
from transformers import AutoModel, AutoTokenizer

class FocalLoss(nn.Module):
    """Focal loss for handling class imbalance in multi-label classification."""
    def __init__(self, alpha=0.25, gamma=2.0):
        super().__init__()
        self.alpha = alpha
        self.gamma = gamma

    def forward(self, logits, targets):
        bce_loss = nn.functional.binary_cross_entropy_with_logits(
            logits, targets, reduction='none'
        )
        probs = torch.sigmoid(logits)
        p_t = probs * targets + (1 - probs) * (1 - targets)
        focal_weight = self.alpha * (1 - p_t) ** self.gamma
        return (focal_weight * bce_loss).mean()

class MultiLabelClassifier(nn.Module):
    def __init__(self, model_name, num_labels, dropout=0.1):
        super().__init__()
        self.encoder = AutoModel.from_pretrained(model_name)
        hidden_size = self.encoder.config.hidden_size
        self.classifier = nn.Sequential(
            nn.Dropout(dropout),
            nn.Linear(hidden_size, hidden_size // 2),
            nn.GELU(),
            nn.Dropout(dropout),
            nn.Linear(hidden_size // 2, num_labels),
        )
        self.loss_fn = FocalLoss(alpha=0.25, gamma=2.0)

    def forward(self, input_ids, attention_mask, labels=None):
        outputs = self.encoder(input_ids=input_ids, attention_mask=attention_mask)
        cls_output = outputs.last_hidden_state[:, 0, :]  # [CLS] token
        logits = self.classifier(cls_output)

        loss = None
        if labels is not None:
            loss = self.loss_fn(logits, labels.float())

        return {"loss": loss, "logits": logits}

# Usage
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")
model = MultiLabelClassifier("bert-base-uncased", num_labels=10)

# Inference
text = "Breaking: Tech giant announces layoffs amid strong quarterly earnings"
inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)
with torch.no_grad():
    output = model(**inputs)
    probs = torch.sigmoid(output["logits"])
    # Apply threshold (0.5 default, tune per-class in production)
    predicted_labels = (probs > 0.5).int()
    print(f"Probabilities: {probs}")
    print(f"Predicted labels: {predicted_labels}")

import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import f1_score

def uncertainty_sampling(model, X_pool, n_samples=50):
    """Select samples where the model is most uncertain (closest to 0.5)."""
    probs = model.predict_proba(X_pool)
    # Entropy-based uncertainty
    entropy = -np.sum(probs * np.log(probs + 1e-10), axis=1)
    uncertain_indices = np.argsort(entropy)[-n_samples:]
    return uncertain_indices

def active_learning_loop(
    X_labeled, y_labeled, X_pool, y_pool_oracle,
    X_test, y_test,
    n_iterations=10, samples_per_iteration=50
):
    """Run an active learning loop with uncertainty sampling."""
    vectorizer = TfidfVectorizer(max_features=10000, ngram_range=(1, 2))
    results = []

    for iteration in range(n_iterations):
        # Fit vectorizer on all available text
        all_text = list(X_labeled) + list(X_pool)
        vectorizer.fit(all_text)

        X_train_vec = vectorizer.transform(X_labeled)
        X_pool_vec = vectorizer.transform(X_pool)
        X_test_vec = vectorizer.transform(X_test)

        # Train model
        model = LogisticRegression(class_weight='balanced', max_iter=1000)
        model.fit(X_train_vec, y_labeled)

        # Evaluate
        preds = model.predict(X_test_vec)
        f1 = f1_score(y_test, preds, average='macro')
        results.append({
            'iteration': iteration,
            'labeled_samples': len(X_labeled),
            'f1_macro': f1
        })
        print(f"Iter {iteration}: {len(X_labeled)} samples, F1={f1:.3f}")

        if len(X_pool) == 0:
            break

        # Select most uncertain samples
        uncertain_idx = uncertainty_sampling(model, X_pool_vec, samples_per_iteration)

        # Simulate human labeling (in production, send to annotation UI)
        X_new = [X_pool[i] for i in uncertain_idx]
        y_new = [y_pool_oracle[i] for i in uncertain_idx]

        # Add to labeled set, remove from pool
        X_labeled = list(X_labeled) + X_new
        y_labeled = list(y_labeled) + y_new
        X_pool = [x for i, x in enumerate(X_pool) if i not in uncertain_idx]
        y_pool_oracle = [y for i, y in enumerate(y_pool_oracle) if i not in uncertain_idx]

    return results

# Example usage
# results = active_learning_loop(X_seed, y_seed, X_unlabeled, y_oracle, X_test, y_test)

# Text Classifier Training Configuration (YAML)
model:
  architecture: bert-base-uncased  # or answerdotai/ModernBERT-base
  num_labels: 5
  problem_type: single_label_classification  # or multi_label_classification
  max_seq_length: 256
  dropout: 0.1

training:
  epochs: 3
  batch_size: 32
  learning_rate: 2e-5
  weight_decay: 0.01
  warmup_ratio: 0.1
  fp16: true
  gradient_accumulation_steps: 1
  early_stopping_patience: 3
  metric_for_best_model: f1_macro

data:
  train_path: ./data/train.jsonl
  eval_path: ./data/eval.jsonl
  test_path: ./data/test.jsonl
  text_column: text
  label_column: label
  class_weights: balanced  # or custom: [1.0, 2.5, 1.0, 3.0, 1.0]

serving:
  engine: onnx  # or torchscript, triton
  quantization: int8  # reduces model size 4x, ~5% F1 drop
  confidence_threshold: 0.85
  fallback: human_review
  max_batch_size: 64
  timeout_ms: 100