Google colab链接
Phoneme Classification
分类训练
训练一个深度神经网络分类器来预测语音语料库TIMIT中每一帧的音素。
范例
数据准备
导入数据
import numpy as npprint('Loading data ...')data_root='./timit_11/'
# 分别导入训练数据，训练标签，测试数据
train = np.load(data_root + 'train_11.npy')
train_label = np.load(data_root + 'train_label_11.npy')
test = np.load(data_root + 'test_11.npy')print('Size of training data: {}'.format(train.shape))
print('Size of testing data: {}'.format(test.shape))
创建自定义数据集
import torch
from torch.utils.data import Dataset# 定义数据集格式，data是数据，label是对应的标签，如果label为空，则表示需要预测
class TIMITDataset(Dataset):def __init__(self, X, y=None):self.data = torch.from_numpy(X).float()if y is not None:y = y.astype(np.int)self.label = torch.LongTensor(y)else:self.label = Nonedef __getitem__(self, idx):if self.label is not None:return self.data[idx], self.label[idx]else:return self.data[idx]def __len__(self):return len(self.data)
分割数据
# 将训练数据二八分，一部分用于训练，一部分用于验证
VAL_RATIO = 0.2percent = int(train.shape[0] * (1 - VAL_RATIO))
train_x, train_y, val_x, val_y = train[:percent], train_label[:percent], train[percent:], train_label[percent:]
print('Size of training set: {}'.format(train_x.shape))
print('Size of validation set: {}'.format(val_x.shape))
将数据导入DataLoader
# 将数据导入到DataLoader中，每一次batch的大小为64，训练阶段shuffle为True是为了打乱数据得到更好的训练效果，而验证阶段则不需要
BATCH_SIZE = 64from torch.utils.data import DataLoadertrain_set = TIMITDataset(train_x, train_y)
val_set = TIMITDataset(val_x, val_y)
train_loader = DataLoader(train_set, batch_size=BATCH_SIZE, shuffle=True) #only shuffle the training data
val_loader = DataLoader(val_set, batch_size=BATCH_SIZE, shuffle=False)
清空不必要的内存
# 由于数据集的大小非常大，所以最好清除不再使用的内存
import gcdel train, train_label, train_x, train_y, val_x, val_y
gc.collect()
创建自定义模型
# 定义训练模型
import torch
import torch.nn as nnclass Classifier(nn.Module):def __init__(self):super(Classifier, self).__init__()self.layer1 = nn.Linear(429, 1024)self.layer2 = nn.Linear(1024, 512)self.layer3 = nn.Linear(512, 128)self.out = nn.Linear(128, 39)self.act_fn = nn.Sigmoid()# 前向传播函数，每过一层就走一次激活函数def forward(self, x):x = self.layer1(x)x = self.act_fn(x)x = self.layer2(x)x = self.act_fn(x)x = self.layer3(x)x = self.act_fn(x)x = self.out(x)return x
训练
检查device
#check device
def get_device():return 'cuda' if torch.cuda.is_available() else 'cpu'
设置随机种子
# 设置同样的种子以确保实验的可重复性
# fix random seed
def same_seeds(seed):torch.manual_seed(seed)if torch.cuda.is_available():torch.cuda.manual_seed(seed)torch.cuda.manual_seed_all(seed)np.random.seed(seed)torch.backends.cudnn.benchmark = Falsetorch.backends.cudnn.deterministic = True
设置训练属性
# fix random seed for reproducibility
same_seeds(0)# get device
device = get_device()
print(f'DEVICE: {device}')# training parameters
num_epoch = 20 # number of training epoch
learning_rate = 0.0001 # learning rate# ckpt检查点文件
# the path where checkpoint saved
model_path = './model.ckpt'# 定义模型，损失函数以及优化器
# create model, define a loss function, and optimizer
model = Classifier().to(device)
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)
开始训练
# start trainingbest_acc = 0.0
for epoch in range(num_epoch):train_acc = 0.0train_loss = 0.0val_acc = 0.0val_loss = 0.0# trainingmodel.train() # set the model to training modefor i, data in enumerate(train_loader):inputs, labels = datainputs, labels = inputs.to(device), labels.to(device)optimizer.zero_grad()outputs = model(inputs)# 计算Loss函数batch_loss = criterion(outputs, labels)# 获取输出中概率最大的类别的索引，作为预测结果。_, train_pred = torch.max(outputs, 1) # get the index of the class with the highest probability# 计算损失并反向传播计算每个参数的梯度batch_loss.backward()# 梯度更新并更新优化器的参数optimizer.step()# 将预测结果和真实标签都从GPU移到CPU，然后逐个元素比较生成一个Bool数组# 然后通过.sum()对Bool数组求和，得到当前批次中正确预测的数量# .item()将张量转为Python的标量类型，以便进行累加train_acc += (train_pred.cpu() == labels.cpu()).sum().item()# .item()将张量转为Python的标量类型，以便进行累加train_loss += batch_loss.item()# validationif len(val_set) > 0:model.eval() # set the model to evaluation modewith torch.no_grad():for i, data in enumerate(val_loader):inputs, labels = datainputs, labels = inputs.to(device), labels.to(device)outputs = model(inputs)batch_loss = criterion(outputs, labels)_, val_pred = torch.max(outputs, 1)val_acc += (val_pred.cpu() == labels.cpu()).sum().item() # get the index of the class with the highest probabilityval_loss += batch_loss.item()print('[{:03d}/{:03d}] Train Acc: {:3.6f} Loss: {:3.6f} | Val Acc: {:3.6f} loss: {:3.6f}'.format(epoch + 1, num_epoch, train_acc/len(train_set), train_loss/len(train_loader), val_acc/len(val_set), val_loss/len(val_loader)))# if the model improves, save a checkpoint at this epochif val_acc > best_acc:best_acc = val_acctorch.save(model.state_dict(), model_path)print('saving model with acc {:.3f}'.format(best_acc/len(val_set)))else:print('[{:03d}/{:03d}] Train Acc: {:3.6f} Loss: {:3.6f}'.format(epoch + 1, num_epoch, train_acc/len(train_set), train_loss/len(train_loader)))# if not validating, save the last epoch
if len(val_set) == 0:torch.save(model.state_dict(), model_path)print('saving model at last epoch')
测试
设置测试集
# create testing dataset
test_set = TIMITDataset(test, None)
test_loader = DataLoader(test_set, batch_size=BATCH_SIZE, shuffle=False)# create model and load weights from checkpoint
model = Classifier().to(device)
model.load_state_dict(torch.load(model_path))
开始预测
predict = []
model.eval() # set the model to evaluation mode
with torch.no_grad():for i, data in enumerate(test_loader):inputs = datainputs = inputs.to(device)outputs = model(inputs)_, test_pred = torch.max(outputs, 1) # get the index of the class with the highest probabilityfor y in test_pred.cpu().numpy():predict.append(y)
写入CSV文件
with open('prediction.csv', 'w') as f:f.write('Id,Class\n')for i, y in enumerate(predict):f.write('{},{}\n'.format(i, y))
