import json import logging import os from pathlib import Path from typing import Any, Dict, Optional, Union import lightning as pl import torch from lightning.pytorch.trainer.states import RunningStage from relik.common.log import get_logger from relik.retriever.callbacks.base import NLPTemplateCallback, PredictionCallback from relik.retriever.pytorch_modules.hf import GoldenRetrieverModel logger = get_logger(__name__, level=logging.INFO) class SavePredictionsCallback(NLPTemplateCallback): def __init__( self, saving_dir: str | os.PathLike | None = None, verbose: bool = False, *args, **kwargs, ): super().__init__() self.saving_dir = saving_dir self.verbose = verbose @torch.no_grad() def __call__( self, trainer: pl.Trainer, pl_module: pl.LightningModule, predictions: Dict, callback: PredictionCallback, *args, **kwargs, ) -> dict: # write the predictions to a file inside the experiment folder if self.saving_dir is None and trainer.logger is None: logger.info( "You need to specify an output directory (`saving_dir`) or a logger to save the predictions.\n" "Skipping saving predictions." ) return {} datasets = callback.datasets for dataloader_idx, dataloader_predictions in predictions.items(): # save to file if self.saving_dir is not None: prediction_folder = Path(self.saving_dir) else: try: prediction_folder = ( Path(trainer.logger.experiment.dir) / "predictions" ) except Exception: logger.info( "You need to specify an output directory (`saving_dir`) or a logger to save the predictions.\n" "Skipping saving predictions." ) return {} prediction_folder.mkdir(exist_ok=True) predictions_path = ( prediction_folder / f"{datasets[dataloader_idx].name}_{dataloader_idx}.json" ) if self.verbose: logger.info(f"Saving predictions to {predictions_path}") with open(predictions_path, "w") as f: for prediction in dataloader_predictions: for k, v in prediction.items(): if isinstance(v, set): prediction[k] = list(v) f.write(json.dumps(prediction) + "\n") class ResetModelCallback(pl.Callback): def __init__( self, question_encoder: str, passage_encoder: str | None = None, verbose: bool = True, ) -> None: super().__init__() self.question_encoder = question_encoder self.passage_encoder = passage_encoder self.verbose = verbose def on_train_epoch_start( self, trainer: pl.Trainer, pl_module: pl.LightningModule, *args, **kwargs ) -> None: if trainer.current_epoch == 0: if self.verbose: logger.info("Current epoch is 0, skipping resetting model") return if self.verbose: logger.info("Resetting model, optimizer and lr scheduler") # reload model from scratch previous_device = pl_module.device trainer.model.model.question_encoder = GoldenRetrieverModel.from_pretrained( self.question_encoder ) trainer.model.model.question_encoder.to(previous_device) if self.passage_encoder is not None: trainer.model.model.passage_encoder = GoldenRetrieverModel.from_pretrained( self.passage_encoder ) trainer.model.model.passage_encoder.to(previous_device) trainer.strategy.setup_optimizers(trainer) class FreeUpIndexerVRAMCallback(pl.Callback): def __call__( self, pl_module: pl.LightningModule, *args, **kwargs, ) -> Any: logger.info("Freeing up GPU memory") # remove the index from the GPU memory # remove the embeddings from the GPU memory first if pl_module.model.document_index is not None: if pl_module.model.document_index.embeddings is not None: try: pl_module.model.document_index.embeddings.cpu() except Exception: logger.warning( "Could not move embeddings to CPU. Skipping freeing up VRAM." ) pass pl_module.model.document_index.embeddings = None import gc gc.collect() torch.cuda.empty_cache() def on_train_epoch_start( self, trainer: pl.Trainer, pl_module: pl.LightningModule, *args, **kwargs ) -> None: return self(pl_module) def on_test_epoch_start( self, trainer: pl.Trainer, pl_module: pl.LightningModule, *args, **kwargs ) -> None: return self(pl_module) class ShuffleTrainDatasetCallback(pl.Callback): def __init__(self, seed: int = 42, verbose: bool = True) -> None: super().__init__() self.seed = seed self.verbose = verbose self.previous_epoch = -1 def on_validation_epoch_end(self, trainer: pl.Trainer, *args, **kwargs): if self.verbose: if trainer.current_epoch != self.previous_epoch: logger.info(f"Shuffling train dataset at epoch {trainer.current_epoch}") # logger.info(f"Shuffling train dataset at epoch {trainer.current_epoch}") if trainer.current_epoch != self.previous_epoch: trainer.datamodule.train_dataset.shuffle_data( seed=self.seed + trainer.current_epoch + 1 ) self.previous_epoch = trainer.current_epoch class PrefetchTrainDatasetCallback(pl.Callback): def __init__(self, verbose: bool = True) -> None: super().__init__() self.verbose = verbose # self.previous_epoch = -1 def on_validation_epoch_end(self, trainer: pl.Trainer, *args, **kwargs): if trainer.datamodule.train_dataset.prefetch_batches: if self.verbose: # if trainer.current_epoch != self.previous_epoch: logger.info( f"Prefetching train dataset at epoch {trainer.current_epoch}" ) # if trainer.current_epoch != self.previous_epoch: trainer.datamodule.train_dataset.prefetch() self.previous_epoch = trainer.current_epoch class SubsampleTrainDatasetCallback(pl.Callback): def __init__(self, seed: int = 43, verbose: bool = True) -> None: super().__init__() self.seed = seed self.verbose = verbose def on_validation_epoch_end(self, trainer: pl.Trainer, *args, **kwargs): if self.verbose: logger.info(f"Subsampling train dataset at epoch {trainer.current_epoch}") trainer.datamodule.train_dataset.random_subsample( seed=self.seed + trainer.current_epoch + 1 ) class SaveRetrieverCallback(pl.Callback): def __init__( self, saving_dir: str | os.PathLike | None = None, verbose: bool = True, *args, **kwargs, ): super().__init__() self.saving_dir = saving_dir self.verbose = verbose self.free_up_indexer_callback = FreeUpIndexerVRAMCallback() @torch.no_grad() def __call__( self, trainer: pl.Trainer, pl_module: pl.LightningModule, *args, **kwargs, ): if self.saving_dir is None and trainer.logger is None: logger.info( "You need to specify an output directory (`saving_dir`) or a logger to save the retriever.\n" "Skipping saving retriever." ) return if self.saving_dir is not None: retriever_folder = Path(self.saving_dir) else: try: retriever_folder = Path(trainer.logger.experiment.dir) / "retriever" except Exception: logger.info( "You need to specify an output directory (`saving_dir`) or a logger to save the " "retriever.\nSkipping saving retriever." ) return retriever_folder.mkdir(exist_ok=True, parents=True) if self.verbose: logger.info(f"Saving retriever to {retriever_folder}") pl_module.model.save_pretrained(retriever_folder) def on_save_checkpoint( self, trainer: pl.Trainer, pl_module: pl.LightningModule, checkpoint: Dict[str, Any], ): self(trainer, pl_module) # self.free_up_indexer_callback(pl_module) class SampleNegativesDatasetCallback(pl.Callback): def __init__(self, seed: int = 42, verbose: bool = True) -> None: super().__init__() self.seed = seed self.verbose = verbose def on_validation_epoch_end(self, trainer: pl.Trainer, *args, **kwargs): if self.verbose: f"Sampling negatives for train dataset at epoch {trainer.current_epoch}" trainer.datamodule.train_dataset.sample_dataset_negatives( seed=self.seed + trainer.current_epoch ) class SubsampleDataCallback(pl.Callback): def __init__(self, seed: int = 42, verbose: bool = True) -> None: super().__init__() self.seed = seed self.verbose = verbose def on_validation_epoch_start(self, trainer: pl.Trainer, *args, **kwargs): if self.verbose: f"Subsampling data for train dataset at epoch {trainer.current_epoch}" if trainer.state.stage == RunningStage.SANITY_CHECKING: return trainer.datamodule.train_dataset.subsample_data( seed=self.seed + trainer.current_epoch ) def on_fit_start(self, trainer: pl.Trainer, *args, **kwargs): if self.verbose: f"Subsampling data for train dataset at epoch {trainer.current_epoch}" trainer.datamodule.train_dataset.subsample_data( seed=self.seed + trainer.current_epoch )