Implémentation de Modèles d'Apprentissage Profond avec MXNet: De la Régression Linéaire aux Réseaux Convolutifs

La régression linéaire constitue une base fondamentale en apprentissage profond. Voici commment générer un jeu de données synthétique et entraîner un modèle simple.

# Importation des bibliothèques nécessaires
from mxnet import ndarray as np
from mxnet import autograd as grad

# Paramètres du jeu de données
nb_entrees = 2
nb_echantillons = 1000
poids_vrais = [2, -3.4]
biais_vrai = 4.2

# Création des caractéristiques et des étiquettes
x = np.random_normal(shape=(nb_echantillons, nb_entrees))
y = poids_vrais[0] * x[:, 0] + poids_vrais[1] * x[:, 1] + biais_vrai
y += .01 * np.random_normal(shape=y.shape)
print(x[0:10], y[0:10])

# Itération sur les données par lots
import alea

taille_lot = 10

def iterateur_donnees():
    indices = list(range(nb_echantillons))
    alea.shuffle(indices)
    for i in range(0, nb_echantillons, taille_lot):
        j = np.array(indices[i:min(i + taille_lot, nb_echantillons)])
        yield np.take(x, j), np.take(y, j)

# Initialisation des paramètres du modèle
w = np.random_normal(shape=(nb_entrees, 1))
b = np.zeros((1,))
params = [w, b]
for param in params:
    param.attach_grad()

# Définition du modèle
def modele(x):
    return np.dot(x, w) + b

# Fonction de perte (erreur quadratique)
def perte_carre(y_pred, y_vrai):
    return (y_pred - y_vrai.reshape(y_pred.shape)) ** 2

# Optimisation par descente de gradient stochastique
def sgd(params, taux_apprentissage):
    for param in params:
        param[:] = param - taux_apprentissage * param.grad

# Boucle d'entraînement
epoques = 5
taux_apprentissage = 0.001
for epoque in range(epoques):
    perte_totale = 0
    for donnees, etiquettes in iterateur_donnees():
        with grad.record():
            sorties = modele(donnees)
            perte = perte_carre(sorties, etiquettes)
        perte.backward()
        sgd(params, taux_apprentissage)
        perte_totale += np.sum(perte).asscalar()
    print("Époque %d, perte moyenne: %f" % (epoque, perte_totale / nb_echantillons))

print("Poids vrais:", poids_vrais, "Poids appris:", w)
print("Biais vrai:", biais_vrai, "Biais appris:", b)

Pour une implémentasion plus concise, on peut utiliser l'API Gluon de MXNet.

from mxnet import ndarray as np
from mxnet import autograd
from mxnet import gluon

nb_entrees = 2
nb_echantillons = 1000
poids_vrais = [2, -3.4]
biais_vrai = 4.2

x = np.random_normal(shape=(nb_echantillons, nb_entrees))
y = poids_vrais[0] * x[:, 0] + poids_vrais[1] * x[:, 1] + biais_vrai
y += 0.01 * np.random_normal(shape=y.shape)

taille_lot = 10
jeu_donnees = gluon.data.ArrayDataset(x, y)
iterateur = gluon.data.DataLoader(jeu_donnees, taille_lot, shuffle=True)

# Création du réseau de neurones
reseau = gluon.nn.Sequential()
reseau.add(gluon.nn.Dense(1))
reseau.initialize()

# Définition de la perte et de l'optimiseur
perte_l2 = gluon.loss.L2Loss()
optimiseur = gluon.Trainer(reseau.collect_params(), 'sgd', {'learning_rate': 0.1})

# Entraînement
epoques = 5
for epoque in range(epoques):
    perte_totale = 0
    for donnees, etiquettes in iterateur:
        with autograd.record():
            previsions = reseau(donnees)
            perte = perte_l2(previsions, etiquettes)
        perte.backward()
        optimiseur.step(taille_lot)
        perte_totale += np.sum(perte).asscalar()
    print("Époque %d, perte moyenne: %f" % (epoque, perte_totale / nb_echantillons))

couche_dense = reseau[0]
print("Poids vrais:", poids_vrais, "Poids appris:", couche_dense.weight.data())
print("Biais vrai:", biais_vrai, "Biais appris:", couche_dense.bias.data())

