Разнообразие нейронных сетей: Обзор основных задач

from typing import Dict, Optional           # Type hints для читаемости кода
import datetime                             # Модуль с функциями для работы с датой и временем

import numpy as np                          # Численные вычисления, массивы
import pandas as pd                         # Табличные данные, DataFrame
import matplotlib.pyplot as plt             # Визуализация графиков и изображений

import torch                                # Основной фреймворк PyTorch
import torch.nn as nn                       # Нейронные сети (модели, слои)
import torch.optim as optim                 # Оптимизаторы
from torch.utils.data import DataLoader, Dataset as TorchDataset  # Поставщик батчей, контейнер данных

import torchvision                          # Обработка изображений
from torchvision import transforms  # Нормализация, ToTensor, аугментации

from datasets import load_dataset, Dataset as HFDataset  # HuggingFace датасеты
from transformers import AutoTokenizer      # Токенизаторы

from sklearn.datasets import load_iris      # Пример датасета (iris)
from sklearn.model_selection import train_test_split  # Разделение train/val
from sklearn.preprocessing import StandardScaler     # Нормализация фич

device = (torch.device('cuda') if torch.cuda.is_available()
          else torch.device('cpu'))
print(f"Training and evaluating on device {device}.")

class ModelEvaluator:
    """
    Универсальный оценщик PyTorch моделей на датасетах.
    Автоматически вычисляет accuracy для train/val выборок.
    """

    def __init__(self, model: nn.Module, device: torch.device, task_type: str):
        """
        Инициализация ModelEvaluator.

        Args:
            model: PyTorch модель для оценки
            device: Устройство вычислений ('cuda' или 'cpu')
            task_type: Тип задачи ('CV', 'NLP', 'TABULAR')
        """
        self.model = model
        self.device = device
        self.task_type = task_type

        self.model.to(self.device)

        available_task_types = ['CV', 'NLP', 'TABULAR']
        if self.task_type not in available_task_types:
            raise ValueError(f"task_type can be only: {available_task_types}")

    def evaluate_dataset(self, loader: DataLoader) -> Dict[str, float]:
        """
        Оценка модели на датасете. Возвращает словарь метрик.

        Args:
            loader: DataLoader с данными

        Returns:
            Метрики, пример: {"accuracy": 0.95}
        """
        self.model.eval()
        correct = 0
        total = 0

        with torch.no_grad():
            for batch in loader:
                if self.task_type in ['CV', 'TABULAR']:
                    inputs, target = batch
                    inputs = inputs.to(self.device)
                    target = target.to(self.device)
                    outputs = self.model(inputs)
                elif self.task_type == 'NLP':
                    inputs = batch['input_ids'].to(self.device)
                    target = batch['label'].to(self.device)
                    attention_mask = batch['attention_mask'].to(self.device)
                    outputs = self.model(inputs, attention_mask)

                _, predicted = torch.max(outputs, dim=1)
                correct += (predicted == target).sum().item()
                total += target.size(0)

        accuracy: float = correct / total
        return {"accuracy": accuracy}

    def evaluate(self, train_loader: DataLoader, val_loader: DataLoader) -> Dict[str, Dict[str, float]]:
        """
        Полная оценка: train + val выборки одним вызовом.

        Args:
            train_loader: Обучающая выборка
            val_loader: Валидационная выборка

        Returns:
            Пример: {"train": {"accuracy": 0.90}, "val": {"accuracy": 0.95}}
        """
        print(f"Evaluating on device {self.device}.")
        metrics = {}
        metrics["train"] = self.evaluate_dataset(train_loader)
        metrics["val"] = self.evaluate_dataset(val_loader)
        return metrics

