Почему моя обученная модель не распознает новые изображения, которые отличаются от тестового набора данных?

Я обучал модель для классификации различных изображений по разным меткам, вначале 2, и если бы это хорошо работало, я добавил бы больше меток для обучения. Я получил хорошие результаты по точности и прецизионности в ходе обучения, проверяя модель во время тренировки с тестовым набором данных, однако когда я пытаюсь протестировать модель с новыми изображениями, которые не были частью обучающего или тестового набора данных.

Это код, который я использовал для обучения модели. Я использовал efficientnet_v2-m, так как мой набор данных состоит из 3.5k изображений, разбитых на 2 метки, то есть примерно 1.7k для каждой метки.

Я просто не понимаю, почему моя обученная модель не распознает новые изображения, которые принадлежат тому же набору изображений, использовавшихся для обучения, но они разные, что имеет смысл, так как мне не следует тестировать модель на тех же изображениях, которые я использовал для обучения.

import os
import csv
import copy
import numpy as np
import torch
from torch import nn
from torch.optim import Adam
from torch.optim.lr_scheduler import ReduceLROnPlateau
from torchvision import datasets, transforms
from torchvision.models import efficientnet_v2_m
from torch.utils.data import random_split, DataLoader
from sklearn.metrics import accuracy_score, confusion_matrix, precision_score

# ---------------------------
# Конфигурация и Настройка
# ---------------------------

# Директория для сохранения моделей и логов
save_dir = os.path.expanduser("~/Desktop/Coding/Models")
if not os.path.exists(save_dir):
    os.makedirs(save_dir)

# CSV файл для сохранения метрик обучения
csv_file = os.path.join(save_dir, "training_metrics.csv")

# Инициализация CSV файла с заголовками столбцов (если он не существует)
def initialize_csv():
    if not os.path.exists(csv_file):
        with open(csv_file, mode="w", newline="") as file:
            writer = csv.writer(file)
            writer.writerow([
                "Эпоха", "Потеря обучения", "Потеря валидации", 
                "Точность валидации", "Прецизионность валидации", "Скорость обучения"
            ])

# Конфигурация устройства (используя M1 Pro GPU или CPU)
device = torch.device("mps" if torch.backends.mps.is_available() else "cpu")
print(f"Используемое устройство: {device}")

# ---------------------------
# Загрузка данных и Предобработка
# ---------------------------

# Определение преобразований нормализации и аугментации
train_transform = transforms.Compose([
    transforms.Resize((600, 1200)),  # Изменение размера до 1200x600 пикселей
    transforms.RandomHorizontalFlip(),  # Аугментация данных: случайное горизонтальное отражение
    transforms.RandomRotation(15),      # Аугментация данных: случайное вращение
    transforms.ColorJitter(brightness=0.2, contrast=0.2, saturation=0.15, hue=0.05),  # Аугментация данных: цветовые искажения
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.5, 0.5, 0.5], std=[0.5, 0.5, 0.5]),
])

# Определение трансформаций для валидации и тестирования
val_test_transform = transforms.Compose([
    transforms.Resize((600, 1200)),  # Изменение размера до 1200x600 пикселей
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.5, 0.5, 0.5], std=[0.5, 0.5, 0.5]),
])

# Загрузка набора данных (организованного по классам в папках)
dataset = datasets.ImageFolder(root="/Users/X/Desktop/FinalDatasets/X", transform=train_transform)

# Разделение набора данных на обучение (70%), валидацию (20%) и тестирование (10%)
dataset_size = len(dataset)
train_size = int(0.7 * dataset_size)
val_size = int(0.2 * dataset_size)
test_size = dataset_size - train_size - val_size

train_dataset, val_dataset, test_dataset = random_split(dataset, [train_size, val_size, test_size])

# Переопределение трансформации для валидационного и тестового наборов данных
val_dataset.dataset.transform = val_test_transform
test_dataset.dataset.transform = val_test_transform

# Создание DataLoader для каждого разбиения
batch_size = 2  # Увеличенный размер пакета для лучшей производительности
num_workers = 0  # Настройте в зависимости от вашей системы
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=num_workers)
val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False, num_workers=num_workers)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False, num_workers=num_workers)

print(f"Размеры набора данных - Обучение: {train_size}, Валидация: {val_size}, Тест: {test_size}")

# ---------------------------
# Определение модели с тонкой настройкой
# ---------------------------

from torchvision.models import efficientnet_v2_m

