autoencoder.ipynb

Kernel: Python 3

Copyright 2020 The TensorFlow Authors.

#@title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Introducción a los autocodificadores

Ver en TensorFlow.org

Ejecutar en Google Colab

Ver código fuente en GitHub

Descargar notebook

En este tutorial, se presentan los autocodificadores con tres ejemplos: fundamentos, eliminación de ruidos y detección de anomalías.

Un autocodificador es un tipo de red neuronal especial que es entrenado para que copie su propia entrada en la salida. Por ejemplo, si se ingresa una imagen con una cifra escrita a mano, un autocodificador primero codifica la imagen en una representación latente de menor dimensión, luego vuelve a decodificar la representación latente en una imagen. Un autocodificador aprende a comprimir los datos mientras minimiza errores de reconstrucción.

Para obtener más información sobre los autocodificadores, considere leer el capítulo 14 de Aprendizaje profundo de Ian Goodfellow, Yoshua Bengio, y Aaron Courville.

Importar TensorFlow y otras bibliotecas

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import tensorflow as tf

from sklearn.metrics import accuracy_score, precision_score, recall_score
from sklearn.model_selection import train_test_split
from tensorflow.keras import layers, losses
from tensorflow.keras.datasets import fashion_mnist
from tensorflow.keras.models import Model

Cargar el conjunto de datos

Para empezar, entrenará un autocodificador básico con el conjunto de datos MNIST de moda. Cada imagen en el conjunto de datos es de 28x28 píxeles.

(x_train, _), (x_test, _) = fashion_mnist.load_data()

x_train = x_train.astype('float32') / 255.
x_test = x_test.astype('float32') / 255.

print (x_train.shape)
print (x_test.shape)

Primer ejemplo: autocodificador básico

Defina un autocodificador con dos capas densas: un encoder, que comprime las imágenes en un vector latente de 64 dimensiones y un decoder, que reconstruye la imagen original desde el espacio latente.

Para definir su modelo, use la API de subclasificación del modelo de Keras.

class Autoencoder(Model):
  def __init__(self, latent_dim, shape):
    super(Autoencoder, self).__init__()
    self.latent_dim = latent_dim
    self.shape = shape
    self.encoder = tf.keras.Sequential([
      layers.Flatten(),
      layers.Dense(latent_dim, activation='relu'),
    ])
    self.decoder = tf.keras.Sequential([
      layers.Dense(tf.math.reduce_prod(shape), activation='sigmoid'),
      layers.Reshape(shape)
    ])

  def call(self, x):
    encoded = self.encoder(x)
    decoded = self.decoder(encoded)
    return decoded


shape = x_test.shape[1:]
latent_dim = 64
autoencoder = Autoencoder(latent_dim, shape)

autoencoder.compile(optimizer='adam', loss=losses.MeanSquaredError())

Entrene el modelo con x_train como la entrada y la salida. El encoder aprenderá a comprimir los conjuntos de datos de 784 dimensiones en el espacio latente, y el decoder aprenderá a reconstruir las imágenes originales.

autoencoder.fit(x_train, x_train,
                epochs=10,
                shuffle=True,
                validation_data=(x_test, x_test))

Ahora que el modelo está entrenado, probemos la codificación y decodificación de imágenes en el conjunto de prueba.

encoded_imgs = autoencoder.encoder(x_test).numpy()
decoded_imgs = autoencoder.decoder(encoded_imgs).numpy()

n = 10
plt.figure(figsize=(20, 4))
for i in range(n):
  # display original
  ax = plt.subplot(2, n, i + 1)
  plt.imshow(x_test[i])
  plt.title("original")
  plt.gray()
  ax.get_xaxis().set_visible(False)
  ax.get_yaxis().set_visible(False)

  # display reconstruction
  ax = plt.subplot(2, n, i + 1 + n)
  plt.imshow(decoded_imgs[i])
  plt.title("reconstructed")
  plt.gray()
  ax.get_xaxis().set_visible(False)
  ax.get_yaxis().set_visible(False)
plt.show()

Segundo ejemplo: eliminación de ruidos de la imagen