class Trainer:
    """
    Универсальный класс для обучения PyTorch моделей.
    """

    def __init__(
        self,
        model: nn.Module,
        optimizer: optim.Optimizer,
        loss_fn: nn.Module,
        train_loader: DataLoader,
        device: torch.device,
        task_type: str,
        clip_grad_norm: Optional[float] = None,
        print_interval: int = 10,
    ):
        """
        Инициализация Trainer.

        Args:
            model: Нейронная сеть для обучения
            optimizer: Оптимизатор
            loss_fn: Функция потерь
            train_loader: Данные для обучения
            device: Устройство для тренировки: 'cuda' или 'cpu'
            task_type: Тип задачи: 'CV', 'TABULAR' или 'NLP'
            clip_grad_norm: Максимальная норма градиента (защита от взрыва)
            print_interval: Печатать loss каждые N эпох
        """
        self.model = model
        self.optimizer = optimizer
        self.loss_fn = loss_fn
        self.train_loader = train_loader
        self.device = device
        self.task_type = task_type
        self.clip_grad_norm = clip_grad_norm
        self.print_interval = print_interval

        self.model.to(self.device)

        available_task_types = ['CV','NLP','TABULAR']
        if self.task_type not in available_task_types:
            raise ValueError(f"task_type can be only: {available_task_types}")

    def train_one_batch(self, inputs: torch.Tensor, target: torch.Tensor, attention_mask: Optional[torch.Tensor] = None) -> float:
        """
        Обучение на одном батче.

        Args:
            inputs: Входные данные
            target: Целевые метки
            attention_mask: Маска внимания для NLP (опционально)

        Returns:
            Значение loss для батча
        """
        self.model.train()

        self.optimizer.zero_grad(set_to_none=True)

        if attention_mask is None:
            outputs = self.model(inputs)
        else:
            outputs = self.model(inputs, attention_mask=attention_mask)
        loss = self.loss_fn(outputs, target)

        loss.backward()

        if self.clip_grad_norm is not None:
            torch.nn.utils.clip_grad_norm_(self.model.parameters(), self.clip_grad_norm)

        self.optimizer.step()

        return loss.item()

    def train_epoch(self, epoch_num: int = 0) -> float:
        """
        Тренировка модели на одной эпохе.

        Args:
            epoch_num: Номер эпохи (для логирования)

        Returns:
            Средний loss по эпохе
        """
        self.model.train()
        loss_train = 0.0
        num_batches = len(self.train_loader)
        for batch in self.train_loader:
            if self.task_type in ['CV','TABULAR']:
                inputs, target = batch
                inputs = inputs.to(self.device)
                target = target.to(self.device)
                loss_train += self.train_one_batch(inputs, target)
            elif self.task_type == 'NLP':
                inputs = batch['input_ids'].to(self.device)
                target = batch['label'].to(self.device)
                attention_mask = batch['attention_mask'].to(self.device)
                loss_train += self.train_one_batch(inputs, target, attention_mask)

        return loss_train / num_batches

    def training_loop(self, n_epochs: int) -> None:
        """
        Основной цикл обучения.

        Args:
            n_epochs: Количество эпох
        """

        print(f"Training on device {self.device}.")

        for epoch in range(1, n_epochs + 1):
            avg_loss = self.train_epoch(epoch)

            if epoch == 1 or epoch % self.print_interval == 0:
                print(f"{datetime.datetime.now()} Epoch {epoch}, Training loss: {avg_loss:.4f}")

def compute_mean_and_std(loader: DataLoader, dim: list, device="cpu") -> tuple[torch.Tensor, torch.Tensor]:
    """
    Вычисление среднего значения и стандартного отклонения в наборе данных.

    Args:
        loader (DataLoader): DataLoader для итерации по набору данных
        dim (list): Размерности, по которым будут считатьcя значения
        device (str): Устройство, на котором выполняются вычисления ("cpu" или "cuda").

    Returns:
        tuple[torch.Tensor, torch.Tensor]: Кортеж, содержащий тензоры среднего значения и стандартного отклонения.
    """
    # 1 Создаем список из всех тензоров
    inputs_list = []
    for inputs, _ in loader:
        inputs_list.append(inputs)

    # 2 Соединяем все тензоры в один большой тензор
    all_inputs = torch.cat(inputs_list, dim=0).float().to(device)  # Преобразуем в float и перемещаем на устройство

    # 3 Вычисляем mean и std
    mean = torch.mean(all_inputs, dim=dim)
    std = torch.std(all_inputs, dim=dim)

    return mean, std