Pour la classification d'images, on utilise la régression softmax. Prenons l'exemple du jeu de données Fashion-MNIST.

from mxnet import gluon
from mxnet import ndarray as np
import matplotlib.pyplot as plt

def transformer(donnees, etiquettes):
    return donnees.astype('float32') / 255, etiquettes.astype('float32')

mnist_train = gluon.data.vision.FashionMNIST(train=True, transform=transformer)
mnist_test = gluon.data.vision.FashionMNIST(train=False, transform=transformer)

# Visualisation des échantillons
def afficher_images(images):
    n = images.shape[0]
    _, figs = plt.subplots(1, n, figsize=(15, 15))
    for i in range(n):
        figs[i].imshow(images[i].reshape((28, 28)).asnumpy())
        figs[i].axes.get_xaxis().set_visible(False)
        figs[i].axes.get_yaxis().set_visible(False)
    plt.show()

def etiquettes_texte(etiquettes):
    noms = ['t-shirt', 'pantalon', 'pull-over', 'robe', 'manteau',
            'sandale', 'chemise', 'baskets', 'sac', 'botte']
    return [noms[int(i)] for i in etiquettes]

donnees, etiquettes = mnist_train[0:9]
afficher_images(donnees)
print(etiquettes_texte(etiquettes))

# Chargement des données par lots
taille_lot = 256
chargeur_train = gluon.data.DataLoader(mnist_train, taille_lot, shuffle=True)
chargeur_test = gluon.data.DataLoader(mnist_test, taille_lot, shuffle=False)

# Initialisation des paramètres pour le modèle
nb_entrees = 784
nb_sorties = 10
poids = np.random_normal(shape=(nb_entrees, nb_sorties))
biais = np.random_normal(shape=nb_sorties)
params = [poids, biais]
for param in params:
    param.attach_grad()

# Fonction softmax
def softmax(x):
    exp = np.exp(x)
    somme = exp.sum(axis=1, keepdims=True)
    return exp / somme

# Définition du modèle
def modele(x):
    return softmax(np.dot(x.reshape((-1, nb_entrees)), poids) + biais)

# Perte d'entropie croisée
def entropie_croisee(y_pred, y_vrai):
    return -np.pick(np.log(y_pred), y_vrai)

# Calcul de la précision
def precision(sorties, etiquettes):
    return np.mean(sorties.argmax(axis=1) == etiquettes).asscalar()

# Évaluation sur un jeu de données
def evaluer_precision(iterateur, modele):
    acc = 0.
    for donnees, etiquettes in iterateur:
        sorties = modele(donnees)
        acc += precision(sorties, etiquettes)
    return acc / len(iterateur)

# Entraînement
taux_apprentissage = .1
epoques = 10
for epoque in range(epoques):
    train_acc = 0.
    train_loss = 0.
    for donnees, etiquettes in chargeur_train:
        with autograd.record():
            sorties = modele(donnees)
            perte = entropie_croisee(sorties, etiquettes)
        perte.backward()
        for param in params:
            param[:] = param - taux_apprentissage * param.grad / taille_lot
        train_loss += np.mean(perte).asscalar()
        train_acc += precision(sorties, etiquettes)
    test_acc = evaluer_precision(chargeur_test, modele)
    print("Époque %d. Perte: %f, Précision train %f, Précision test %f" % (epoque, train_loss / len(chargeur_train), train_acc / len(chargeur_train), test_acc / len(chargeur_train)))

# Prédiction sur des échantillons
donnees, etiquettes = mnist_test[0:9]
afficher_images(donnees)
print('Étiquettes réelles:', etiquettes_texte(etiquettes))
previsions = modele(donnees).argmax(axis=1)
print('Étiquettes prédites:', etiquettes_texte(previsions.asnumpy()))

Les réseaux multicouches (MLP) ajoutent des couches cachées et des fonctinos d'activation pour capturer des relations non linéaires.

import d2lzh
from mxnet import autograd
from mxnet import ndarray as np

taille_lot = 256
chargeur_train, chargeur_test = d2lzh.load_data_fashion_mnist(taille_lot)

nb_entrees = 28 * 28
nb_sorties = 10
nb_caches = 256
echelle_poids = .01