También se puede entrenar un autocodificador para eliminar el ruido de las imágenes. En la siguiente sección, creará una versión con ruido del conjunto de datos MNIST de moda y aplicará ruido aleatorio en cada imagen. Luego, entrenará al autocodificador con una imagen con ruido como entrada y la imagen original como el destino.

Importaremos el conjunto de datos para omitir las modificaciones hechas anteriormente.

(x_train, _), (x_test, _) = fashion_mnist.load_data()

x_train = x_train.astype('float32') / 255.
x_test = x_test.astype('float32') / 255.

x_train = x_train[..., tf.newaxis]
x_test = x_test[..., tf.newaxis]

print(x_train.shape)

Agregamos ruido aleatorio a las imágenes.

noise_factor = 0.2
x_train_noisy = x_train + noise_factor * tf.random.normal(shape=x_train.shape) 
x_test_noisy = x_test + noise_factor * tf.random.normal(shape=x_test.shape) 

x_train_noisy = tf.clip_by_value(x_train_noisy, clip_value_min=0., clip_value_max=1.)
x_test_noisy = tf.clip_by_value(x_test_noisy, clip_value_min=0., clip_value_max=1.)

Trazamos las imágenes con ruido.

n = 10
plt.figure(figsize=(20, 2))
for i in range(n):
    ax = plt.subplot(1, n, i + 1)
    plt.title("original + noise")
    plt.imshow(tf.squeeze(x_test_noisy[i]))
    plt.gray()
plt.show()

Definir un autocodificador convolucional

En este ejemplo, entrenará un autocodificador convolucional con las capas Conv2D en el encoder, y las capas Conv2DTranspose en el decoder.

class Denoise(Model):
  def __init__(self):
    super(Denoise, self).__init__()
    self.encoder = tf.keras.Sequential([
      layers.Input(shape=(28, 28, 1)),
      layers.Conv2D(16, (3, 3), activation='relu', padding='same', strides=2),
      layers.Conv2D(8, (3, 3), activation='relu', padding='same', strides=2)])

    self.decoder = tf.keras.Sequential([
      layers.Conv2DTranspose(8, kernel_size=3, strides=2, activation='relu', padding='same'),
      layers.Conv2DTranspose(16, kernel_size=3, strides=2, activation='relu', padding='same'),
      layers.Conv2D(1, kernel_size=(3, 3), activation='sigmoid', padding='same')])

  def call(self, x):
    encoded = self.encoder(x)
    decoded = self.decoder(encoded)
    return decoded

autoencoder = Denoise()

autoencoder.compile(optimizer='adam', loss=losses.MeanSquaredError())

autoencoder.fit(x_train_noisy, x_train,
                epochs=10,
                shuffle=True,
                validation_data=(x_test_noisy, x_test))

Veamos un resumen del codificador. Note cómo se reduce el tamaño de las imágenes de 28x28 a 7x7.

autoencoder.encoder.summary()

El decodificador aumenta el tamaño de las imágenes de 7x7 a 28x28.

autoencoder.decoder.summary()

Se trazan las imágenes con ruido y las imágenes sin ruido generadas por el autocodificador.

encoded_imgs = autoencoder.encoder(x_test_noisy).numpy()
decoded_imgs = autoencoder.decoder(encoded_imgs).numpy()

n = 10
plt.figure(figsize=(20, 4))
for i in range(n):

    # display original + noise
    ax = plt.subplot(2, n, i + 1)
    plt.title("original + noise")
    plt.imshow(tf.squeeze(x_test_noisy[i]))
    plt.gray()
    ax.get_xaxis().set_visible(False)
    ax.get_yaxis().set_visible(False)

    # display reconstruction
    bx = plt.subplot(2, n, i + n + 1)
    plt.title("reconstructed")
    plt.imshow(tf.squeeze(decoded_imgs[i]))
    plt.gray()
    bx.get_xaxis().set_visible(False)
    bx.get_yaxis().set_visible(False)
plt.show()

Tercer ejemplo: detección de anomalías

Descripción general