# 1 Создаем словарь с классами
class_names = ['T-shirt/top', 'Trouser', 'Pullover', 'Dress', 'Coat',
               'Sandal', 'Shirt', 'Sneaker', 'Bag', 'Ankle boot']
class_names_dict = dict(zip(range(len(class_names)), class_names))

# 2 Загрузка и подготовка данных (без нормализации), получение Mean и Std
trainset = torchvision.datasets.FashionMNIST(root='./data', train=True, download=True,
                                             transform=transforms.Compose([
                                                 transforms.ToTensor()
                                             ]))
# 2.1 создаем dataloader
dataloader = DataLoader(trainset, batch_size=64, shuffle=False)
# 2.2 считаем mean и std
mean, std = compute_mean_and_std(dataloader, [0, 2, 3])
print(f"Mean: {mean}")
print(f"Std Dev: {std}")

# 3 Загрузка и подготовка данных (с нормализацией)
trainset = torchvision.datasets.FashionMNIST(
    root='./data', train=True, download=True,
    transform=transforms.Compose([
        transforms.ToTensor(),
        transforms.Normalize(mean, std)
    ]))  # Тренировочный набор
valset = torchvision.datasets.FashionMNIST(
    root='./data', train=False, download=True,
    transform=transforms.Compose([
        transforms.ToTensor(),
        transforms.Normalize(mean, std)
    ]))  # Валидационный набор

# 4 Создаем dataloders
batch_size = 256
train_loader = DataLoader(trainset, batch_size=batch_size,
                          shuffle=True)
val_loader = DataLoader(valset, batch_size=batch_size,
                        shuffle=False)


# 5 Определяем архитектуру нейронной сети
class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = nn.Conv2d(1, 28, kernel_size=3, padding=1)
        self.conv2 = nn.Conv2d(28, 56, kernel_size=3, padding=1)
        self.dropout1 = nn.Dropout(0.1)
        self.dropout2 = nn.Dropout(0.2)
        self.fc1 = nn.Linear(56 * 14 * 14, 112)
        self.fc2 = nn.Linear(112, 10)

    def forward(self, x):
        x = nn.functional.relu(self.conv1(x))
        x = nn.functional.relu(self.conv2(x))
        x = nn.functional.max_pool2d(self.dropout1(x), 2)
        x = torch.flatten(x, 1)
        x = nn.functional.relu(self.fc1(self.dropout2(x)))
        x = self.fc2(x)
        return x


# 6 Настройка модели, оптимизатора и функции потерь
model = Net().to(device)
optimizer = optim.Adam(model.parameters(), lr=1e-2)
loss_fn = nn.CrossEntropyLoss()

# 7.1 Создание и настройка Trainer
task_type = 'CV'
trainer = Trainer(
    model=model,
    optimizer=optimizer,
    loss_fn=loss_fn,
    train_loader=train_loader,
    device=device,
    task_type=task_type,
    clip_grad_norm=1.0,
    print_interval=1,
)
# 7.2 Запуск процесса обучения
trainer.training_loop(n_epochs=10)

# 8 Оценка обученной модели
evaluator = ModelEvaluator(model, device, task_type)
metrics = evaluator.evaluate(train_loader, val_loader)
print(metrics)

number = 10
data, label = valset[number]
plt.imshow(data.squeeze(), cmap='gray')
plt.show()

# predict
model.eval()
outputs = model(data.to(device).unsqueeze(0))
_, predicted = torch.max(outputs, dim=1)