# Функция для создания модели с тонкой настройкой
def EfficientNetV2_M_custom_finetune(num_classes=2, fine_tune_at_layer=-3):
    model = efficientnet_v2_m(pretrained=True)  # Загрузка предобученной EfficientNetV2-M

    # Заморозить все слои, кроме последних 'fine_tune_at_layer' слоев
    if fine_tune_at_layer != 0:
        for name, param in model.named_parameters():
            # Пример: fine_tune_at_layer=-5 для разморозки последних 5 слоев
            layer_num = int(name.split('.')[1]) if len(name.split('.')) > 1 and name.split('.')[1].isdigit() else -1
            if layer_num < fine_tune_at_layer:
                param.requires_grad = False
    else:
        # Заморозить все слои, кроме классификатора
        for param in model.parameters():
            param.requires_grad = False

    # Изменить классификатор для 2 классов
    model.classifier[1] = nn.Linear(model.classifier[1].in_features, num_classes)

    return model

# Инициализация модели с тонкой настройкой
model = EfficientNetV2_M_custom_finetune(num_classes=2, fine_tune_at_layer=-5).to(device)

# ---------------------------
# Определение пользовательской функции потерь с учётом веса классов
# ---------------------------

class CustomPenaltyLossWithWeights(nn.Module):
    def __init__(self, class_weights):
        super(CustomPenaltyLossWithWeights, self).__init__()
        # Включение весов классов в CrossEntropyLoss
        self.cross_entropy_loss = nn.CrossEntropyLoss(weight=class_weights)
        self.weight_matrix = torch.tensor([
            [0, 1],
            [1, 0],
        ], device=device, dtype=torch.float32)

    def forward(self, outputs, targets):
        # Вычисление базовой CrossEntropyLoss с весами
        base_loss = self.cross_entropy_loss(outputs, targets)

        # Получение предсказанного класса
        predicted_class = torch.argmax(outputs, dim=1)

        # Применение штрафа в зависимости от того, насколько предсказание отличается от истинного класса
        penalties = self.weight_matrix[targets, predicted_class]

        # Смешивание базовой потери со штрафом (среднее значение штрафов)
        loss = base_loss + torch.mean(penalties)

        return loss

# Вычисление весов классов (обратное частотам классов)
class_counts = [1.7e3, 1.7e3]  # [Количество класса 0, Количество класса 1]
total_samples = sum(class_counts)
class_weights = torch.tensor([total_samples / count for count in class_counts], dtype=torch.float32).to(device)

# Инициализация пользовательской функции потерь с весами
criterion = CustomPenaltyLossWithWeights(class_weights)

# ---------------------------
# Настройка оптимизатора и планировщика
# ---------------------------

from torch.optim.lr_scheduler import ReduceLROnPlateau

# Определение групп параметров с разными скоростями обучения
optimizer = Adam([
    {'params': model.features[: -5].parameters(), 'lr': 1e-5},  # Замороженные или низкая скорость обучения
    {'params': model.features[-5:].parameters(), 'lr': 1e-4},  # Тонко настроенные слои
    {'params': model.classifier.parameters(), 'lr': 1e-3}     # Новая добавленная классификатор
])

# Инициализация планировщика скорости обучения
scheduler = ReduceLROnPlateau(optimizer, mode="min", factor=0.5, patience=3)

# ---------------------------
# Функции ведения учета в CSV
# ---------------------------

# Функция для ведения учета деталей обучения в CSV
def log_to_csv(epoch, train_loss, val_loss, val_accuracy, val_precision, learning_rate):
    with open(csv_file, mode="a", newline="") as file:
        writer = csv.writer(file)
        writer.writerow([
            epoch, train_loss, val_loss, 
            val_accuracy * 100, val_precision * 100, learning_rate
        ])

# Функция для сохранения модели после обучения
def save_model(model, epoch):
    model_save_path = os.path.join(save_dir, f"cnn_model_epoch_{epoch+1}.pth")
    torch.save(model.state_dict(), model_save_path)
    print(f"Модель сохранена в {model_save_path}")

# Инициализация CSV лог файла
initialize_csv()

# ---------------------------
# Функция оценки
# ---------------------------

def evaluate_model(model, dataloader):
    model.eval()
    true_labels = []
    predicted_labels = []
    val_loss = 0.0

    with torch.no_grad():
        for images, labels in dataloader:
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)
            loss = criterion(outputs, labels)
            val_loss += loss.item()

            _, predicted = torch.max(outputs, 1)
            true_labels.extend(labels.cpu().numpy())
            predicted_labels.extend(predicted.cpu().numpy())

    # Вычисление точности валидации, прецизионности и матрицы путаницы
    accuracy = accuracy_score(true_labels, predicted_labels)
    precision = precision_score(true_labels, predicted_labels, average="macro")
    cm = confusion_matrix(true_labels, predicted_labels)  # Добавьте эту строку для возврата матрицы путаницы
    return accuracy, val_loss / len(dataloader), precision, cm  # Теперь возвращает четыре значения