w1 = np.random_normal(shape=(nb_entrees, nb_caches), scale=echelle_poids)
b1 = np.zeros(nb_caches)
w2 = np.random_normal(shape=(nb_caches, nb_sorties), scale=echelle_poids)
b2 = np.zeros(nb_sorties)

params = [w1, b1, w2, b2]
for param in params:
    param.attach_grad()

# Fonction d'activation ReLU
def relu(x):
    return np.maximum(x, 0)

# Définition du MLP
def mlp(x):
    x = x.reshape((-1, nb_entrees))
    h = relu(np.dot(x, w1) + b1)
    return np.dot(h, w2) + b2

# Perte softmax
perte_softmax = gluon.loss.SoftmaxCrossEntropyLoss()

def precision(y_pred, y_vrai):
    return (y_pred.argmax(axis=1) == y_vrai.astype('float32')).mean().asscalar()

# Entraînement
taux_apprentissage = .5
epoques = 5
for epoque in range(epoques):
    train_loss = 0.
    train_acc = 0.
    for donnees, etiquettes in chargeur_train:
        with autograd.record():
            sorties = mlp(donnees)
            perte = perte_softmax(sorties, etiquettes)
        perte.backward()
        d2lzh.sgd(params, taux_apprentissage, taille_lot)
        train_loss += np.mean(perte).asscalar()
        train_acc += precision(sorties, etiquettes)
    test_acc = d2lzh.evaluate_accuracy(chargeur_test, mlp)
    print("Époque %d. Perte: %f, Précision train %f, Précision test %f" % (epoque, train_loss / len(chargeur_train), train_acc / len(chargeur_train), test_acc / len(chargeur_train)))

La sélection de modèle aborde les problèmes de sous-apprentissage et de sur-apprentissage. Un exemple avec des données polynomiales.

from mxnet import autograd
from mxnet import ndarray as np
from mxnet import gluon
import matplotlib.pyplot as plt

nb_train = 100
nb_test = 100
poids_vrais = [1.2, -3.4, 5.6]
biais_vrai = 5.0

X = np.random_normal(shape=(nb_train + nb_test, 1))
x = np.concat(X, np.power(X, 2), np.power(X, 3))
y = poids_vrais[0] * x[:, 0] + poids_vrais[1] * x[:, 1] + poids_vrais[2] * x[:, 2] + biais_vrai
y += .1 * np.random_normal(shape=y.shape)
y_train, y_test = y[:nb_train], y[nb_train:]

def perte_carre(y_pred, y_vrai):
    return (y_pred - y_vrai.reshape(y_pred.shape)) ** 2 / 2

def tester_modele(reseau, x, y):
    return perte_carre(reseau(x), y).mean().asscalar()

def entrainer_modele(x_train, x_test, y_train, y_test):
    reseau = gluon.nn.Sequential()
    with reseau.name_scope():
        reseau.add(gluon.nn.Dense(1))
    reseau.initialize()
    
    taux_apprentissage = 0.01
    epoques = 100
    taille_lot = 10
    
    jeu_train = gluon.data.ArrayDataset(x_train, y_train)
    iterateur_train = gluon.data.DataLoader(jeu_train, taille_lot, shuffle=True)
    
    optimiseur = gluon.Trainer(reseau.collect_params(), 'sgd', {'learning_rate': taux_apprentissage})
    perte_l2 = gluon.loss.L2Loss()
    
    pertes_train = []
    pertes_test = []
    
    for epoque in range(epoques):
        for donnees, etiquettes in iterateur_train:
            with autograd.record():
                previsions = reseau(donnees)
                perte = perte_l2(previsions, etiquettes)
            perte.backward()
            optimiseur.step(taille_lot)
        pertes_train.append(perte_l2(reseau(x_train), y_train).mean().asscalar())
        pertes_test.append(perte_l2(reseau(x_test), y_test).mean().asscalar())
    
    plt.plot(pertes_train)
    plt.plot(pertes_test)
    plt.legend(['train', 'test'])
    plt.show()
    
    return ('poids appris', reseau[0].weight.data(), 'biais appris', reseau[0].bias.data())

# Ajustement polynomial
entrainer_modele(x[:nb_train, :], x[nb_train:, :], y[:nb_train], y[nb_train:])