# validate
real_answer = class_names_dict[label]
model_answer = class_names_dict[predicted.item()]
if real_answer == model_answer:
    result = 'correct'
else:
    result = 'incorrect'

print(f"""
Real answer: {real_answer}
Model answer: {model_answer}

Result:
Model response is {result}
""")

# 1 Создаем словарь с классами
class_names = ['Negative', 'Positive']
class_names_dict = dict(zip(range(len(class_names)), class_names))

def tokenize_dataset(dataset):
    # Инициализация токенизатора (использовал BERT)
    tokenizer = AutoTokenizer.from_pretrained('bert-base-uncased')

    # Функция для токенизации и подготовки данных
    def tokenize_function(examples):
        return tokenizer(examples["text"], padding="max_length", truncation=True, max_length=256)

    #  Токенизация всего датасета
    tokenized_dataset = dataset.map(tokenize_function, batched=True)

    tokenized_dataset.set_format("torch", columns=["input_ids", "attention_mask", "label"])
    return tokenized_dataset

# 2 Загрузка данных и создание Dataloader
batch_size = 64
# 2.1 Train
trainset = load_dataset("imdb", split="train", cache_dir="./data")
train_loader = DataLoader(tokenize_dataset(trainset), batch_size=batch_size, shuffle=True)
# 2.2 Test
valset = load_dataset("imdb", split="test", cache_dir="./data")
val_loader = DataLoader(tokenize_dataset(valset), batch_size=batch_size, shuffle=False)

# 3 Определение модели (LSTM)
class LSTMClassifier(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, num_classes, lstm_layers, dropout):
        super(LSTMClassifier, self).__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim)
        self.lstm = nn.LSTM(embedding_dim, hidden_dim, num_layers=lstm_layers, batch_first=True, dropout=dropout,
                            bidirectional=True)
        self.dropout = nn.Dropout(dropout)
        self.linear = nn.Linear(hidden_dim * 2, num_classes)

    def forward(self, input_ids, attention_mask):
        # attention_mask присутствует, но не используется в LSTM

        embedded = self.embedding(input_ids)
        lstm_out, _ = self.lstm(embedded)

        pooled_output = torch.mean(lstm_out, dim=1)

        pooled_output = self.dropout(pooled_output)
        logits = self.linear(pooled_output)
        return logits

# 4 Параметры модели
vocab_size = AutoTokenizer.from_pretrained('bert-base-uncased').vocab_size
embedding_dim = 128
hidden_dim = 64
num_classes = 2
lstm_layers = 2
dropout = 0.5

# 5 Настройка модели, оптимизатора и функции потерь
model = LSTMClassifier(vocab_size, embedding_dim, hidden_dim, num_classes, lstm_layers, dropout).to(device)
optimizer = optim.AdamW(model.parameters(), lr=1e-4, weight_decay=1e-2)
loss_fn = nn.CrossEntropyLoss()

# 6.1 Создание и настройка тренера (Trainer)
task_type = 'NLP'
trainer = Trainer(
    model=model,
    optimizer=optimizer,
    loss_fn=loss_fn,
    train_loader=train_loader,
    device=device,
    task_type=task_type,
    clip_grad_norm=1.0,
    print_interval=1,
)
# 6.2 Запуск процесса обучения
trainer.training_loop(n_epochs=10)

# 7 Оценка обученной модели
evaluator = ModelEvaluator(model, device, task_type)
metrics = evaluator.evaluate(train_loader, val_loader)
print(metrics)

number = 49
data = valset[number]
print(data['text'])
dataset_dict = {
    "text": [data["text"]],
    "label": [data["label"]]}
single_dataset = HFDataset.from_dict(dataset_dict)
single_dataset = tokenize_dataset(single_dataset)