# ---------------------------
# Цикл обучения с тонкой настройкой
# ---------------------------

num_epochs = 6  # Общее количество эпох
unfreeze_epochs = [3, 4, 5]  # Эпохи, на которых необходимо разморозить больше слоев

for epoch in range(num_epochs):
    model.train()  # Установите модель в режим обучения
    running_loss = 0.0

    # Этап обучения
    for batch_idx, (images, labels) in enumerate(train_loader, 1):
        images, labels = images.to(device), labels.to(device)

        # Обнуление градиентов
        optimizer.zero_grad()

        # Прямой проход
        outputs = model(images)
        loss = criterion(outputs, labels)
        running_loss += loss.item()

        # Обратный проход и оптимизация
        loss.backward()
        optimizer.step()

        # Печать потерь после каждого пакета
        print(f"Эпоха [{epoch+1}/{num_epochs}], Пакет [{batch_idx}/{len(train_loader)}], Потеря: {loss.item():.4f}")

    # Этап валидации после каждой эпохи
    val_accuracy, val_loss, val_precision, cm = evaluate_model(model, val_loader)
    average_train_loss = running_loss / len(train_loader)
    print(f"Эпоха [{epoch+1}/{num_epochs}] Сводка - Потеря обучения: {average_train_loss:.4f}, "
      f"Потеря валидации: {val_loss:.4f}, Точность валидации: {val_accuracy*100:.2f}%, "
      f"Прецизионность валидации: {val_precision*100:.2f}%")

    # Печать матрицы путаницы
    print("Матрица путаницы (валидация):")
    print(cm)

    # Ведение учета деталей в CSV
    current_lr = optimizer.param_groups[0]['lr']
    log_to_csv(epoch + 1, average_train_loss, val_loss, val_accuracy, val_precision, current_lr)

    # Шаг планировщика скорости обучения на основе потери валидации
    scheduler.step(val_loss)

    # Сохранение модели в конце каждой эпохи
    save_model(model, epoch)

    # Разморозка дополнительных слоев на указанных эпохах
    if epoch + 1 in unfreeze_epochs:
        # Определите количество слоев для разморозки постепенно
        # Эта логика предполагает, что 'features' индексируются; настройте в зависимости от фактического имени слоя
        layers_to_unfreeze = -(5 + (unfreeze_epochs.index(epoch + 1) + 1))
        for param in model.features[layers_to_unfreeze:].parameters():
            param.requires_grad = True
        # Обновите оптимизатор, чтобы включить только что размороженные параметры
        optimizer = Adam([
            {'params': model.features[: layers_to_unfreeze].parameters(), 'lr': 1e-5},
            {'params': model.features[layers_to_unfreeze:].parameters(), 'lr': 1e-4},
            {'params': model.classifier.parameters(), 'lr': 1e-3}
        ])
        print(f"Разморожены дополнительные слои на эпохе {epoch+1}")

print("Обучение завершено!")

# ---------------------------
# Тестирование и Оценка
# ---------------------------

def test_model(model, dataloader):
    model.eval()
    true_labels = []
    predicted_labels = []
    test_loss = 0.0

    with torch.no_grad():
        for images, labels in dataloader:
            images, labels = images.to(device), labels.to(device)
            outputs = model(images)
            loss = criterion(outputs, labels)
            test_loss += loss.item()

            _, predicted = torch.max(outputs, 1)
            true_labels.extend(labels.cpu().numpy())
            predicted_labels.extend(predicted.cpu().numpy())

    # Вычисление точности теста и прецизионности
    accuracy = accuracy_score(true_labels, predicted_labels)
    precision = precision_score(true_labels, predicted_labels, average="macro")
    cm = confusion_matrix(true_labels, predicted_labels)

    print(f"Тестовая точность: {accuracy * 100:.2f}%")
    print(f"Тестовая прецизионность: {precision * 100:.2f}%")
    print("Матрица путаницы:")
    print(cm)

    # Сохранение метрик в CSV файл с уникальным именем
    metrics_file = os.path.join(save_dir, f"test_metrics_epoch_{num_epochs}.csv")
    with open(metrics_file, mode="w", newline="") as file:
        writer = csv.writer(file)
        writer.writerow(["Точность", "Прецизионность"])
        writer.writerow([accuracy * 100, precision * 100])
        writer.writerow([])
        writer.writerow(["Матрица путаницы"])
        for row in cm:
            writer.writerow(row.tolist())

    print(f"Метрики теста сохранены в {metrics_file}")

# Тестирование основной модели после обучения
print("\nТестирование основной модели на тестовом наборе:")
test_model(model, test_loader)

print("Все оценки завершены!")

```