使用3层前馈神经网络

通过PyTorch构建前馈神经网络，并对二维数据点进行分类。在该例子当中，所有的训练数据和测试数据都是通过高斯混合模型GMM生成的：

import numpy as np
from sklearn.mixture import GaussianMixture
import torch
import torch.nn as nn
import torch.optim as optim
from matplotlib import pyplot


def GenerateData():
    GMM1 = GaussianMixture(n_components=2, covariance_type='spherical')
    GMM1.weights_ = np.array([0.5, 0.5])
    GMM1.means_ = np.array([[0, 0], [1, 1]])
    GMM1.covariances_ = np.array([0.05, 0.05])

    GMM2 = GaussianMixture(n_components=2, covariance_type='spherical')
    GMM2.weights_ = np.array([0.5, 0.5])
    GMM2.means_ = np.array([[0, 1], [1, 0]])
    GMM2.covariances_ = np.array([0.05, 0.05])

    X_train = np.concatenate([GMM1.sample(1000)[0], GMM2.sample(1000)[0]])
    Y_train = np.array([0.0] * 1000 + [1.0] * 1000)

    X_test = np.concatenate([GMM1.sample(100)[0], GMM2.sample(100)[0]])
    Y_test = np.array([0.0] * 100 + [1.0] * 100)

    return X_train, Y_train, X_test, Y_test


class MyNet(nn.Module):
    # 3层神经网络初始化函数
    def __init__(self):
        super(MyNet, self).__init__()
        self.feedforward1 = nn.Linear(2, 8)  # 输入层2维输入，8维输出
        self.feedforward2 = nn.Linear(8, 8)  # 隐藏层将第一层的8维输出作为输入，输出8维隐藏节点
        self.feedforward_output = nn.Linear(8, 1)  # 输出层将第二层的8维输出作为输入，输出一维节点

    # 神经网络的正向传播函数（推理过程）
    def forward(self, x):
        y = torch.sigmoid(self.feedforward1(x))
        y = torch.sigmoid(self.feedforward2(y))
        y = torch.sigmoid(self.feedforward_output(y))
        return y


if __name__ == "__main__":

    X_train, Y_train, X_test, Y_test = GenerateData()

    mynet = MyNet()  # 创建神经网络
    criterion = nn.MSELoss()  # 使用均方差损失函数
    optimizer = optim.Adam(mynet.parameters(), lr=0.1)  # 选择Adam优化算法, lr学习率
    num_steps = 200
    for epoch in range(num_steps):
        optimizer.zero_grad()
        outputs = torch.squeeze(
            mynet(torch.from_numpy(X_train).float())
        )
        loss = criterion(
            outputs, torch.from_numpy(Y_train).float()
        )
        loss.backward()  # 反向传播
        optimizer.step()  # 迭代
        print('epoch:', epoch, 'loss:', loss.item())

    Y_test_predict = np.squeeze(mynet(torch.from_numpy(X_test).float()).data.numpy())

    pyplot.subplot(2, 1, 1)
    pyplot.title('Ground truth')
    pyplot.plot(X_test[Y_test < 0.5, 0], X_test[Y_test < 0.5, 1], 'ro')
    pyplot.plot(X_test[Y_test >= 0.5, 0], X_test[Y_test >= 0.5, 1], 'bo')

    pyplot.subplot(2, 1, 2)
    pyplot.title('Predictions')
    pyplot.plot(X_test[Y_test_predict < 0.5, 0], X_test[Y_test_predict < 0.5, 1], 'ro')
    pyplot.plot(X_test[Y_test_predict >= 0.5, 0], X_test[Y_test_predict >= 0.5, 1], 'bo')

    pyplot.show()

使用循环神经网络

更换使用循环神经网络RNN模型，进行1维序列分类任务。

为了简化问题，我们假定：
序列的长度是固定的。我们将其长度设为T=4。
我们只有两个类别，所以这是一个二值分类的问题。
类别0表示序列是一个递增序列。例如[0.1, 0.2, 0.3, 0.4]，或者 [-0.5, 0.2, 3.4, 6.2]。
类别1表示序列是一个递减序列。例如[0.4, 0.3, 0.2, 0.1]，或者 [5.1, 4.2, 1.1, -4.3]。

该数据中的一些序列并不是严格递增或者严格递减的，而是人工添加了一些噪声用来增加难度。

生成pickle数据集

import numpy as np
import pickle
import random

n_train = 10000
n_test = 100

train_samples = []
train_labels = []
test_samples = []
test_labels = []

for i in range(n_train + n_test):
  if i < n_train:
    samples = train_samples
    labels = train_labels
  else:
    samples = test_samples
    labels = test_labels

  label = random.choice([0, 1])  # 创建标签
  start = random.uniform(-40.0, 40.0)  # 数据集值域
  sample = [start]
  next = start + random.uniform(-1, 5.0) * (-label * 2 + 1)  # 添加噪声数据
  sample.append(next)
  next = next + random.uniform(-2, 10.0) * (-label * 2 + 1)
  sample.append(next)
  next = next + random.uniform(-2, 10.0) * (-label * 2 + 1)
  sample.append(next)


  print("label:", label)
  print("sample:", sample)
  labels.append(label)
  samples.append(sample)

  data = {
    "train_samples": train_samples,
    "train_labels": train_labels,
    "test_samples": test_samples,
    "test_labels": test_labels,
  }

