Блог

Я использую нейронную сеть с шипами для данных, которая имеет 21 функцию с размером пакета 128. После многих итераций обучения я получаю следующую ошибку (эта ошибка возникает не сразу!):

RuntimeError: shape '[128, -1]' is invalid for input of size 378 pytorch

Когда я пошел, чтобы распечатать, какие формы тензоров были раньше, я получаю следующее:

 Train
torch.Size([128, 21])
Test
torch.Size([128, 21])
 

Это моя сеть:

 class SpikingNeuralNetwork(nn.Module):
    """
        Parameters in SpikingNeuralNetwork class:
        
        1. number_inputs: Number of inputs to the SNN.
        2. number_hidden: Number of hidden layers.
        3. number_outputs: Number of output classes.
        4. beta: Decay rate. 
    """
    def __init__(self, number_inputs, number_hidden, number_outputs, beta):
        super().__init__()
        self.number_inputs = number_inputs
        self.number_hidden = number_hidden
        self.number_outputs = number_outputs
        self.beta = beta

        # Initialize layers
        self.fc1 = nn.Linear(self.number_inputs, self.number_hidden) # Applies linear transformation to all input points
        self.lif1 = snn.Leaky(beta = self.beta) # Integrates weighted input over time, emitting a spike if threshold condition is met
        self.fc2 = nn.Linear(self.number_hidden, self.number_outputs) # Applies linear transformation to output spikes of lif1
        self.lif2 = snn.Leaky(beta = self.beta) # Another spiking neuron, integrating the weighted spikes over time

    """
        Forward propagation of SNN. The code below function will only be called once the input argument x 
        is explicitly passed into net.

        @param x: input passed into the network
        @return layer of output after applying final spiking neuron
    """
    def forward(self, x):
        num_steps = 25

        # Initialize hidden states at t = 0
        mem1 = self.lif1.init_leaky()
        mem2 = self.lif2.init_leaky()

        # Record the final layer
        spk2_rec = []
        mem2_rec = []

        for step in range(num_steps):
            cur1 = self.fc1(x)
            spk1, mem1 = self.lif1(cur1, mem1)
            cur2 = self.fc2(spk1)
            spk2, mem2 = self.lif2(cur2, mem2)
            spk2_rec.append(spk2)
            mem2_rec.append(mem2)

        return torch.stack(spk2_rec, dim = 0), torch.stack(mem2_rec, dim = 0)
 

Это мой цикл обучения:

 def training_loop(net, train_loader, test_loader, dtype, device, optimizer):
    num_epochs = 1
    loss_history = []
    test_loss_history = []
    counter = 0

    # Temporal dynamics
    num_steps = 25

    # Outer training loop
    for epoch in range(num_epochs):
        iter_counter = 0
        train_batch = iter(train_loader)

        # Minibatch training loop
        for data, targets in train_batch:
            data = data.to(device)
            targets = targets.to(device)

            # Forward pass
            net.train()
            print("Train")
            print(data.size())
            spk_rec, mem_rec = net(data.view(batch_size, -1))

            # Initialize the loss and sum over time
            loss_val = torch.zeros((1), dtype = dtype, device = device)
            for step in range(num_steps):
                loss_val  = loss_function(mem_rec[step], targets.long().flatten().to(device))

            # Gradient calculation and weight update
            optimizer.zero_grad()
            loss_val.backward()
            optimizer.step()

            # Store loss history for future plotting
            loss_history.append(loss_val.item())

            # Test set
            with torch.no_grad():
                net.eval()
                test_data, test_targets = next(iter(test_loader))
                test_data = test_data.to(device)
                test_targets = test_targets.to(device)

                # Test set forward pass
                print("Test")
                print(test_data.size())
                test_spk, test_mem = net(test_data.view(batch_size, -1))

                # Test set loss
                test_loss = torch.zeros((1), dtype = dtype, device = device)
                for step in range(num_steps):
                    test_loss  = loss_function(test_mem[step], test_targets.long().flatten().to(device))
                test_loss_history.append(test_loss.item())

                # Print train/test loss and accuracy
                if counter % 50 == 0:
                    train_printer(epoch, iter_counter, counter, loss_history, data, targets, test_data, test_targets)
                counter = counter   1
                iter_counter = iter_counter   1
    
    return loss_history, test_loss_history
 

Ошибка возникает при spk_rec, mem_rec = net(data.view(batch_size, -1)) .

Код был принят из https://snntorch.readthedocs.io/en/latest/tutorials/tutorial_5.html , где он первоначально использовался для набора данных MNIST. Однако я не работаю с набором данных изображений. Я работаю с набором данных, который имеет 21 объект и предсказывает только одну цель (со 100 классами). Я попытался изменить data.view(batch_size, -1) и test_data.view(batch_size, -1) на data.view(batch_size, 21) и test_data.view(batch_size, 21) на основе некоторых других ответов на форуме, которые я видел, и моя программа сейчас выполняется через цикл обучения. Есть ли у кого-нибудь какие-либо предложения о том, как я могу пройти обучение без ошибок?

