ML: Вступ до PyTorch: 3. Нейронні мережі
Послідовність шарів
У цьому документі обговорюється створення і навчання нейронних мереж на бібліотеці PyTorch, основи якої описані тут і тут. Довідник з основних типів шарів, помилок і оптимізаторів знаходиться тут.
Прості нейронні мережі з послідовною архітектурою можна створювати за допомогою стопкового інтерфейсу. Наприклад, двошарова повнозв’язна мережа з двома входами (nX=2 ознаки), одним (nY=1) сигмоїдним виходом (2 класи) і п’ятьма нейронами (nH=5) у прихованому шарі має вигляд:
import torch
import torch.nn as nn
nX, nH, nY = 2, 5, 1
model = nn.Sequential(
nn.Linear(nX, nH), # перший шар
nn.Sigmoid(), # активація прихованого шару
nn.Linear(nH, nY), # другий, вихідний шар
nn.Sigmoid() ) # його активаційна фу-я
Шар Linear робить лінійне перетворення вхідного тензора форми (N, nX), де N - число прикладів. На виході шару виходить тензор форми (N, nH), який пропускається через сигмоїд (нові nH ознаки). Другий шар дає тензор (N, nY) від якого також береться через сигмоїд. Якщо його значення менше 0.5 - це перший клас, якщо більше 0.5 - другий.
Функціональна архітектура
При проектуванні складних нейронних мереж їхню архітектуру зручніше задавати у функціональному вигляді.
Для цього створюється нащадок класу nn.Module.
У ньому перевизначається конструктор і метод forward.
У конструкторі задаються необхідні шари і ініціалізуються їхні параметри.
Звернімо увагу на виклик конструктора предка, якому необхідно передати ім’я нашого класу.
У методі forward (пряме поширення) будується архітектура мережі (шари зв’язуються
в обчислювальний граф):
class TwoLayersNet(nn.Module):
def __init__(self, nX, nH, nY):
super(TwoLayersNet, self).__init__() # конструктор предка з цим ім’ям
self.fc1 = nn.Linear(nX, nH) # створюємо параметри моделі
self.fc2 = nn.Linear(nH, nY) # у повнозв’язних шарах
def forward(self, x): # задається прямий прохід
x = self.fc1(x) # вихід першого шару
x = nn.Sigmoid()(x) # пропускаємо через Sigmoid
x = self.fc2(x) # вихід другого шару
x = nn.Sigmoid()(x) # пропускаємо через сигмоїд
return x
model = TwoLayersNet(2, 5, 1) # екземпляр мережі
Виклик nn.Line(in_features, out_features) створює два екземпляри лінійного повнозв’язного шару fc1, fc2 (fully connected). Одночасно з цим задаються випадкові значення їхніх ваг і зміщень. Дужковий оператор fc1(x) у методі forward запускає обчислення всередині шару і на виході видає результуючий тензор.
Модельні дані
Створимо на PyTorch модельні дані для об’єктів двох видів (два класи),
що характеризуються двома ознаками. Нехай на 2D площині
об’єкти першого класу заповнюють одиничний квадрат $[0...1]^2$, крім круга в центрі квадрата, де знаходяться
об’єкти другого класу:
X = torch.rand (1200,2) Y = (torch.sum((X - 0.5)**2, axis=1) < 0.1).float().view(-1,1)
За допомогою бібліотеки matplotlib намалюємо те, що в результаті вийшло:
import matplotlib.pyplot as plt # побудова графіків
plt.figure (figsize=(5, 5)) # розміри (квадрат)
plt.scatter(X.numpy()[:,0], X.numpy()[:,1], c=Y.numpy()[:,0],
s=30, cmap=plt.cm.Paired, edgecolors='k')
plt.show() # виводимо рисунок
Навчання мережі
Для навчання мережі необхідно створити функцію помилки (те, що слід мінімізувати) і оптимізатор (який буде мінімізувати помилку). Оптимізатору передаються параметри моделі. Як помилку виберемо бінарну крос-ентропію (BCELoss), а оптимізатором буде SGD (Stochastic Gradient Descent):
model = TwoLayersNet(2, 5, 1) # екземпляр мережі
loss = nn.BCELoss()
optimizer = torch.optim.SGD(model.parameters(), # параметри моделі
lr=0.5, momentum=0.8) # параметри оптимізатора
На кожній ітерації з навчальних прикладів формується масив батчів,
запускається пряме поширення
y = model(bx) і
обчислюється помилка loss (з порівняння виходу y моделі з "правильними" значеннями yb).
На обчислювальному графі помилки метод loss.backward() дає
градієнти параметрів моделі, за допомогою яких оптимізатор optimizer у методі step()
отримує нові значення параметрів:
def fit(model, X,Y, batch_size=100, train=True):
model.train(train) # важливо для Dropout, BatchNorm
sumL, sumA, numB = 0, 0, int( len(X)/batch_size ) # помилка, точність, батчів
for i in range(0, numB*batch_size, batch_size):
xb = X[i: i+batch_size] # поточний батч,
yb = Y[i: i+batch_size] # X,Y - torch тензори
y = model(xb) # пряме поширення
L = loss(y, yb) # обчислюємо помилку
if train: # у режимі навчання
optimizer.zero_grad() # обнуляємо градієнти
L.backward() # обчислюємо градієнти
optimizer.step() # підправляємо параметри
sumL += L.item() # сумарна помилка (item з графа)
sumA += (y.round() == yb).float().mean() # точність визначення класу
return sumL/numB, sumA/numB # середня помилка і точність
Функція fit робить прохід по всіх даних. Це називається епохою навчання. Якщо параметр leran=True, то відбувається навчання моделі (зміна її параметрів). Значення leran=False використовується при оцінці якості моделі без її зміни. Як метрики якості використовується середня помилка за всіма прикладами і середня точність (частка правильно передбачених класів). Зазвичай потрібно досить багато епох навчання:
# режим оцінки моделі:
print( "before: loss: %.4f accuracy: %.4f" % fit(model, X,Y, train=False) )
epochs = 1000 # число епох
for epoch in range(epochs): # епоха - прохід по всіх прикладах
L,A = fit(model, X, Y) # одна епоха
if epoch % 100 == 0 or epoch == epochs-1:
print(f'epoch: {epoch:5d} loss: {L:.4f} accuracy: {A:.4f}' )
У результаті отримаємо щось типу:
before: loss: 0.6340 accuracy: 0.6950 epoch: 0 loss: 0.6327 accuracy: 0.6550 .... epoch: 999 loss: 0.0334 accuracy: 0.9908
Для покращення навчання зазвичай варто перемішувати дані. Це можна зробити двома способами. Перший запускається перед функцією fit на початку кожної епохи:
idx = torch.randperm( len(X) ) # перемішаний список індексів
X = X[idx]
Y = Y[idx]
У цьому методі створюється нова пам’ять для всіх даних, що для великих навчальних масивів
не завжди добре. Другий спосіб робить випадкову вибірку лише для поточного батча
всередині функції fit:
idx = torch.randint(high = len(X), size = (batch_size,) )
xb = X[idx]
yb = Y[idx]
Пам’яті при цьому використовується менше, але при проході по одній епосі в навчання
можуть потрапити не всі приклади, а в батч можуть потрапити однакові приклади.
Цей метод використовується в класі DataLoader (параметр shuffle), див. нижче.
Таким чином, порівняно з Keras у TensorFloor, процедуру навчання необхідно написати самостійно. Однак у складних випадках це дозволяє втручатися в процес на будь-якому етапі. Наприклад, можна вставити свій оптимізатор :)
. if train: # у режимі навчання
L.backward() # обчислюємо градієнти
with torch.no_grad():
for p in model.parameters():
p.add_(p.grad, alpha=-0.7) # p += -0.7*grad
p.grad.zero_()
Вивід структури мережі
Найпростіший спосіб побачити шари мережі — це просто вивести на друк модель:
print(model) # текстове представлення моделі
Зручніше скористатися стороннім модулем modelsummary. У його функцію summary передається вхідний тензор з будь-якими значеннями (але правильної форми) з одного прикладу (у shape це буде перша розмірність -1):
from modelsummary import summary summary(model, torch.zeros(1, 2), show_input=False) # аналог summary у keras
-----------------------------------------------------------------------
Layer (type) Output Shape Param #
=======================================================================
Linear-1 [-1, 5] 15
Linear-2 [-1, 1] 6
=======================================================================
Total params: 21
Trainable params: 21
Non-trainable params: 0
Якщо встановити show_input=True, то замість форми виходу
буде друкуватися форма входу кожного шару
Для складених моделей можна запускати функцію summary
з параметром show_hierarchical=True.
for layer in model:
print("***", layer)
for param in layer.parameters():
print(param.data.numpy())
У загальному випадку параметри моделі виводяться таким чином:
for param in model.parameters():
print(param.numel(), param.size(), param.data.numpy())
tot = 0
for k, v in model.state_dict().items():
pars = np.prod(list(v.shape)); tot += pars
print(f'{k:20s} :{pars:7d} shape: {tuple(v.shape)} ')
print(f"{'total':20s} :{tot:7d}")
fc1.weight : 10 shape: (5, 2) fc1.bias : 5 shape: (5,) fc2.weight : 5 shape: (1, 5) fc2.bias : 1 shape: (1,) total : 21
Нарешті, бібліотека torchviz дозволяє зобразити обчислювальний граф моделі (рисунок праворуч):
import torchviz
torchviz.make_dot(model(X),
params = dict(model.named_parameters()) )
Вузол AddmmBackward відповідає функції torch.addmm(v, m1, m2, b=1, a=1), яка перемножує матриці і додає до них вектор: b v + a (m1 @ m2).
Збереження і завантаження
Метод torch.save зберігає в бінарному вигляді будь-який словник, у тому числі стани моделі і оптимізатора:
import datetime
state = {'info': "Це моя мережа", # опис
'date': datetime.datetime.now(), # дата і час
'model' : model.state_dict(), # параметри моделі
'optimizer': optimizer.state_dict()} # стан оптимізатора
torch.save(state, 'state.pt') # зберігаємо файл
Потім їх можна завантажити, створивши попередньо модель і оптимізатор (його параметри можуть бути будь-якими,
бо вони завантажаться з файлу):
state = torch.load('state.pt') # завантажуємо файл
m = TwoLayersNet(2, 5, 1) # екземпляр мережі
optimizer = torch.optim.SGD(m.parameters(),lr=1) # оптимізатор (будь-які параметри)
m. load_state_dict(state['model']) # отримуємо параметри моделі
optimizer.load_state_dict(state['optimizer']) # отримуємо стан оптимізатора
print(state['info'], state['date']) # допоміжна інформація
Класи датасетів
У PyTorch прийнято обгортати навчальні дані в клас. Він є нащадком класу Dataset і перевизначає метод __len__ числа прикладів і метод __getitem__ отримання прикладу за індексом idx:
class MyDataset(torch.utils.data.Dataset): # Нащадок класу Dataset
def __init__(self, N = 10, *args, **kwargs):
super().__init__(*args,**kwargs)
self.x = torch.rand(N,1) # випадкові дані
def __len__(self): # Кількість прикладів
return len(self.x)
def __getitem__(self, idx): # отримати idx-тий приклад
return {'input' : self.x[idx],
'target': 2*self.x[idx]}
data = MyDataset()
for sample in data:
print(sample) # {'input': tensor([0.1545]), 'target': tensor([0.3090])} ...
Створивши датасет, його можна передати об’єкту DataLoader,
який буде видавати батчі, перемішувати, розбивати на навчання і тест, нормалізувати дані і т.п.
(див. документацію).
train_loader = torch.utils.data.DataLoader(dataset=data, batch_size=12, shuffle=False)
for batch in train_loader: # отримуємо батчі для тренування
print(batch['input'], batch['target'])
Обчислення на GPU
Для обчислення на GPU (як і у загальному випадку) необхідно створити відповідні обчислювальні пристрої:
gpu = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
cpu = torch.device("cpu")
Після створення моделі її можна відправити на графічну карту і (якщо дозволяє її пам’ять)
те саме зробити з навчальними даними:
model = TwoLayersNet(2, 5, 1) # екземпляр мережі model.to(gpu) # відправляємо його на GPU X = X.to(gpu) # відправляємо навчальні дані на GPU Y = Y.to(gpu)
Якщо для навчальних даних місця на GPU не вистачає, їх можна відправляти туди батчами.
При навчанні помилка пересилається з GPU в CPU. Зауважимо, що для маленьких моделей і навчальних масивів даних використання GPU не доцільне і може навіть сповільнити навчання порівняно з CPU.
Власні шари
Шари, як і вся мережа, є нащадком класу nn.Module. Приклад реалізації лінійного шару:
import math
from torch.nn.parameter import Parameter
class My_Linear(nn.Module):
def __init__(self, in_F, out_F):
super(My_Linear, self).__init__()
self.weight = Parameter(torch.Tensor(out_F, in_F))
self.bias = Parameter(torch.Tensor(out_F))
self.reset_parameters()
def reset_parameters(self):
for p in self.parameters():
stdv = 1.0 / math.sqrt(p.shape[0])
p.data.uniform_(-stdv, stdv)
def forward(self, x):
return x @ self.weight.t() + self.bias
Власний оптимізатор
Нижче наведено код простого SGD-оптимізатора. Звернімо увагу на декоратор @torch.no_grad() перед методом step. Він означає, що етап зміни параметрів буде виконуватися при вимкненому режимі побудови графа.
class SGD(torch.optim.Optimizer):
def __init__(self, params, lr=0.1, momentum=0):
defaults = dict(lr=lr, momentum=momentum)
super(SGD, self).__init__(params, defaults)
@torch.no_grad()
def step(self):
for group in self.param_groups:
momentum = group['momentum']
lr = group['lr']
for p in group['params']:
if p.grad is None:
continue
grad = p.grad
if momentum != 0:
p_state = self.state[p]
if 'momentum_buf' not in p_state:
buf = p_state['momentum_buf'] = torch.clone(grad)
else:
buf = p_state['momentum_buf']
buf.mul_(momentum).add_( grad )
grad = buf
p.add_(grad, alpha = -lr)
optimizer = SGD(model.parameters(), lr=0.1, momentum=0.9)
Параметри готових оптимізаторів можна також змінювати безпосередньо в процесі навчання
def adjust_optim(optimizer, epoch):
if epoch == 1000:
optimizer.param_groups[0]['betas'] = (0.3, optimizer.param_groups[0]['betas'][1])
if epoch > 1000:
optimizer.param_groups[0]['lr'] *= 0.9999