with open('data.pickle', 'wb') as f:
  pickle.dump(data, f)

数据预览：

...
label: 0
sample: [13.661404789383802, 16.25823434021173, 25.836030702074233, 29.80743875466039]
label: 0
sample: [-36.08627999686152, -34.34148605882467, -30.952065898268796, -24.826507234567075]
label: 0
sample: [29.623166657112577, 30.665468166023125, 39.4453612889517, 42.46217272533641]
label: 1
sample: [37.09265024612682, 34.9627251848789, 29.480888940841062, 29.03037901349507]
label: 1
sample: [-36.48292930041589, -36.2517650316274, -40.63062181571942, -44.71483963355948]
label: 0
sample: [-27.787531018991984, -25.49688544939839, -18.617854249376403, -16.697955035196976]
label: 1
sample: [-17.61357006905537, -19.71666704640389, -29.423836269702587, -34.81544894336959]
label: 0
sample: [-7.9647956435803735, -8.697630914998363, -3.043016298343562, 0.04254751655408029]
label: 1
sample: [3.1478411158586965, 1.4399195145968626, -3.1482779908675864, -7.777294451546566]
label: 0
sample: [-32.022188264777625, -31.690097636594118, -29.23549322202779, -27.041525433045393]
label: 1
sample: [-9.694922608645385, -10.338911638390199, -14.366987058323842, -18.46181546341211]
label: 1
sample: [23.440205855160336, 23.57074807644042, 17.634275111581218, 12.857671664491825]
label: 0
sample: [-19.175740480534067, -17.885292738296663, -17.089990831414497, -7.9255258345168595]
label: 1
sample: [-2.3408396273441525, -3.1195397721511418, -11.953863530700303, -20.060383291780298]
label: 0
sample: [-36.046741153634905, -31.262165117768486, -26.99303716939375, -18.734469192006976]
label: 1
sample: [-32.91181436961641, -33.15143640702147, -31.42082286186909, -29.983718884940473]
label: 1
sample: [-38.49879958380717, -38.615988339688904, -41.350719083767615, -41.62130549151447]

生成的pickle数据可以通过以下代码来读取：

import pickle

with open('data.pickle', mode='rb') as f:
  data = pickle.load(f)
  
train_samples = data['train_samples']
train_labels = data['train_labels']
test_samples = data['test_samples']
test_labels = data['test_labels']

构建RNN进行1维序列的训练、推理

import numpy as np
import pickle
import torch
import torch.nn as nn
import torch.optim as optim


def GetData():
  with open('data.pickle', mode='rb') as f:
    data = pickle.load(f)

  train_samples = np.array(data['train_samples'])
  train_labels = np.array(data['train_labels'])
  test_samples = np.array(data['test_samples'])
  test_labels = np.array(data['test_labels'])
  return train_samples, train_labels, test_samples, test_labels

class MyNet(nn.Module):
  def __init__(self):
    super(MyNet, self).__init__()
    # Define the RNN layer and a final linear layer
    self.rnn = nn.RNN(
      input_size=1,
      hidden_size=8,
      num_layers=1)
    self.feedforward_output = nn.Linear(
      in_features=8,
      out_features=1)

  def forward(self, x):
    h0 = torch.zeros(1, x.shape[1], 8)
    y, hn = self.rnn(x, h0)
    y = torch.sigmoid(self.feedforward_output(y))
    return y


def main():
  train_samples, train_labels, test_samples, test_labels = GetData()

  mynet = MyNet()

  # Train
  criterion = nn.BCELoss()
  optimizer = optim.Adam(mynet.parameters(), lr=0.1)
  num_steps = 100
  print('start training')
  for epoch in range(num_steps):
      optimizer.zero_grad()
      inputs = np.expand_dims(train_samples.transpose(), -1)
      outputs = torch.squeeze(
          mynet(torch.from_numpy(inputs).float()))
      last_outputs = outputs[-1, :]
      loss = criterion(
          last_outputs, torch.from_numpy(train_labels).float())
      loss.backward()
      optimizer.step()
      print('epoch:', epoch, 'loss:', loss.item())
  print('finished training')

  # Evaluate
  inputs = np.expand_dims(test_samples.transpose(), -1)
  test_predict = np.squeeze(mynet(torch.from_numpy(inputs).float()).data.numpy())
  last_predict = (test_predict[-1, :] > 0.5)
  accuracy = sum(last_predict == test_labels) / len(test_labels)
  print('accuracy on test data:', accuracy)

if __name__ == '__main__':
  main()

准确率：

...
epoch: 97 loss: 0.1513727456331253
epoch: 98 loss: 0.13302730023860931
epoch: 99 loss: 0.13972853124141693
finished training
accuracy on test data: 0.95

Reference:
[1] 王泉-声纹识别：从理论到编程实战
[2] 王泉-声纹识别GitHub代码仓