РЕДАКТИРОВАТЬ: теперь я получаю сообщение об ошибке RuntimeError: shape '[128, 21]' is invalid for input of size 378 spk_rec, mem_rec = net(data.view(batch_size, -1)) .

Вот мои загрузчики данных:

     train_loader = DataLoader(dataset = train, batch_size = batch_size, shuffle = True)
    test_loader = DataLoader(dataset = test, batch_size = batch_size, shuffle = True)
 

Мой размер пакета равен 128.

Пытаюсь запустить его самостоятельно, чтобы попытаться решить вашу проблему, мне также повезло: net params и snn.snn.Утечка

 import torch
from torch import nn
from torch.utils.data import DataLoader


class SpikingNeuralNetwork(nn.Module):
    """
        Parameters in SpikingNeuralNetwork class:

        1. number_inputs: Number of inputs to the SNN.
        2. number_hidden: Number of hidden layers.
        3. number_outputs: Number of output classes.
        4. beta: Decay rate.
    """

    def __init__(self, number_inputs, number_hidden, number_outputs, beta):
        super().__init__()
        self.number_inputs = number_inputs
        self.number_hidden = number_hidden
        self.number_outputs = number_outputs
        self.beta = beta

        # Initialize layers
        self.fc1 = nn.Linear(self.number_inputs,
                             self.number_hidden)  # Applies linear transformation to all input points
        self.lif1 = snn.Leaky(
            beta=self.beta)  # Integrates weighted input over time, emitting a spike if threshold condition is met
        self.fc2 = nn.Linear(self.number_hidden,
                             self.number_outputs)  # Applies linear transformation to output spikes of lif1
        self.lif2 = snn.Leaky(beta=self.beta)  # Another spiking neuron, integrating the weighted spikes over time

    """
        Forward propagation of SNN. The code below function will only be called once the input argument x 
        is explicitly passed into net.

        @param x: input passed into the network
        @return layer of output after applying final spiking neuron
    """

    def forward(self, x):
        num_steps = 25

        # Initialize hidden states at t = 0
        mem1 = self.lif1.init_leaky()
        mem2 = self.lif2.init_leaky()

        # Record the final layer
        spk2_rec = []
        mem2_rec = []

        for step in range(num_steps):
            cur1 = self.fc1(x)
            spk1, mem1 = self.lif1(cur1, mem1)
            cur2 = self.fc2(spk1)
            spk2, mem2 = self.lif2(cur2, mem2)
            spk2_rec.append(spk2)
            mem2_rec.append(mem2)

        return torch.stack(spk2_rec, dim=0), torch.stack(mem2_rec, dim=0)


batch_size = 2
train = torch.rand(128, 21)
test = torch.rand(128, 21)
train_loader = DataLoader(dataset=train, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(dataset=test, batch_size=batch_size, shuffle=True)

net = SpikingNeuralNetwork(number_inputs=1)
loss_function = nn.CrossEntropyLoss()
optimizer = nn.optim.Adam(net.parameters(), lr=0.1)


def training_loop(net, train_loader, test_loader, dtype, device, optimizer):
    num_epochs = 1
    loss_history = []
    test_loss_history = []
    counter = 0

    # Temporal dynamics
    num_steps = 25

    # Outer training loop
    for epoch in range(num_epochs):
        iter_counter = 0
        train_batch = iter(train_loader)

        # Minibatch training loop
        for data, targets in train_batch:
            data = data.to(device)
            targets = targets.to(device)

            # Forward pass
            net.train()
            print("Train")
            print(data.size())
            spk_rec, mem_rec = net(data.view(batch_size, -1))

            # Initialize the loss and sum over time
            loss_val = torch.zeros((1), dtype=dtype, device=device)
            for step in range(num_steps):
                loss_val  = loss_function(mem_rec[step], targets.long().flatten().to(device))

            # Gradient calculation and weight update
            optimizer.zero_grad()
            loss_val.backward()
            optimizer.step()

            # Store loss history for future plotting
            loss_history.append(loss_val.item())

            # Test set
            with torch.no_grad():
                net.eval()
                test_data, test_targets = next(iter(test_loader))
                test_data = test_data.to(device)
                test_targets = test_targets.to(device)

                # Test set forward pass
                print("Test")
                print(test_data.size())
                test_spk, test_mem = net(test_data.view(batch_size, -1))

                # Test set loss
                test_loss = torch.zeros((1), dtype=dtype, device=device)
                for step in range(num_steps):
                    test_loss  = loss_function(test_mem[step], test_targets.long().flatten().to(device))
                test_loss_history.append(test_loss.item())

                # Print train/test loss and accuracy
                if counter % 50 == 0:
                    train_printer(epoch, iter_counter, counter, loss_history, data, targets, test_data, test_targets)
                counter = counter   1
                iter_counter = iter_counter   1

    return loss_history, test_loss_history

 

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

1. Вместо spk_rec, mem_rec = net(data.view(batch_size, -1)) этого попробуйте spk_rec, mem_rec = net(data.flatten(1)) сохранить первое измерение ваших данных.
2. В качестве альтернативы вам может потребоваться установить drop_last=True функции загрузки данных.