neumf_model.py 18.1 KB
Newer Older
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
# Copyright 2018 The TensorFlow Authors. All Rights Reserved.
#
# 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
#
#     http://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.
# ==============================================================================
"""Defines NeuMF model for NCF framework.

Some abbreviations used in the code base:
NeuMF: Neural Matrix Factorization
NCF: Neural Collaborative Filtering
GMF: Generalized Matrix Factorization
MLP: Multi-Layer Perceptron

GMF applies a linear kernel to model the latent feature interactions, and MLP
uses a nonlinear kernel to learn the interaction function from data. NeuMF model
is a fused model of GMF and MLP to better model the complex user-item
interactions, and unifies the strengths of linearity of MF and non-linearity of
MLP for modeling the user-item latent structures.

In NeuMF model, it allows GMF and MLP to learn separate embeddings, and combine
the two models by concatenating their last hidden layer.
"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

36
37
import sys

38
39
from six.moves import xrange  # pylint: disable=redefined-builtin
import tensorflow as tf
40
from tensorflow.contrib import tpu as contrib_tpu
41

Shawn Wang's avatar
Shawn Wang committed
42
from official.recommendation import constants as rconst
43
from official.recommendation import movielens
Shining Sun's avatar
Shining Sun committed
44
from official.recommendation import ncf_common
Shawn Wang's avatar
Shawn Wang committed
45
from official.recommendation import stat_utils
46
from official.utils.logs import mlperf_helper
47
48


49
def sparse_to_dense_grads(grads_and_vars):
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
  """Convert sparse gradients to dense gradients.

  All sparse gradients, which are represented as instances of tf.IndexedSlices,
  are converted to dense Tensors. Dense gradients, which are represents as
  Tensors, are unchanged.

  The purpose of this conversion is that for small embeddings, which are used by
  this model, applying dense gradients with the AdamOptimizer is faster than
  applying sparse gradients.

  Args
    grads_and_vars: A list of (gradient, variable) tuples. Each gradient can
      be a Tensor or an IndexedSlices. Tensors are unchanged, and IndexedSlices
      are converted to dense Tensors.
  Returns:
    The same list of (gradient, variable) as `grads_and_vars`, except each
    IndexedSlices gradient is converted to a Tensor.
  """

  # Calling convert_to_tensor changes IndexedSlices into Tensors, and leaves
  # Tensors unchanged.
  return [(tf.convert_to_tensor(g), v) for g, v in grads_and_vars]


74
75
def neumf_model_fn(features, labels, mode, params):
  """Model Function for NeuMF estimator."""
76
77
78
  if params.get("use_seed"):
    tf.set_random_seed(stat_utils.random_int32())

79
  users = features[movielens.USER_COLUMN]
80
  items = features[movielens.ITEM_COLUMN]
81

Shining Sun's avatar
Shining Sun committed
82
83
84
  user_input = tf.keras.layers.Input(tensor=users)
  item_input = tf.keras.layers.Input(tensor=items)
  logits = construct_model(user_input, item_input, params).output
85

86
  # Softmax with the first column of zeros is equivalent to sigmoid.
Shining Sun's avatar
Shining Sun committed
87
  softmax_logits = ncf_common.convert_to_softmax_logits(logits)
88

89
  if mode == tf.estimator.ModeKeys.EVAL:
90
    duplicate_mask = tf.cast(features[rconst.DUPLICATE_MASK], tf.float32)
Shining Sun's avatar
Shining Sun committed
91
92
93
94
95
    return _get_estimator_spec_with_metrics(
        logits,
        softmax_logits,
        duplicate_mask,
        params["num_neg"],
Reed's avatar
Reed committed
96
        params["match_mlperf"],
97
        use_tpu_spec=params["use_xla_for_gpu"])
98

99
100
  elif mode == tf.estimator.ModeKeys.TRAIN:
    labels = tf.cast(labels, tf.int32)
101
    valid_pt_mask = features[rconst.VALID_POINT_MASK]
102
103
104
105
106
107
108
109
110
111
112

    mlperf_helper.ncf_print(key=mlperf_helper.TAGS.OPT_NAME, value="adam")
    mlperf_helper.ncf_print(key=mlperf_helper.TAGS.OPT_LR,
                            value=params["learning_rate"])
    mlperf_helper.ncf_print(key=mlperf_helper.TAGS.OPT_HP_ADAM_BETA1,
                            value=params["beta1"])
    mlperf_helper.ncf_print(key=mlperf_helper.TAGS.OPT_HP_ADAM_BETA2,
                            value=params["beta2"])
    mlperf_helper.ncf_print(key=mlperf_helper.TAGS.OPT_HP_ADAM_EPSILON,
                            value=params["epsilon"])

113
114
115
116
117
118
    optimizer = tf.compat.v1.train.AdamOptimizer(
        learning_rate=params["learning_rate"],
        beta1=params["beta1"],
        beta2=params["beta2"],
        epsilon=params["epsilon"])
    if params["use_tpu"]:
Shining Sun's avatar
Shining Sun committed
119
      # TODO(seemuch): remove this contrib import
120
      optimizer = contrib_tpu.CrossShardOptimizer(optimizer)
121

122
123
    mlperf_helper.ncf_print(key=mlperf_helper.TAGS.MODEL_HP_LOSS_FN,
                            value=mlperf_helper.TAGS.BCE)
Shining Sun's avatar
Shining Sun committed
124

125
    loss = tf.compat.v1.losses.sparse_softmax_cross_entropy(
126
        labels=labels,
127
128
        logits=softmax_logits,
        weights=tf.cast(valid_pt_mask, tf.float32)
129
130
    )

131
132
133
    # This tensor is used by logging hooks.
    tf.identity(loss, name="cross_entropy")

134
135
    global_step = tf.compat.v1.train.get_global_step()
    tvars = tf.compat.v1.trainable_variables()
136
137
    gradients = optimizer.compute_gradients(
        loss, tvars, colocate_gradients_with_ops=True)
138
    gradients = sparse_to_dense_grads(gradients)
139
140
    minimize_op = optimizer.apply_gradients(
        gradients, global_step=global_step, name="train")
141
    update_ops = tf.compat.v1.get_collection(tf.compat.v1.GraphKeys.UPDATE_OPS)
142
143
144
145
146
147
148
149
    train_op = tf.group(minimize_op, update_ops)

    return tf.estimator.EstimatorSpec(mode=mode, loss=loss, train_op=train_op)

  else:
    raise NotImplementedError


Shining Sun's avatar
Shining Sun committed
150
151
152
153
def _strip_first_and_last_dimension(x, batch_size):
  return tf.reshape(x[0, :], (batch_size,))


154
def construct_model(user_input, item_input, params):
155
  # type: (tf.Tensor, tf.Tensor, dict) -> tf.keras.Model
156
157
158
  """Initialize NeuMF model.

  Args:
Shining Sun's avatar
Shining Sun committed
159
160
    user_input: keras input layer for users
    item_input: keras input layer for items
161
162
163
    params: Dict of hyperparameters.
  Raises:
    ValueError: if the first model layer is not even.
164
  Returns:
165
    model:  a keras Model for computing the logits
166
167
168
169
170
171
172
173
174
175
176
  """
  num_users = params["num_users"]
  num_items = params["num_items"]

  model_layers = params["model_layers"]

  mf_regularization = params["mf_regularization"]
  mlp_reg_layers = params["mlp_reg_layers"]

  mf_dim = params["mf_dim"]

177
178
179
180
  mlperf_helper.ncf_print(key=mlperf_helper.TAGS.MODEL_HP_MF_DIM, value=mf_dim)
  mlperf_helper.ncf_print(key=mlperf_helper.TAGS.MODEL_HP_MLP_LAYER_SIZES,
                          value=model_layers)

181
182
183
  if model_layers[0] % 2 != 0:
    raise ValueError("The first layer size should be multiple of 2!")

184
185
186
  # Initializer for embedding layers
  embedding_initializer = "glorot_uniform"

187
188
189
  def mf_slice_fn(x):
    x = tf.squeeze(x, [1])
    return x[:, :mf_dim]
Shining Sun's avatar
Shining Sun committed
190

191
192
193
  def mlp_slice_fn(x):
    x = tf.squeeze(x, [1])
    return x[:, mf_dim:]
Shining Sun's avatar
Shining Sun committed
194

195
196
197
  # It turns out to be significantly more effecient to store the MF and MLP
  # embedding portions in the same table, and then slice as needed.
  embedding_user = tf.keras.layers.Embedding(
198
199
      num_users,
      mf_dim + model_layers[0] // 2,
200
201
      embeddings_initializer=embedding_initializer,
      embeddings_regularizer=tf.keras.regularizers.l2(mf_regularization),
202
203
204
      input_length=1,
      name="embedding_user")(
          user_input)
205
206

  embedding_item = tf.keras.layers.Embedding(
207
208
      num_items,
      mf_dim + model_layers[0] // 2,
209
210
      embeddings_initializer=embedding_initializer,
      embeddings_regularizer=tf.keras.regularizers.l2(mf_regularization),
211
212
213
      input_length=1,
      name="embedding_item")(
          item_input)
214
215
216
217
218
219
220
221
222
223
224
225

  # GMF part
  mf_user_latent = tf.keras.layers.Lambda(
      mf_slice_fn, name="embedding_user_mf")(embedding_user)
  mf_item_latent = tf.keras.layers.Lambda(
      mf_slice_fn, name="embedding_item_mf")(embedding_item)

  # MLP part
  mlp_user_latent = tf.keras.layers.Lambda(
      mlp_slice_fn, name="embedding_user_mlp")(embedding_user)
  mlp_item_latent = tf.keras.layers.Lambda(
      mlp_slice_fn, name="embedding_item_mlp")(embedding_item)
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249

  # Element-wise multiply
  mf_vector = tf.keras.layers.multiply([mf_user_latent, mf_item_latent])

  # Concatenation of two latent features
  mlp_vector = tf.keras.layers.concatenate([mlp_user_latent, mlp_item_latent])

  num_layer = len(model_layers)  # Number of layers in the MLP
  for layer in xrange(1, num_layer):
    model_layer = tf.keras.layers.Dense(
        model_layers[layer],
        kernel_regularizer=tf.keras.regularizers.l2(mlp_reg_layers[layer]),
        activation="relu")
    mlp_vector = model_layer(mlp_vector)

  # Concatenate GMF and MLP parts
  predict_vector = tf.keras.layers.concatenate([mf_vector, mlp_vector])

  # Final prediction layer
  logits = tf.keras.layers.Dense(
      1, activation=None, kernel_initializer="lecun_uniform",
      name=movielens.RATING_COLUMN)(predict_vector)

  # Print model topology.
250
251
  model = tf.keras.models.Model([user_input, item_input], logits)
  model.summary()
252
253
  sys.stdout.flush()

254
  return model
255
256


Shining Sun's avatar
Shining Sun committed
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def _get_estimator_spec_with_metrics(logits,              # type: tf.Tensor
                                     softmax_logits,      # type: tf.Tensor
                                     duplicate_mask,      # type: tf.Tensor
                                     num_training_neg,    # type: int
                                     match_mlperf=False,  # type: bool
                                     use_tpu_spec=False   # type: bool
                                    ):
  """Returns a EstimatorSpec that includes the metrics."""
  cross_entropy, \
  metric_fn, \
  in_top_k, \
  ndcg, \
  metric_weights = compute_eval_loss_and_metrics_helper(
      logits,
      softmax_logits,
      duplicate_mask,
      num_training_neg,
      match_mlperf,
      use_tpu_spec)

  if use_tpu_spec:
278
    return contrib_tpu.TPUEstimatorSpec(
Shining Sun's avatar
Shining Sun committed
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
        mode=tf.estimator.ModeKeys.EVAL,
        loss=cross_entropy,
        eval_metrics=(metric_fn, [in_top_k, ndcg, metric_weights]))

  return tf.estimator.EstimatorSpec(
      mode=tf.estimator.ModeKeys.EVAL,
      loss=cross_entropy,
      eval_metric_ops=metric_fn(in_top_k, ndcg, metric_weights)
  )


def compute_eval_loss_and_metrics_helper(logits,              # type: tf.Tensor
                                         softmax_logits,      # type: tf.Tensor
                                         duplicate_mask,      # type: tf.Tensor
                                         num_training_neg,    # type: int
                                         match_mlperf=False,  # type: bool
                                         use_tpu_spec=False   # type: bool
                                        ):
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
  """Model evaluation with HR and NDCG metrics.

  The evaluation protocol is to rank the test interacted item (truth items)
  among the randomly chosen 999 items that are not interacted by the user.
  The performance of the ranked list is judged by Hit Ratio (HR) and Normalized
  Discounted Cumulative Gain (NDCG).

  For evaluation, the ranked list is truncated at 10 for both metrics. As such,
  the HR intuitively measures whether the test item is present on the top-10
  list, and the NDCG accounts for the position of the hit by assigning higher
  scores to hits at top ranks. Both metrics are calculated for each test user,
  and the average scores are reported.

  If `match_mlperf` is True, then the HR and NDCG computations are done in a
  slightly unusual way to match the MLPerf reference implementation.
  Specifically, if the evaluation negatives contain duplicate items, it will be
  treated as if the item only appeared once. Effectively, for duplicate items in
  a row, the predicted score for all but one of the items will be set to
  -infinity

  For example, suppose we have that following inputs:
  logits_by_user:     [[ 2,  3,  3],
                       [ 5,  4,  4]]

  items_by_user:     [[10, 20, 20],
                      [30, 40, 40]]

  # Note: items_by_user is not explicitly present. Instead the relevant \
          information is contained within `duplicate_mask`

  top_k: 2

  Then with match_mlperf=True, the HR would be 2/2 = 1.0. With
  match_mlperf=False, the HR would be 1/2 = 0.5. This is because each user has
  predicted scores for only 2 unique items: 10 and 20 for the first user, and 30
  and 40 for the second. Therefore, with match_mlperf=True, it's guaranteed the
  first item's score is in the top 2. With match_mlperf=False, this function
  would compute the first user's first item is not in the top 2, because item 20
  has a higher score, and item 20 occurs twice.

  Args:
    logits: A tensor containing the predicted logits for each user. The shape
      of logits is (num_users_per_batch * (1 + NUM_EVAL_NEGATIVES),) Logits
