Tùy chỉnh Trainer#

from ultralytics import YOLO
from ultralytics.models.yolo.detect import DetectionTrainer

class CustomTrainer(DetectionTrainer):
    """A custom trainer that extends DetectionTrainer with additional functionality."""

    # Thêm các tùy chỉnh của bạn tại đây

model = YOLO("yolo26n.pt")
model.train(data="coco8.yaml", epochs=10, trainer=CustomTrainer)

import numpy as np

from ultralytics import YOLO
from ultralytics.models.yolo.detect import DetectionTrainer
from ultralytics.utils import LOGGER

class MetricsTrainer(DetectionTrainer):
    """Custom trainer that computes and logs F1 score at the end of each epoch."""

    def validate(self):
        """Run validation and compute per-class F1 scores."""
        metrics, fitness = super().validate()
        if metrics is None:
            return metrics, fitness

        if hasattr(self.validator, "metrics") and hasattr(self.validator.metrics, "box"):
            box = self.validator.metrics.box
            f1_per_class = box.f1
            class_indices = box.ap_class_index
            names = self.validator.names

            mean_f1 = float(np.mean(f1_per_class)) if len(f1_per_class) else 0.0

            LOGGER.info(f"Mean F1 Score: {mean_f1:.4f}")
            per_class_str = [f"{names[i]}: {f1_per_class[j]:.3f}" for j, i in enumerate(class_indices)]
            LOGGER.info(f"Per-class F1: {per_class_str}")

        return metrics, fitness

model = YOLO("yolo26n.pt")
model.train(data="coco8.yaml", epochs=5, trainer=MetricsTrainer)

import numpy as np

from ultralytics import YOLO
from ultralytics.models.yolo.detect import DetectionTrainer

class WeightedTrainer(DetectionTrainer):
    """Detection trainer with hand-picked class-weight ratios."""

    def compute_class_weights(self, class_counts):
        """Return custom per-class weights that the detection loss applies."""
        weights = np.ones_like(class_counts)
        weights[0] = 2.0
        weights[1] = 3.0
        return weights

model = YOLO("yolo26n.pt")
model.train(data="custom.yaml", epochs=10, cls_pw=1.0, trainer=WeightedTrainer)

from ultralytics import YOLO
from ultralytics.models.yolo.detect import DetectionTrainer

class CustomSaveTrainer(DetectionTrainer):
    """Trainer that saves the best model based on mAP@0.5 instead of default fitness."""

    def validate(self):
        """Override fitness to use mAP@0.5 for best model selection."""
        previous_best = self.best_fitness
        metrics, fitness = super().validate()
        if metrics is None:
            return metrics, fitness
        fitness = metrics["metrics/mAP50(B)"]
        self.best_fitness = fitness if previous_best is None else max(previous_best, fitness)
        return metrics, fitness

model = YOLO("yolo26n.pt")
model.train(data="coco8.yaml", epochs=20, trainer=CustomSaveTrainer)

from ultralytics import YOLO
from ultralytics.models.yolo.detect import DetectionTrainer
from ultralytics.utils import LOGGER

FREEZE_EPOCHS = 5

def unfreeze_backbone(trainer):
    """Callback to unfreeze the user-requested layers after FREEZE_EPOCHS."""
    if trainer.epoch == FREEZE_EPOCHS:
        user_freeze = [x for x in trainer.freeze_layer_names if x not in {".dfl", "teacher_model."}]
        LOGGER.info(f"Epoch {trainer.epoch}: Unfreezing requested layers for fine-tuning")
        for name, param in trainer.model.named_parameters():
            if (
                not param.requires_grad
                and ".dfl" not in name
                and "teacher_model." not in name
                and any(x in name for x in user_freeze)
            ):
                param.requires_grad = True
                LOGGER.info(f"  Unfroze: {name}")
        trainer.freeze_layer_names = [x for x in trainer.freeze_layer_names if x not in user_freeze]

class FreezingTrainer(DetectionTrainer):
    """Trainer with backbone freezing for first N epochs."""

    def __init__(self, *args, **kwargs):
        """Initialize and register the unfreeze callback."""
        super().__init__(*args, **kwargs)
        self.add_callback("on_train_epoch_start", unfreeze_backbone)

model = YOLO("yolo26n.pt")
model.train(data="coco8.yaml", epochs=20, freeze=10, trainer=FreezingTrainer)

from ultralytics import YOLO
from ultralytics.models.yolo.detect import DetectionTrainer
from ultralytics.utils import LOGGER
from ultralytics.utils.torch_utils import unwrap_model

class PerLayerLRTrainer(DetectionTrainer):
    """Trainer with different learning rates for backbone and head."""

    backbone_lr_ratio = 0.1

    def build_optimizer(self, model, name="auto", lr=0.001, momentum=0.9, decay=1e-5, iterations=1e5):
        """Reuse the trainer optimizer and lower its backbone parameter-group rates."""
        optimizer = super().build_optimizer(model, name, lr, momentum, decay, iterations)
        unwrapped = unwrap_model(model)
        backbone_len = len(unwrapped.yaml["backbone"])
        backbone = {
            id(p)
            for name, p in unwrapped.named_parameters()
            if any(name.startswith(f"model.{i}.") for i in range(backbone_len))
        }

        groups = []
        for group in optimizer.param_groups:
            head_params = [p for p in group["params"] if id(p) not in backbone]
            backbone_params = [p for p in group["params"] if id(p) in backbone]
            if head_params:
                groups.append({**group, "params": head_params})
            if backbone_params:
                groups.append({**group, "params": backbone_params, "lr": group["lr"] * self.backbone_lr_ratio})
        optimizer.param_groups = groups

        LOGGER.info(f"PerLayerLR: {len(backbone)} backbone params at {self.backbone_lr_ratio}x the head rate")
        return optimizer

model = YOLO("yolo26n.pt")
model.train(data="coco8.yaml", epochs=20, trainer=PerLayerLRTrainer)

from torch import nn

from ultralytics import RTDETR
from ultralytics.models.rtdetr.train import RTDETRTrainer

class SyncBNTrainer(RTDETRTrainer):
    """RT-DETR trainer that converts BatchNorm to SyncBatchNorm for multi-GPU training."""

    def set_model_attributes(self):
        """Run the parent setup, then convert BN to SyncBatchNorm when training on multiple GPUs."""
        super().set_model_attributes()
        if self.world_size > 1:
            self.model = nn.SyncBatchNorm.convert_sync_batchnorm(self.model)

model = RTDETR("rtdetr-l.pt")
model.train(data="coco8.yaml", epochs=20, device=[0, 1], trainer=SyncBNTrainer)

import torch

from ultralytics import RTDETR
from ultralytics.models.rtdetr.train import RTDETRTrainer
from ultralytics.utils.torch_utils import TORCH_2_0

class CustomClipTrainer(RTDETRTrainer):
    """RT-DETR trainer with configurable gradient clipping."""

    clip_grad_norm = 0.1  # norm gradient tối đa; đặt thành 0 để tắt clipping

    def optimizer_step(self):
        """Run an optimizer step with a configurable gradient-norm clip."""
        self.scaler.unscale_(self.optimizer)
        if self.clip_grad_norm > 0:
            kwargs = {"foreach": False} if self.device.type == "npu" and TORCH_2_0 else {}
            torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=self.clip_grad_norm, **kwargs)
        self.scaler.step(self.optimizer)
        self.scaler.update()
        self.optimizer.zero_grad()
        if self.ema:
            self.ema.update(self.model)

model = RTDETR("rtdetr-l.pt")
model.train(data="coco8.yaml", epochs=20, trainer=CustomClipTrainer)