Error when predicting from checkpoint

implementation help
1

I am having trouble in predicting after load a model from checkpoint:

> model = AutoDPlncRPI.load_from_checkpoint("tb-logs/new_test-epoch=00-val_loss=0.69-v1.ckpt")
> model.eval()
> x = torch.randn(1,96)
> y_hat = model(x)
Traceback (most recent call last):
  File "/opt/homebrew/anaconda3/envs/SeqLevelDPRPI/lib/python3.10/site-packages/IPython/core/interactiveshell.py", line 3433, in run_code
    exec(code_obj, self.user_global_ns, self.user_ns)
  File "<ipython-input-6-bd0200007a4a>", line 1, in <module>
    model(x)
  File "/opt/homebrew/anaconda3/envs/SeqLevelDPRPI/lib/python3.10/site-packages/torch/nn/modules/module.py", line 1130, in _call_impl
    return forward_call(*input, **kwargs)
  File "/opt/homebrew/anaconda3/envs/SeqLevelDPRPI/lib/python3.10/site-packages/lightning/pytorch/core/module.py", line 656, in forward
    return super().forward(*args, **kwargs)
  File "/opt/homebrew/anaconda3/envs/SeqLevelDPRPI/lib/python3.10/site-packages/torch/nn/modules/module.py", line 201, in _forward_unimplemented
    raise NotImplementedError(f"Module [{type(self).__name__}] is missing the required \"forward\" function")
NotImplementedError: Module [AutoDPlncRPI] is missing the required "forward" function

This is my model:

class AutoDPlncRPI(pl.LightningModule):
    def __init__(self, seq_length, d_model, vocabs_length, nhead, num_encoder_layers, dropout, lr):
        super().__init__()
        # save hyperparameters
        self.save_hyperparameters()
        # model
        self.embedding = EmbeddingLayer(d_model=d_model, input_vocab_size=vocabs_length)
        seq_length, embed_dim = self._count_pos_encoder_input_size(seq_length)
        self.pos_encoder = PositionalEncoding(d_model=embed_dim, dropout=dropout,
                                              max_len=seq_length)
        self.transformer_encoder = TransformerEncoder(d_model=embed_dim, nhead=nhead,
                                                      num_encoder_layers=num_encoder_layers, dropout=dropout)
        self.cnn = CnnThreeLayer(d_model=embed_dim, dropout=dropout)
        self.output_layer = OutputLayer(flatten_size=self._count_flatten_size(seq_length), dropout=dropout)
        # loss function
        self.output_activation = nn.Sigmoid()
        self.loss_fn = nn.BCELoss()
        # metrics
        self.accuracy = torchmetrics.Accuracy(task="binary", threshold=0.5, num_classes=2)
        self.f1_score = torchmetrics.F1Score(task="binary", threshold=0.5, num_classes=2)
        self.auroc = torchmetrics.AUROC(task="binary", threshold=0.5, num_classes=2)
        # configure optimizers
        self.lr = lr

    def _count_pos_encoder_input_size(self, seq_length):
        """
        This function will calculate the seq length after convolving by cnn_layer
        """
        zeros = torch.zeros([1, seq_length], dtype=torch.long)
        x = self.embedding(zeros)
        # x = self.cnn_layer(x)
        return x.size()[1], x.size()[2]

    def _count_flatten_size(self, seq_length):
        """
        This function will calculate the seq length after cnn
        """
        zeros = torch.zeros([1, seq_length], dtype=torch.long)
        flatten = nn.Flatten()
        x = self.embedding(zeros)
        # x = self.cnn_layer(x)
        x = self.pos_encoder(x)
        x = self.transformer_encoder(x)
        x = self.cnn(x)
        x = flatten(x)
        return x.size()[1]

    def _common_steps(self, batch, batch_idx):
        x, y = batch
        y_hat = self.embedding(x)
        # y_hat = self.cnn_layer(y_hat)
        y_hat = self.pos_encoder(y_hat)
        y_hat = self.transformer_encoder(y_hat)
        y_hat = self.cnn(y_hat)
        y_hat = self.output_layer(y_hat)
        y_hat = self.output_activation(y_hat)
        loss = self.loss_fn(y_hat, y)
        accuracy = self.accuracy(y_hat, y)
        f1_score = self.f1_score(y_hat, y)
        auroc = self.auroc(y_hat, y)
        return loss, accuracy, f1_score, auroc

    def training_step(self, batch, batch_idx):
        train_loss, accuracy, f1_score, auroc = self._common_steps(batch, batch_idx)
        self.log_dict({"train_loss": train_loss, "train_accuracy": accuracy,
                       "train_f1_score": f1_score, "train_auroc": auroc})
        return train_loss

    def validation_step(self, batch, batch_idx):
        val_loss, accuracy, f1_score, auroc = self._common_steps(batch, batch_idx)
        self.log_dict({"val_loss": val_loss, "val_accuracy": accuracy,
                       "val_f1_score": f1_score, "val_auroc": auroc})
        return val_loss

    def test_step(self, batch, batch_idx):
        test_loss, accuracy, f1_score, auroc = self._common_steps(batch, batch_idx)
        self.log_dict({"test_loss": test_loss, "test_accuracy": accuracy,
                       "test_f1_score": f1_score, "test_auroc": auroc})
        return test_loss

    def configure_optimizers(self):
        optimizer = torch.optim.AdamW(self.parameters(), lr=self.lr)
        # scheduler = get_cosine_schedule_with_warmup(optimizer=optimizer, num_warmup_steps=50,
        #                                             num_training_steps=self.trainer.max_epochs)
        scheduler = CosineAnnealingLR(optimizer=optimizer, T_max=self.trainer.max_epochs,
                                      eta_min=3 * 1e-4, verbose=True)
        return [optimizer], [scheduler]
2

Hello @breezewjc952

As the error from PyTorch already nicely tells you, you are missing the implementation of the forward function!

You should probably define this on your model:

def forward(self, batch):
    x, y = batch
    y_hat = self.embedding(x)
    # y_hat = self.cnn_layer(y_hat)
    y_hat = self.pos_encoder(y_hat)
    y_hat = self.transformer_encoder(y_hat)
    y_hat = self.cnn(y_hat)
    y_hat = self.output_layer(y_hat)
    y_hat = self.output_activation(y_hat)
    return y_hat

(given what you have used in the step methods.