La régularisation, comme la pénalité L2, aide à prévenir le sur-apprentissage en pénalisant les poids importants.

from mxnet import ndarray as np
from mxnet import autograd
import alea
import matplotlib.pyplot as plt

nb_train = 20
nb_test = 100
nb_entrees = 200
poids_vrais = np.ones((nb_entrees, 1)) * 0.01
biais_vrai = 0.05

x = np.random_normal(shape=(nb_train + nb_test, nb_entrees))
y = np.dot(x, poids_vrais)
y += .01 * np.random_normal(shape=y.shape)

x_train, x_test = x[:nb_train, :], x[nb_train:, :]
y_train, y_test = y[:nb_train], y[nb_train:]

taille_lot = 1

def iterateur_donnees(nb_exemples):
    indices = list(range(nb_exemples))
    alea.shuffle(indices)
    for i in range(0, nb_exemples, taille_lot):
        j = np.array(indices[i:min(i + taille_lot, nb_exemples)])
        yield x.take(j), y.take(j)

def initialiser_params():
    w = np.random_normal(shape=(nb_entrees, 1)) * 0.01
    b = np.zeros((1,))
    for param in (w, b):
        param.attach_grad()
    return (w, b)

def modele(x, lambd, w, b):
    return np.dot(x, w) + b

def perte_carre(y_pred, y_vrai):
    return (y_pred - y_vrai.reshape(y_pred.shape)) ** 2

def sgd(params, taux):
    for param in params:
        param[:] = param - taux * param.grad

def tester(params, x, y):
    return perte_carre(modele(x, 0, *params), y).mean().asscalar()

def entrainer(lambd):
    epoques = 10
    taux_apprentissage = 0.003
    params = initialiser_params()
    pertes_train = []
    pertes_test = []
    for epoque in range(epoques):
        for donnees, etiquettes in iterateur_donnees(nb_train):
            with autograd.record():
                previsions = modele(donnees, lambd, *params)
                (w, b) = params
                perte = perte_carre(previsions, etiquettes) + lambd * ((w ** 2).sum() + b ** 2)
            perte.backward()
            sgd(params, taux_apprentissage)
        pertes_train.append(tester(params, x_train, y_train))
        pertes_test.append(tester(params, x_test, y_test))
    plt.plot(pertes_train)
    plt.plot(pertes_test)
    plt.legend(['train', 'test'])
    plt.show()
    print('poids appris[:10]', params[0][:10], 'biais appris:', params[1])

# Sans régularisation (sur-apprentissage)
entrainer(0)
# Avec régularisation L2
entrainer(2)

Dans le cadre de la prédiction de prix immobiliers (Kaggle), on utilise la validation croisée et l'optimisation Adam.

import d2lzh as d2l
from mxnet import autograd, gluon, nd
from mxnet.gluon import data as gdata, loss as gloss, nn
import numpy as np
import pandas as pd

donnees_train = pd.read_csv('train.csv')
donnees_test = pd.read_csv('test.csv')

toutes_carac = pd.concat((donnees_train.iloc[:, 1:-1], donnees_test.iloc[:, 1:]))
carac_numeriques = toutes_carac.dtypes[toutes_carac.dtypes != 'object'].index
toutes_carac[carac_numeriques] = toutes_carac[carac_numeriques].apply(lambda x: (x - x.mean()) / (x.std()))
toutes_carac[carac_numeriques] = toutes_carac[carac_numeriques].fillna(0)
toutes_carac = pd.get_dummies(toutes_carac, dummy_na=True)

nb_train = donnees_train.shape[0]
features_train = nd.array(toutes_carac[:nb_train].values)
features_test = nd.array(toutes_carac[nb_train:].values)
etiquettes_train = nd.array(donnees_train.SalePrice.values).reshape((-1, 1))

perte = gloss.L2Loss()

def creer_reseau():
    reseau = nn.Sequential()
    reseau.add(nn.Dense(1))
    reseau.initialize()
    return reseau

def rmse_log(reseau, features, etiquettes):
    preds_clippees = nd.clip(reseau(features), 1, float('inf'))
    return nd.sqrt(2 * perte(preds_clippees.log(), etiquettes.log()).mean()).asscalar()