En este ejemplo, entrenará un autocodificador para detectar anomalías en el conjunto de datos ECG5000. Este conjunto de datos contiene 5000 electrocardiogramas, cada uno con 140 puntos de datos. Usará una versión simplificada del conjunto de datos, donde cada ejemplo fue etiquetado con el número 0 (que corresponde a un ritmo cardíaco anormal) o con el número 1 (que corresponde a un ritmo cardíaco normal). Lo que se busca es identificar los ritmos anormales.

Nota: Este es un conjunto de datos etiquetados, así que podría formularse como un problema de aprendizaje supervisado. El objetivo de este ejemplo es ilustrar los conceptos de detección de anomalías que pueden aplicarse a conjuntos de datos más grandes, donde no hay etiquetas disponibles (por ejemplo, si tiene miles de ritmos cardíacos normales y solo una pequeña cantidad de ritmos cardíacos anormales).

¿Cómo se detectan las anomalías con el autocodificador? Recuerde que un autocodificador está entrenado para minimizar los errores de la reconstrucción. Entrenará un autocodificador solo con los ritmos cardíacos normales, y lo usará para reconstruir todos los datos. Nuestra hipótesis es que los ritmos cardíacos anormales presentarán más errores de reconstrucción. Luego de esto, usted clasificará un ritmo como anormal si el error de reconstrucción supera un umbral fijo.

Cargar datos ECG

El conjunto de datos que usará se basa en uno de timeseriesclassification.com.

# Download the dataset
dataframe = pd.read_csv('http://storage.googleapis.com/download.tensorflow.org/data/ecg.csv', header=None)
raw_data = dataframe.values
dataframe.head()

# The last element contains the labels
labels = raw_data[:, -1]

# The other data points are the electrocadriogram data
data = raw_data[:, 0:-1]

train_data, test_data, train_labels, test_labels = train_test_split(
    data, labels, test_size=0.2, random_state=21
)

Normalizar los datos en [0,1].

min_val = tf.reduce_min(train_data)
max_val = tf.reduce_max(train_data)

train_data = (train_data - min_val) / (max_val - min_val)
test_data = (test_data - min_val) / (max_val - min_val)

train_data = tf.cast(train_data, tf.float32)
test_data = tf.cast(test_data, tf.float32)

Entrenará al autocodificador solo con los ritmos cardíacos normales, que están etiquetados en el conjunto de datos con el número 1. Separe los ritmos normales de los ritmos anormales.

train_labels = train_labels.astype(bool)
test_labels = test_labels.astype(bool)

normal_train_data = train_data[train_labels]
normal_test_data = test_data[test_labels]

anomalous_train_data = train_data[~train_labels]
anomalous_test_data = test_data[~test_labels]

Trace un ECG normal.

plt.grid()
plt.plot(np.arange(140), normal_train_data[0])
plt.title("A Normal ECG")
plt.show()

Trace un ECG anormal.

plt.grid()
plt.plot(np.arange(140), anomalous_train_data[0])
plt.title("An Anomalous ECG")
plt.show()

Construir el modelo

class AnomalyDetector(Model):
  def __init__(self):
    super(AnomalyDetector, self).__init__()
    self.encoder = tf.keras.Sequential([
      layers.Dense(32, activation="relu"),
      layers.Dense(16, activation="relu"),
      layers.Dense(8, activation="relu")])
    
    self.decoder = tf.keras.Sequential([
      layers.Dense(16, activation="relu"),
      layers.Dense(32, activation="relu"),
      layers.Dense(140, activation="sigmoid")])
    
  def call(self, x):
    encoded = self.encoder(x)
    decoded = self.decoder(encoded)
    return decoded

autoencoder = AnomalyDetector()

autoencoder.compile(optimizer='adam', loss='mae')

Note como el autocodificador está entrenado solo con los ECG normales, pero se evalúa con todo el conjunto de prueba.

history = autoencoder.fit(normal_train_data, normal_train_data, 
          epochs=20, 
          batch_size=512,
          validation_data=(test_data, test_data),
          shuffle=True)

plt.plot(history.history["loss"], label="Training Loss")
plt.plot(history.history["val_loss"], label="Validation Loss")
plt.legend()