340
      for a user are grouped, and the last element of the group is the true
341
342
343
344
345
346
347
348
349
350
351
352
      element.

    softmax_logits: The same tensor, but with zeros left-appended.

    duplicate_mask: A vector with the same shape as logits, with a value of 1
      if the item corresponding to the logit at that position has already
      appeared for that user.

    num_training_neg: The number of negatives per positive during training.

    match_mlperf: Use the MLPerf reference convention for computing rank.

Reed's avatar
Reed committed
353
354
355
    use_tpu_spec: Should a TPUEstimatorSpec be returned instead of an
      EstimatorSpec. Required for TPUs and if XLA is done on a GPU. Despite its
      name, TPUEstimatorSpecs work with GPUs
356
357

  Returns:
Shining Sun's avatar
Shining Sun committed
358
359
360
361
362
    cross_entropy: the loss
    metric_fn: the metrics function
    in_top_k: hit rate metric
    ndcg: ndcg metric
    metric_weights: metric weights
363
  """
364
365
  in_top_k, ndcg, metric_weights, logits_by_user = compute_top_k_and_ndcg(
      logits, duplicate_mask, match_mlperf)
366
367
368

  # Examples are provided by the eval Dataset in a structured format, so eval
  # labels can be reconstructed on the fly.
369
370
371
  eval_labels = tf.reshape(shape=(-1,), tensor=tf.one_hot(
      tf.zeros(shape=(logits_by_user.shape[0],), dtype=tf.int32) +
      rconst.NUM_EVAL_NEGATIVES, logits_by_user.shape[1], dtype=tf.int32))
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390

  eval_labels_float = tf.cast(eval_labels, tf.float32)

  # During evaluation, the ratio of negatives to positives is much higher
  # than during training. (Typically 999 to 1 vs. 4 to 1) By adjusting the
  # weights for the negative examples we compute a loss which is consistent with
  # the training data. (And provides apples-to-apples comparison)
  negative_scale_factor = num_training_neg / rconst.NUM_EVAL_NEGATIVES
  example_weights = (
      (eval_labels_float + (1 - eval_labels_float) * negative_scale_factor) *
      (1 + rconst.NUM_EVAL_NEGATIVES) / (1 + num_training_neg))

  # Tile metric weights back to logit dimensions
  expanded_metric_weights = tf.reshape(tf.tile(
      metric_weights[:, tf.newaxis], (1, rconst.NUM_EVAL_NEGATIVES + 1)), (-1,))

  # ignore padded examples
  example_weights *= tf.cast(expanded_metric_weights, tf.float32)

391
  cross_entropy = tf.compat.v1.losses.sparse_softmax_cross_entropy(
392
393
394
395
      logits=softmax_logits, labels=eval_labels, weights=example_weights)

  def metric_fn(top_k_tensor, ndcg_tensor, weight_tensor):
    return {
396
397
398
399
400
401
        rconst.HR_KEY: tf.compat.v1.metrics.mean(top_k_tensor,
                                                 weights=weight_tensor,
                                                 name=rconst.HR_METRIC_NAME),
        rconst.NDCG_KEY: tf.compat.v1.metrics.mean(ndcg_tensor,
                                                   weights=weight_tensor,
                                                   name=rconst.NDCG_METRIC_NAME)
402
403
    }

Shining Sun's avatar
Shining Sun committed
404
  return cross_entropy, metric_fn, in_top_k, ndcg, metric_weights
405
406
407
408
409


def compute_top_k_and_ndcg(logits,              # type: tf.Tensor
                           duplicate_mask,      # type: tf.Tensor
                           match_mlperf=False   # type: bool
Shawn Wang's avatar
Delint.  
Shawn Wang committed
410
                          ):
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
  """Compute inputs of metric calculation.

  Args:
    logits: A tensor containing the predicted logits for each user. The shape
      of logits is (num_users_per_batch * (1 + NUM_EVAL_NEGATIVES),) Logits
      for a user are grouped, and the first element of the group is the true
      element.
    duplicate_mask: A vector with the same shape as logits, with a value of 1
      if the item corresponding to the logit at that position has already
      appeared for that user.
    match_mlperf: Use the MLPerf reference convention for computing rank.

  Returns:
    is_top_k, ndcg and weights, all of which has size (num_users_in_batch,), and
    logits_by_user which has size
    (num_users_in_batch, (rconst.NUM_EVAL_NEGATIVES + 1)).
  """
  logits_by_user = tf.reshape(logits, (-1, rconst.NUM_EVAL_NEGATIVES + 1))
429
430
  duplicate_mask_by_user = tf.cast(
      tf.reshape(duplicate_mask, (-1, rconst.NUM_EVAL_NEGATIVES + 1)),
431
      logits_by_user.dtype)
432
433
434
435
436
437
438
439
440

  if match_mlperf:
    # Set duplicate logits to the min value for that dtype. The MLPerf
    # reference dedupes during evaluation.
    logits_by_user *= (1 - duplicate_mask_by_user)
    logits_by_user += duplicate_mask_by_user * logits_by_user.dtype.min

  # Determine the location of the first element in each row after the elements
  # are sorted.
441
  sort_indices = tf.argsort(
442
443
444
445
446
447
448
      logits_by_user, axis=1, direction="DESCENDING")

  # Use matrix multiplication to extract the position of the true item from the
  # tensor of sorted indices. This approach is chosen because both GPUs and TPUs
  # perform matrix multiplications very quickly. This is similar to np.argwhere.
  # However this is a special case because the target will only appear in
  # sort_indices once.
449
450
  one_hot_position = tf.cast(tf.equal(sort_indices, rconst.NUM_EVAL_NEGATIVES),
                             tf.int32)
451
452
453
454
455
  sparse_positions = tf.multiply(
      one_hot_position, tf.range(logits_by_user.shape[1])[tf.newaxis, :])
  position_vector = tf.reduce_sum(sparse_positions, axis=1)

  in_top_k = tf.cast(tf.less(position_vector, rconst.TOP_K), tf.float32)
456
457
  ndcg = tf.math.log(2.) / tf.math.log(
      tf.cast(position_vector, tf.float32) + 2)
458
459
460
461
462
463
464
  ndcg *= in_top_k

  # If a row is a padded row, all but the first element will be a duplicate.
  metric_weights = tf.not_equal(tf.reduce_sum(duplicate_mask_by_user, axis=1),
                                rconst.NUM_EVAL_NEGATIVES)

  return in_top_k, ndcg, metric_weights, logits_by_user