Блог

Я изучаю RBM и скопировал и проанализировал код Хисмаэля Косты из его репозитория на github. Я изменил код, чтобы он работал только с tensorflow.v2. Каким-то образом при обучении функция потерь прыгает повсюду, а не падает.

Набор данных определенно сходится.

 length = 64*100

col0 = np.random.randint(0,2,length)
col1 = np.random.randint(0,2,length)
col2 = np.bitwise_or(col0, col1)
train_X = np.array([col0, col1, col2], dtype=np.float32).T
 

Сначала я подумал, что это неправильное применение GradientTape или неправильный синтаксис оптимизатора, так как эту часть я написал в основном сам, но, похоже, все в порядке.
Ни количество скрытых узлов, ни learnig_rate, ни количество шагов ничего не меняют.
Я также попробовал SGD optimizer, но безрезультатно.

Не мог бы кто-нибудь, пожалуйста, заглянуть в мой код:

 import tensorflow as tf
import numpy as np

class RBM(tf.keras.Model):
    
    def __init__(self, nv, nh, cd_steps=3):
        super().__init__()
        self.W = tf.Variable(tf.random.truncated_normal((nv, nh), dtype=tf.float64) * 0.01) 
        self.bv = tf.Variable(tf.zeros((nv, 1), dtype=tf.float64))
        self.bh = tf.Variable(tf.zeros((nh, 1), dtype=tf.float64))
        self.cd_steps = cd_steps 
        self.modelW = None 
            
        
    def bernoulli(self, p):
        return tf.nn.relu(tf.sign(p- tf.random.uniform(p.shape, dtype=tf.float64)))
        
    def sample_h(self, v):
        ph_given_v = tf.sigmoid(tf.matmul(v, self.W)   tf.squeeze(self.bh))
        return self.bernoulli(ph_given_v)
    
    def sample_v(self, h):
        pv_given_h = tf.sigmoid(tf.matmul(h, tf.transpose(self.W))   tf.squeeze(self.bv))
        return self.bernoulli(pv_given_h)
   
    def gibbs_step(self, i, k, vk):
        hk = self.sample_h(vk)
        vk = self.sample_v(hk)
        return i 1, k, vk
    
    def energy(self, v):
        b_term = tf.matmul(v, self.bv)
        linear_tranform = tf.matmul(v, self.W)   tf.squeeze(self.bh)
        h_term = tf.reduce_sum(tf.math.log(tf.exp(linear_tranform)   1), axis=1) 
        return tf.reduce_mean(-h_term -b_term)
    
    def loss(self, v, vk):
        return self.energy(v) - self.energy(vk) 
     
    
    def call(self, X, lr=0.01, batch_size=64, epochs=5):
        X = tf.data.Dataset.from_tensor_slices(X).batch(batch_size)
        optimizer = tf.keras.optimizers.Adam(learning_rate=lr)
        for epoch in range(epochs):
            losses = []
            for n_batch, batch in enumerate(X):
                print('batch: ', n_batch, end='r')
                with tf.GradientTape() as tape:
                    v = batch
                    vk = tf.identity(v)

                    i = tf.constant(0)
                    _, _, vk = tf.nest.map_structure(tf.stop_gradient,
                                                     tf.while_loop(cond = lambda i,k, *args: i<=k,
                                                                   body = self.gibbs_step,
                                                                   loop_vars =[i, tf.constant(0), v])
                                                    )

                    l = self.loss(v, vk)
                # grads = tape.gradient(l, [self.W, self.bv, self.bh])
                # optimizer.apply_gradients(zip(grads, self.trainable_variables))
                optimizer.minimize(l, var_list = [self.W, self.bv, self.bh], tape=tape) # this does the same job as the 2 lines abow (doesn't it?)
                

            losses.append(l)
            print('Epoch: {} Cost: {}'.format(epoch, np.mean(losses)))
        self.modelW = self.W


    def predict(self, Q, batch_size=64):
        Q = tf.data.Dataset.from_tensor_slices(Q).batch(batch_size)
        answers = []
        for n_batch, batch in enumerate(Q):
            print('batch: ', n_batch, end='r')
            v = batch
            i = tf.constant(0)
            _, _, vk = tf.nest.map_structure(tf.stop_gradient,
                                                tf.while_loop(cond = lambda i,k, *args: i<=k,
                                                              body = self.gibbs_step,
                                                              loop_vars =[i, tf.constant(0), v])
                                               )
        
            answers.append(vk.numpy())
        return np.concatenate(answers)
                    
        
 
 

просто выполните приведенный ниже код, чтобы запустить класс с набором данных.

 rbm = RBM(nv=train_X.shape[1], nh=5, cd_steps=3)
rbm(X=train_X, lr=0.001, epochs=10)
 

Хорошо, это просто я был идиотом:

1. Инициирование весов, вероятно, не было идеальным: кажется, что это работает лучше, когда они полностью случайны:

 self.W = tf.Variable(tf.random.truncated_normal((nv, nh), dtype=tf.float32)) 
self.bv = tf.Variable(tf.random.truncated_normal((nv, 1), dtype=tf.float32))
self.bh = tf.Variable(tf.random.truncated_normal((nh, 1), dtype=tf.float32))
 

2. вычисление потерь:
   Это лучше работает с квадратом:

 def loss(self, v, vk):
    return (self.energy(v) - self.energy(vk))**2 
 

3. неправильный отступ:
   строка «losses.append (l)» должна иметь такой же отступ, как и строка над ней:

     optimizer.minimize(l, var_list = [self.W, self.bv, self.bh], tape=tape)
    losses.append(l)
 

4. настройка learning_rate для оптимизатора
5. также очень помогло построение графика потерь. Некоторые тонкие изменения легче оценить, когда они видны на графике, чем в стеке чисел.

Итак, тема закрыта.