inputs = single_dataset['input_ids'].to(device)
target = single_dataset['label'].to(device)
attention_mask = single_dataset['attention_mask'].to(device)

# predict
model.eval()
outputs = model(inputs, attention_mask)
_, predicted = torch.max(outputs, dim=1)

# validate
real_answer = class_names_dict[int(target[0])]
model_answer = class_names_dict[int(predicted[0])]
if real_answer == model_answer:
    result = 'correct'
else:
    result = 'incorrect'

print(f"""
Real answer: {real_answer}
Model answer: {model_answer}

Result:
Model response is {result}
""")

# 1 Создаем словарь с классами
class_names = ['setosa', 'versicolor', 'virginica']
class_names_dict = dict(zip(range(len(class_names)), class_names))

# 2 Загрузка и подготовка данных
iris = load_iris()
iris_data = iris['data']
iris_target = iris['target']

# 3 Разделение на train и val
train_data, val_data, train_target, val_target = train_test_split(
    iris_data, iris_target, test_size=0.2, random_state=42
)

# 4 Нормализация данных (StandardScaler)
scaler = StandardScaler()
train_data_normalized = scaler.fit_transform(train_data)
val_data_normalized = scaler.transform(val_data)

# 5 Создание кастомного Dataset
class IrisDataset(TorchDataset):
    def __init__(self, X, y, transform=None):
        self.X = X.astype(np.float32)
        self.y = y.astype(np.int64)
        self.transform = transform

    def __len__(self):
        return len(self.X)

    def __getitem__(self, idx):
        sample = self.X[idx]
        label = self.y[idx]

        if self.transform:
            sample = self.transform(sample)

        return sample, label


# 6 Создание Dataset
trainset = IrisDataset(train_data_normalized, train_target)
valset = IrisDataset(val_data_normalized, val_target)

# 7 Создание DataLoaders
batch_size = 32
train_loader = DataLoader(trainset, batch_size=batch_size, shuffle=True)
val_loader = DataLoader(valset, batch_size=batch_size, shuffle=False)

# 8 Определяем архитектуру нейронной сети
class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.fc1 = nn.Linear(4, 8)
        self.fc2 = nn.Linear(8, 16)
        self.fc3 = nn.Linear(16, 3)
        self.dropout = nn.Dropout(0.1)

    def forward(self, x):
        x = nn.functional.relu(self.fc1(x))
        x = self.dropout(x)
        x = nn.functional.relu(self.fc2(x))
        x = self.dropout(x)
        x = self.fc3(x)
        return x


# 9 Настройка модели, оптимизатора и функции потерь
model = Net().to(device)
optimizer = optim.Adam(model.parameters(), lr=1e-2)
loss_fn = nn.CrossEntropyLoss()

# 10.1 Создание и настройка тренера (Trainer)
task_type = 'TABULAR'
trainer = Trainer(
    model=model,
    optimizer=optimizer,
    loss_fn=loss_fn,
    train_loader=train_loader,
    device=device,
    task_type=task_type,
    clip_grad_norm=1.0,
    print_interval=1,
)
# 10.2 Запуск процесса обучения
trainer.training_loop(n_epochs=10)

# 11 Оценка обученной модели
evaluator = ModelEvaluator(model, device, task_type)
metrics = evaluator.evaluate(train_loader, val_loader)
print(metrics)

number = 10
data, label = valset[number]
df = pd.DataFrame([data], columns=iris['feature_names'])
print(df)

# predict
model.eval()
with torch.no_grad():
    data = torch.from_numpy(data).float()
    outputs = model(data.to(device).unsqueeze(0))
    _, predicted = torch.max(outputs, dim=1)

# validate
real_answer = class_names_dict[label]
model_answer = class_names_dict[predicted.item()]
if real_answer == model_answer:
    result = 'correct'
else:
    result = 'incorrect'

print(f"""
Real answer: {real_answer}
Model answer: {model_answer}

Result:
Model response is {result}
""")