Pronto clasificará un ECG como anormal si el error de reconstrucción es mayor a una desviación estándar de los ejemplos de entrenamiento normales. Primero, trazaremos un ECG normal con el conjunto de entrenamiento, la reconstrucción después de que el autocodificador lo codifique y descodifique y el error de reconstrucción.

encoded_data = autoencoder.encoder(normal_test_data).numpy()
decoded_data = autoencoder.decoder(encoded_data).numpy()

plt.plot(normal_test_data[0], 'b')
plt.plot(decoded_data[0], 'r')
plt.fill_between(np.arange(140), decoded_data[0], normal_test_data[0], color='lightcoral')
plt.legend(labels=["Input", "Reconstruction", "Error"])
plt.show()

Cree un trazado similar, pero ahora para un ejemplo de prueba anormal.

encoded_data = autoencoder.encoder(anomalous_test_data).numpy()
decoded_data = autoencoder.decoder(encoded_data).numpy()

plt.plot(anomalous_test_data[0], 'b')
plt.plot(decoded_data[0], 'r')
plt.fill_between(np.arange(140), decoded_data[0], anomalous_test_data[0], color='lightcoral')
plt.legend(labels=["Input", "Reconstruction", "Error"])
plt.show()

Detectar anomalías

Detecte anomalías al calcular si la pérdida de la reconstrucción es mayor al umbral fijo. En este tutorial, calculará el error promedio de la media para los ejemplos normales del conjunto de entrenamiento, luego clasificará los ejemplos futuros como anormales si el error de reconstrucción es mayor que la desviación estándar del conjunto de entrenamiento.

Trace el error de reconstrucción en los ECG normales del conjunto de entrenamiento.

reconstructions = autoencoder.predict(normal_train_data)
train_loss = tf.keras.losses.mae(reconstructions, normal_train_data)

plt.hist(train_loss[None,:], bins=50)
plt.xlabel("Train loss")
plt.ylabel("No of examples")
plt.show()

Escoja un valor de umbral que esté una desviación estándar por sobre la media.

threshold = np.mean(train_loss) + np.std(train_loss)
print("Threshold: ", threshold)

Nota: Existen otras estrategias que se pueden usar para seleccionar un valor de umbral que se mencionó anteriormente, que debería clasificar a los ejemplos de prueba en anormales, el enfoque correcto dependerá de su conjunto de datos. Puede obtener más información en los enlaces al final de este tutorial.

SI examina el error de reconstrucción para los ejemplos anormales en el conjunto de prueba, notará que la mayoría tiene un valor de error de reconstrucción más alto que el umbral. Al variar el umbral, puede ajustar la precisión y coincidencia de su clasificador.

reconstructions = autoencoder.predict(anomalous_test_data)
test_loss = tf.keras.losses.mae(reconstructions, anomalous_test_data)

plt.hist(test_loss[None, :], bins=50)
plt.xlabel("Test loss")
plt.ylabel("No of examples")
plt.show()

Clasifique un ECG como anormal si el valor de error de reconstrucción es mayor que el umbral.

def predict(model, data, threshold):
  reconstructions = model(data)
  loss = tf.keras.losses.mae(reconstructions, data)
  return tf.math.less(loss, threshold)

def print_stats(predictions, labels):
  print("Accuracy = {}".format(accuracy_score(labels, predictions)))
  print("Precision = {}".format(precision_score(labels, predictions)))
  print("Recall = {}".format(recall_score(labels, predictions)))

preds = predict(autoencoder, test_data, threshold)
print_stats(preds, test_labels)

Próximos pasos

Para obtener más información sobre la detección de anomalías con autocodificadores, échele un vistazo a este ejemplo interactivo generado con TensorFlow.js de Victor Dibia. Para ver un caso del mundo real, puede aprender cómo Airbus detecta anomalías en datos de telemetría de ISS con TensorFlow. Para más información sobre los fundamentos, considere esta entrada de blog de François Chollet. Para más detalles, lea el capítulo 14 de Aprendizaje profundo de Ian Goodfellow, Yoshua Bengio, y Aaron Courville.