Repository: SuperDuperDB/superduperdb
File: superduperdb/ext/transformers/model.py (Python)

Code before:
        ('data_collator', dill_serializer),
        ('callbacks', dill_serializer),
        ('optimizers', dill_serializer),
        ('preprocess_logits_for_metrics', dill_serializer),
    )

    output_dir: str = ''
    data_collator: t.Optional[transformers.data.data_collator.DataCollator] = None
    callbacks: t.Optional[t.List[transformers.trainer_callback.TrainerCallback]] = None
    optimizers: t.Tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR] = (
        None,
        None,
    )
    preprocess_logits_for_metrics: t.Optional[
        t.Callable[[torch.Tensor, torch.Tensor], torch.Tensor]
    ] = None

    def __post_init__(self, artifacts):
        assert self.output_dir == '' or self.output_dir == self.identifier
        self.output_dir = self.identifier
        TrainingArguments.__post_init__(self)
        return Trainer.__post_init__(self, artifacts)

    @property
    def native_arguments(self):
        kwargs = {k: getattr(self, k) for k in _TRAINING_DEFAULTS}
        return TrainingArguments(output_dir=self.identifier, **kwargs)

    def _build_trainer(
        self,
        model: 'TextClassificationPipeline',
        db: Datalayer,
        train_dataset: QueryDataset,
        valid_dataset: QueryDataset,
    ):
        def preprocess_function(examples):
            return model.pipeline.tokenizer(
                examples['text'], padding="max_length", truncation=True
            )

        train_dataset = NativeDataset.from_list(
            [train_dataset[i] for i in range(len(train_dataset))]
        )
        train_dataset = train_dataset.map(preprocess_function)
        valid_dataset = NativeDataset.from_list(
            [valid_dataset[i] for i in range(len(valid_dataset))]
        )
        valid_dataset = valid_dataset.map(preprocess_function)

        def compute_metrics(eval_pred):
            output = {}

Review comment (suggestion):
TrainingArguments instantiation operation should not be placed during import, otherwise some bad errors may occur due to its internal `__post_init__`

Code after:
    )

    output_dir: str = ''
    data_collator: t.Optional[transformers.data.data_collator.DataCollator] = None
    callbacks: t.Optional[t.List[transformers.trainer_callback.TrainerCallback]] = None
    optimizers: t.Tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR] = (
        None,
        None,
    )
    preprocess_logits_for_metrics: t.Optional[
        t.Callable[[torch.Tensor, torch.Tensor], torch.Tensor]
    ] = None

    def __post_init__(self, artifacts):
        assert self.output_dir == '' or self.output_dir == self.identifier
        self.output_dir = self.identifier
        TrainingArguments.__post_init__(self)
        return Trainer.__post_init__(self, artifacts)

    @property
    def native_arguments(self):
        _TRAINING_DEFAULTS = {
            k: v
            for k, v in TrainingArguments('_tmp').to_dict().items()
            if k != 'output_dir'
        }
        kwargs = {k: getattr(self, k) for k in _TRAINING_DEFAULTS}
        return TrainingArguments(output_dir=self.identifier, **kwargs)

    def _build_trainer(
        self,
        model: 'TextClassificationPipeline',
        db: Datalayer,
        train_dataset: QueryDataset,
        valid_dataset: QueryDataset,
    ):
        def preprocess_function(examples):
            return model.pipeline.tokenizer(
                examples['text'], padding="max_length", truncation=True
            )

        train_dataset = NativeDataset.from_list(
            [train_dataset[i] for i in range(len(train_dataset))]
        )
        train_dataset = train_dataset.map(preprocess_function)
        valid_dataset = NativeDataset.from_list(
            [valid_dataset[i] for i in range(len(valid_dataset))]
        )
        valid_dataset = valid_dataset.map(preprocess_function)

        def compute_metrics(eval_pred):