(self, X, learning_rate=0.1, epochs=1, batch_sz=100, show_fig=False)
| 23 | self.rng = RandomStreams() |
| 24 | |
| 25 | def fit(self, X, learning_rate=0.1, epochs=1, batch_sz=100, show_fig=False): |
| 26 | # cast to float32 |
| 27 | learning_rate = np.float32(learning_rate) |
| 28 | |
| 29 | |
| 30 | N, D = X.shape |
| 31 | n_batches = N // batch_sz |
| 32 | |
| 33 | W0 = init_weights((D, self.M)) |
| 34 | self.W = theano.shared(W0, 'W_%s' % self.id) |
| 35 | self.c = theano.shared(np.zeros(self.M), 'c_%s' % self.id) |
| 36 | self.b = theano.shared(np.zeros(D), 'b_%s' % self.id) |
| 37 | self.params = [self.W, self.c, self.b] |
| 38 | self.forward_params = [self.W, self.c] |
| 39 | |
| 40 | X_in = T.matrix('X_%s' % self.id) |
| 41 | |
| 42 | # attach it to the object so it can be used later |
| 43 | # must be sigmoidal because the output is also a sigmoid |
| 44 | H = T.nnet.sigmoid(X_in.dot(self.W) + self.c) |
| 45 | self.hidden_op = theano.function( |
| 46 | inputs=[X_in], |
| 47 | outputs=H, |
| 48 | ) |
| 49 | |
| 50 | # we won't use this cost to do any updates |
| 51 | # but we would like to see how this cost function changes |
| 52 | # as we do contrastive divergence |
| 53 | X_hat = self.forward_output(X_in) |
| 54 | cost = -(X_in * T.log(X_hat) + (1 - X_in) * T.log(1 - X_hat)).mean() |
| 55 | cost_op = theano.function( |
| 56 | inputs=[X_in], |
| 57 | outputs=cost, |
| 58 | ) |
| 59 | |
| 60 | # do one round of Gibbs sampling to obtain X_sample |
| 61 | H = self.sample_h_given_v(X_in) |
| 62 | X_sample = self.sample_v_given_h(H) |
| 63 | |
| 64 | # define the objective, updates, and train function |
| 65 | objective = T.mean(self.free_energy(X_in)) - T.mean(self.free_energy(X_sample)) |
| 66 | |
| 67 | # need to consider X_sample constant because you can't take the gradient of random numbers in Theano |
| 68 | updates = [(p, p - learning_rate*T.grad(objective, p, consider_constant=[X_sample])) for p in self.params] |
| 69 | train_op = theano.function( |
| 70 | inputs=[X_in], |
| 71 | updates=updates, |
| 72 | ) |
| 73 | |
| 74 | costs = [] |
| 75 | print("training rbm: %s" % self.id) |
| 76 | for i in range(epochs): |
| 77 | print("epoch:", i) |
| 78 | X = shuffle(X) |
| 79 | for j in range(n_batches): |
| 80 | batch = X[j*batch_sz:(j*batch_sz + batch_sz)] |
| 81 | train_op(batch) |
| 82 | the_cost = cost_op(X) # technically we could also get the cost for Xtest here |
no test coverage detected