forked from pclubiitk/model-zoo
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmodel.py
198 lines (172 loc) · 9.44 KB
/
model.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
# -*- coding: utf-8 -*-
"""model.ipynb
Automatically generated by Colaboratory.
Original file is located at
https://colab.research.google.com/drive/1rlEs6_Z_XKP2BLdrHAS27MY986dzzEPS
"""
from __future__ import division
from collections import Counter, defaultdict
import os
from random import shuffle
import tensorflow.compat.v1 as tf
from utils import _context_windows, _window, _device_for_node, _batchify, _plot_with_labels
class GloVeModel():
def __init__(self, embedding_size, context_size, max_vocab_size=100000, min_occurrences=1,
scaling_factor=3/4, cooccurrence_cap=100, batch_size=512, learning_rate=0.05):
self.embedding_size = embedding_size
if isinstance(context_size, tuple):
self.left_context, self.right_context = context_size
elif isinstance(context_size, int):
self.left_context = self.right_context = context_size
else:
raise ValueError("`context_size` should be an int or a tuple of two ints")
self.max_vocab_size = max_vocab_size
self.min_occurrences = min_occurrences
self.scaling_factor = scaling_factor
self.cooccurrence_cap = cooccurrence_cap
self.batch_size = batch_size
self.learning_rate = learning_rate
self.__words = None
self.__word_to_id = None
self.__cooccurrence_matrix = None
self.__embeddings = None
def fit_to_corpus(self, corpus):
self.__fit_to_corpus(corpus, self.max_vocab_size, self.min_occurrences,
self.left_context, self.right_context)
self.__build_graph()
def __fit_to_corpus(self, corpus, vocab_size, min_occurrences, left_size, right_size):
word_counts = Counter()
cooccurrence_counts = defaultdict(float)
for region in corpus:
word_counts.update(region)
for l_context, word, r_context in _context_windows(region, left_size, right_size):
for i, context_word in enumerate(l_context[::-1]):
# add (1 / distance from focal word) for this pair
cooccurrence_counts[(word, context_word)] += 1 / (i + 1)
for i, context_word in enumerate(r_context):
cooccurrence_counts[(word, context_word)] += 1 / (i + 1)
self.__words = [word for word, count in word_counts.most_common(vocab_size)
if count >= min_occurrences]
self.__word_to_id = {word: i for i, word in enumerate(self.__words)}
self.__cooccurrence_matrix = {
(self.__word_to_id[words[0]], self.__word_to_id[words[1]]): count
for words, count in cooccurrence_counts.items()
if words[0] in self.__word_to_id and words[1] in self.__word_to_id}
def __build_graph(self):
self.__graph = tf.Graph()
with self.__graph.as_default(), self.__graph.device(_device_for_node):
count_max = tf.constant([self.cooccurrence_cap], dtype=tf.float32,
name='max_cooccurrence_count')
scaling_factor = tf.constant([self.scaling_factor], dtype=tf.float32,
name="scaling_factor")
self.__focal_input = tf.placeholder(tf.int32, shape=[self.batch_size],
name="focal_words")
self.__context_input = tf.placeholder(tf.int32, shape=[self.batch_size],
name="context_words")
self.__cooccurrence_count = tf.placeholder(tf.float32, shape=[self.batch_size],
name="cooccurrence_count")
focal_embeddings = tf.Variable(
tf.random_uniform([self.vocab_size, self.embedding_size], 1.0, -1.0),
name="focal_embeddings")
context_embeddings = tf.Variable(
tf.random_uniform([self.vocab_size, self.embedding_size], 1.0, -1.0),
name="context_embeddings")
focal_biases = tf.Variable(tf.random_uniform([self.vocab_size], 1.0, -1.0),
name='focal_biases')
context_biases = tf.Variable(tf.random_uniform([self.vocab_size], 1.0, -1.0),
name="context_biases")
focal_embedding = tf.nn.embedding_lookup([focal_embeddings], self.__focal_input)
context_embedding = tf.nn.embedding_lookup([context_embeddings], self.__context_input)
focal_bias = tf.nn.embedding_lookup([focal_biases], self.__focal_input)
context_bias = tf.nn.embedding_lookup([context_biases], self.__context_input)
weighting_factor = tf.minimum(
1.0,
tf.pow(
tf.div(self.__cooccurrence_count, count_max),
scaling_factor))
embedding_product = tf.reduce_sum(tf.multiply(focal_embedding, context_embedding), 1)
log_cooccurrences = tf.log(tf.to_float(self.__cooccurrence_count))
distance_expr = tf.square(tf.add_n([
embedding_product,
focal_bias,
context_bias,
tf.negative(log_cooccurrences)]))
single_losses = tf.multiply(weighting_factor, distance_expr)
self.__total_loss = tf.reduce_sum(single_losses)
tf.summary.scalar("GloVe_loss", self.__total_loss)
self.__optimizer = tf.train.AdagradOptimizer(self.learning_rate).minimize(
self.__total_loss)
self.__summary = tf.summary.merge_all()
self.__combined_embeddings = tf.add(focal_embeddings, context_embeddings,
name="combined_embeddings")
def train(self, num_epochs, log_dir=None, summary_batch_interval=1000,
tsne_epoch_interval=None):
should_write_summaries = log_dir is not None and summary_batch_interval
should_generate_tsne = log_dir is not None and tsne_epoch_interval
batches = self.__prepare_batches()
total_steps = 0
with tf.Session(graph=self.__graph) as session:
if should_write_summaries:
print("Writing TensorBoard summaries to {}".format(log_dir))
summary_writer = tf.summary.FileWriter(log_dir, graph=session.graph)
tf.global_variables_initializer().run()
for epoch in range(num_epochs):
shuffle(batches)
for batch_index, batch in enumerate(batches):
i_s, j_s, counts = batch
if len(counts) != self.batch_size:
continue
feed_dict = {
self.__focal_input: i_s,
self.__context_input: j_s,
self.__cooccurrence_count: counts}
session.run([self.__optimizer], feed_dict=feed_dict)
if should_write_summaries and (total_steps + 1) % summary_batch_interval == 0:
summary_str = session.run(self.__summary, feed_dict=feed_dict)
summary_writer.add_summary(summary_str, total_steps)
total_steps += 1
if should_generate_tsne and (epoch + 1) % tsne_epoch_interval == 0:
current_embeddings = self.__combined_embeddings.eval()
output_path = os.path.join(log_dir, "epoch{:03d}.png".format(epoch + 1))
self.generate_tsne(output_path, embeddings=current_embeddings)
self.__embeddings = self.__combined_embeddings.eval()
if should_write_summaries:
summary_writer.close()
def embedding_for(self, word_str_or_id):
if isinstance(word_str_or_id, str):
return self.embeddings[self.__word_to_id[word_str_or_id]]
elif isinstance(word_str_or_id, int):
return self.embeddings[word_str_or_id]
def __prepare_batches(self):
if self.__cooccurrence_matrix is None:
raise NotFitToCorpusError(
"Need to fit model to corpus before preparing training batches.")
cooccurrences = [(word_ids[0], word_ids[1], count)
for word_ids, count in self.__cooccurrence_matrix.items()]
i_indices, j_indices, counts = zip(*cooccurrences)
return list(_batchify(self.batch_size, i_indices, j_indices, counts))
@property
def vocab_size(self):
return len(self.__words)
@property
def words(self):
if self.__words is None:
raise NotFitToCorpusError("Need to fit model to corpus before accessing words.")
return self.__words
@property
def embeddings(self):
if self.__embeddings is None:
raise NotTrainedError("Need to train model before accessing embeddings")
return self.__embeddings
def id_for_word(self, word):
if self.__word_to_id is None:
raise NotFitToCorpusError("Need to fit model to corpus before looking up word ids.")
return self.__word_to_id[word]
def generate_tsne(self, path=None, size=(100, 100), word_count=1000, embeddings=None):
if embeddings is None:
embeddings = self.embeddings
from sklearn.manifold import TSNE
tsne = TSNE(perplexity=30, n_components=2, init='pca', n_iter=5000)
low_dim_embs = tsne.fit_transform(embeddings[:word_count, :])
labels = self.words[:word_count]
return _plot_with_labels(low_dim_embs, labels, path, size)