I’ve been experimenting with Transformer Architecture (TA) neural networks for several months. I reached a milestone recently when I created an end-to-end demo of using PyTorch TA for unsupervised anomaly detection.

Briefly, source data is fed to a TA network which creates a condensed latent representation of the data. The network is trained to reproduce its input. Put another way, the network is a TA-based autoencoder. After training, data items with large reconstruction error are tagged as anomalous.

For my demo, I used the UCI Digits dataset. Each data item is a crude 8 by 8 image of a handwritten digit from ‘0’ to ‘9’. The UCI Digits are essentially a scaled-down version of the MNIST dataset. Each of the 64 pixels of a UCI Digit data item are values between 0 and 16 (rather than 0 to 255 in MNIST).

The UCI Digits data has 3823 training items and 1797 test items. I used a 100-item subset of the training data because TA networks require lots of processing (due to the attention mechanism) and so training can be very slow.

My demo program is only about 200 lines of code. But the code is extremely dense and there are many tricky details.

The key code in the TA network definition is:

def forward(self, x):
  # x is torch.Size([bs, 64])
  # encode phase
  z = self.embed(x)         # [bs, 64, 4]
  z = z.reshape(-1, 64, 4)  # [bs, 64, 4]
  z = self.pos_enc(z)       # [bs, 64, 4]
  z = self.trans_enc(z)     # [bs, 64, 4]
 
  # decode phase
  z = z.reshape(-1, 4*64)   # [bs, 256]
  z = T.tanh(self.fc1(z))   # [bs, 128]
  z = self.fc2(z)           # [bs, 64]
  return z 

The input has 64 integer values (similar to word tokens). Each integer token is converted to 4 float values (similar to a word embedding). The embedded values are augmented with trigonometry positional encoding values. The result is fed to a transformer encoder layer. The output of the transformer encoder is passed to two standard fully connected neural layers that produce an output with 64 values (the same size as the input).

The network is trained using mean squared error loss between the source input and the generated output.

There are dozens of significant design alternatives that I need to explore. For example, using a transformer decoder instead of fully connected linear layers for the decode phase.

I showed this code to some of my work colleagues (Ricky L, Raja D, Bryan X) and they pointed out that I should do some experiments to determine if the technique can find anomalies. Put another way, the code runs but does it actually do what it’s supposed to do?

Anyway, good fun. Getting everything to work was an interesting challenge.

There have been hundreds of superheroes in comic books who could transform from ordinary men to, well, super guys. Here are three relatively obscure insect-based transformation heroes.

Right: Bee-Man appeared in the mid-1960s He was a NASA technician named Barry Eames. He gained powers after being stung by bees returned from Mars. He was originally evil but turned good.

Demo code. Replace “lt”, “gt”, “lte”, “gte” with Boolean symbol operators. NOTE: This code works but is so complicated that it almost certainly has quite a few bugs.

# experiment.py

# Transformer based reconstruction error
# PyTorch 1.10.0-CPU Anaconda3-2020.02  Python 3.7.6
# Windows 10/11 

import numpy as np
import matplotlib.pyplot as plt
import torch as T

device = T.device('cpu') 

# -----------------------------------------------------------

class UCI_Digits_Dataset(T.utils.data.Dataset):
  # like 8,12,0,16, . . 15,7
  # 64 pixel values [0-16], digit [0-9]

  def __init__(self, src_file):
    tmp_xy = np.loadtxt(src_file, usecols=range(0,65),
      delimiter=",", comments="#", dtype=np.int64)
    tmp_x = tmp_xy[:,0:64]
    # tmp_x /= 16.0  # no normalization for this scenario
    tmp_y = tmp_xy[:,64]

    self.x_data = T.tensor(tmp_x, dtype=T.int64).to(device)
    self.y_data = T.tensor(tmp_y, dtype=T.int64).to(device)

  def __len__(self):
    return len(self.x_data)

  def __getitem__(self, idx):
    pixels = self.x_data[idx]
    label = self.y_data[idx]
    return (pixels, label)  # as a tuple

# -----------------------------------------------------------

class Transformer_Net(T.nn.Module):
  def __init__(self):
    # vocab_size = 17
    # embed_dim = 4
    # seq_len = 64 (no label)
    super(Transformer_Net, self).__init__()
    self.embed = T.nn.Embedding(17, 4)  # pseudo word embed

    self.pos_enc = \
      PositionalEncoding(4, dropout=0.00)  # positional

    self.enc_layer = T.nn.TransformerEncoderLayer(d_model=4,
      nhead=2, dim_feedforward=100, 
      batch_first=True)  # d_model divisible by nhead

    self.trans_enc = T.nn.TransformerEncoder(self.enc_layer,
      num_layers=6)

    self.fc1 = T.nn.Linear(4*64, 128)  # 256-128
    self.fc2 = T.nn.Linear(128, 64)  # output size = input
 
  def forward(self, x):
    # x is torch.Size([bs, 64])
    # encode phase
    z = self.embed(x)         # [bs, 64, 4]
    z = z.reshape(-1, 64, 4)  # [bs, 64, 4]
    z = self.pos_enc(z)       # [bs, 64, 4]
    z = self.trans_enc(z)     # [bs, 64, 4]
 
    # decode phase
    z = z.reshape(-1, 4*64)   # [bs, 256]
    z = T.tanh(self.fc1(z))   # [bs, 128]
    z = self.fc2(z)           # [bs, 64]
    return z 