def entrainer(reseau, features_train, etiquettes_train, features_test, etiquettes_test, epoques, taux_apprentissage, weight_decay, taille_lot):
    iterateur_train = gdata.DataLoader(gdata.ArrayDataset(features_train, etiquettes_train), taille_lot, shuffle=True)
    optimiseur = gluon.Trainer(reseau.collect_params(), 'adam', {'learning_rate': taux_apprentissage, 'wd': weight_decay})
    
    pertes_train = []
    pertes_test = []
    for epoque in range(epoques):
        for donnees, etiquettes in iterateur_train:
            with autograd.record():
                perte_lot = perte(reseau(donnees), etiquettes)
            perte_lot.backward()
            optimiseur.step(taille_lot)
        pertes_train.append(rmse_log(reseau, features_train, etiquettes_train))
        if etiquettes_test is not None:
            pertes_test.append(rmse_log(reseau, features_test, etiquettes_test))
    return pertes_train, pertes_test

# Validation croisée k-fold
def validation_croisee(k, features_train, etiquettes_train, epoques, taux_apprentissage, weight_decay, taille_lot):
    taille_fold = features_train.shape[0] // k
    moy_train, moy_valid = 0, 0
    for i in range(k):
        idx_debut = i * taille_fold
        idx_fin = (i + 1) * taille_fold
        features_valid = features_train[idx_debut:idx_fin, :]
        etiquettes_valid = etiquettes_train[idx_debut:idx_fin]
        features_train_fold = nd.concat(features_train[:idx_debut, :], features_train[idx_fin:, :], dim=0)
        etiquettes_train_fold = nd.concat(etiquettes_train[:idx_debut], etiquettes_train[idx_fin:])
        reseau = creer_reseau()
        p_train, p_valid = entrainer(reseau, features_train_fold, etiquettes_train_fold, features_valid, etiquettes_valid, epoques, taux_apprentissage, weight_decay, taille_lot)
        moy_train += p_train[-1]
        moy_valid += p_valid[-1]
        print('Fold %d, rmse train %f, rmse valid %f' % (i, p_train[-1], p_valid[-1]))
    return moy_train / k, moy_valid / k

k = 5
epoques = 10000
taux = 5
weight_decay = 0
taille_lot = 64
train_rmse, valid_rmse = validation_croisee(k, features_train, etiquettes_train, epoques, taux, weight_decay, taille_lot)
print('Validation croisée %d folds: moyenne train rmse %f, moyenne valid rmse %f' % (k, train_rmse, valid_rmse))

Les concepts de convolution incluent le remplissage (padding) et le pas (stride) pour contrôler la taille des sorties.

from mxnet import nd
from mxnet.gluon import nn

def tester_convolution(conv2d, x):
    conv2d.initialize()
    x = x.reshape((1, 1) + x.shape)
    y = conv2d(x)
    return y.reshape(y.shape[2:])

x = nd.random.uniform(shape=(8, 8))
conv2d = nn.Conv2D(1, kernel_size=3, padding=1)
print(tester_convolution(conv2d, x).shape)

conv2d = nn.Conv2D(1, kernel_size=(5, 3), padding=(2, 1))
print(tester_convolution(conv2d, x).shape)

conv2d = nn.Conv2D(1, kernel_size=3, padding=1, strides=2)
print(tester_convolution(conv2d, x).shape)

Pour les canaux multiples en entrée et en sortie, on étend les opérations de convolution.

import d2lzh as d2l
from mxnet import nd

def correlation2d_multi_entree(X, K):
    return nd.add_n(*[d2l.corr2d(x, k) for x, k in zip(X, K)])

X = nd.array([[[0, 1, 2], [3, 4, 5], [6, 7, 8]], [[1, 2, 3], [4, 5, 6], [7, 8, 9]]])
K = nd.array([[[0, 1], [2, 3]], [[1, 2], [3, 4]]])

y = correlation2d_multi_entree(X, K)
print(y)

def correlation2d_multi_sortie(X, K):
    return nd.stack(*[correlation2d_multi_entree(X, k) for k in K])

K = nd.stack(K, K + 1, K + 2)
print(correlation2d_multi_sortie(X, K))

Étiquettes: MXNet Gluon Régression Linéaire Classification Réseau de Neurones Multicouche

Publié le 15 juillet à 15h02