# -----------------------------------------------------------

class PositionalEncoding(T.nn.Module):  # documentation code
  def __init__(self, d_model: int, dropout: float=0.1,
   max_len: int=5000):
    super(PositionalEncoding, self).__init__()  # old syntax
    self.dropout = T.nn.Dropout(p=dropout)
    pe = T.zeros(max_len, d_model)  # like 10x4
    position = \
      T.arange(0, max_len, dtype=T.float).unsqueeze(1)
    div_term = T.exp(T.arange(0, d_model, 2).float() * \
      (-np.log(10_000.0) / d_model))
    pe[:, 0::2] = T.sin(position * div_term)
    pe[:, 1::2] = T.cos(position * div_term)
    pe = pe.unsqueeze(0).transpose(0, 1)
    self.register_buffer('pe', pe)  # allows state-save

  def forward(self, x):
    x = x + self.pe[:x.size(0), :]
    return self.dropout(x)

# -----------------------------------------------------------

def make_err_list(model, ds):
  # assumes model.eval()
  result_lst = []
  n_features = 64
  ldr = T.utils.data.DataLoader(ds, batch_size=1,
    shuffle=False)
  for bix, batch in enumerate(ldr):
    X = batch[0]  # the inputs
    with T.no_grad():
      Y = model(X)

    err = T.sum((X-Y)*(X-Y)).item()  # SSE all features
    err /= n_features                # norm'ed SSE
    result_lst.append( (bix,err) )   # idx data item, err

  return result_lst 

# -----------------------------------------------------------

def display_digit(ds, idx):
  # ds is a PyTorch Dataset
  data = ds[idx][0]  # [0] is the pixels, [1] is the label
  pixels = np.array(data)  # tensor to numpy
  pixels = pixels.reshape((8,8))
  for i in range(8):
    for j in range(8):
      pxl = pixels[i,j]  # or [i][j] syntax
      # print("%.2X" % pxl, end="")  # hexidecimal
      print("%3d" % pxl, end="")
    print("")

  plt.imshow(pixels, cmap=plt.get_cmap('gray_r'))
  plt.show() 
  plt.close() 

# -----------------------------------------------------------

def main():
  # 0. get started
  print("\nBegin Transformer based anomaly experiment ")
  T.manual_seed(1)
  np.random.seed(1)

  # 1. create Dataset object
  print("\nLoading UCI digits data ")
  train_data = ".\\Data\\uci_digits_train_100.txt"
  # train_data = ".\\Data\\optdigits_train_3823.txt"
  train_ds = UCI_Digits_Dataset(train_data)
  bat_size = 2
  train_ldr = T.utils.data.DataLoader(train_ds,
    batch_size=bat_size, shuffle=True)

  # 2. create network
  print("\nCreating Transformer autoencoder ")
  net = Transformer_Net().to(device)
  net.train()  # set mode

  # 3. train 
  loss_func = T.nn.MSELoss()
  lrn_rate = 0.01
  opt = T.optim.SGD(net.parameters(), lr=lrn_rate)
  max_epochs = 20
  log_every = 4

  print("\nStarting training ")
  for epoch in range(max_epochs):
    epoch_loss = 0.0
    for bix, batch in enumerate(train_ldr):
      X = batch[0]  # 64 input pixels
      Y = batch[0].type(T.float32)   # target same as input

      opt.zero_grad()
      oupt = net(X)
      loss_val = loss_func(oupt, Y)  # a tensor
      epoch_loss += loss_val.item()  # for progress display
      loss_val.backward()            # compute gradients
      opt.step()                     # update weights

    if epoch % log_every == 0:
      print("epoch = %4d   loss = %0.4f" % (epoch, epoch_loss))

  print("Done ")

  # 4. compute and store reconstruction errors
  print("\nComputing reconstruction errors ")
  net.eval()
  err_list = make_err_list(net, train_ds)
  err_list.sort(key=lambda x: x[1], \
    reverse=True)  # high error to low
  print("Done ")

  # 5. show most anomalous item(s)
  print("Items with largest reconstruction error: ")
  for i in range(5):
    (idx,err) = err_list[i]
    print(" [%4d]  %0.4f" % (idx, err)) 

  print("\nMost anomalous data item: ")
  idx = err_list[0][0]  # first item, index
  display_digit(train_ds, idx)

  print("\nEnd experiemnt ")

# -----------------------------------------------------------

if __name__ == "__main__":
  main()