feat: model_config 支持 summaries_set（PCOC / scalars）训练期 TensorBoard 监控

easy_rec/python/model/easy_rec_estimator.py

@@ -162,6 +162,7 @@ def _train_model_fn(self, features, labels, run_config):
         is_training=True)
     predict_dict = model.build_predict_graph()
     loss_dict = model.build_loss_graph()
+    model.build_summary_graph()
 
     regularization_losses = tf.get_collection(
         tf.GraphKeys.REGULARIZATION_LOSSES)

easy_rec/python/model/easy_rec_model.py

@@ -178,6 +178,9 @@ def build_loss_graph(self):
   def build_metric_graph(self, eval_config):
     return self._metric_dict
 
+  def build_summary_graph(self):
+    return
+
   @abstractmethod
   def get_outputs(self):
     pass

easy_rec/python/model/rank_model.py

@@ -509,6 +509,90 @@ def build_metric_graph(self, eval_config):
           ))
     return self._metric_dict
 
+  def _resolve_pred_tensor(self, pred_name):
+    if pred_name not in self._prediction_dict:
+      raise ValueError(
+          'summaries pred_name "%s" not found in prediction_dict keys: %s' %
+          (pred_name, sorted(self._prediction_dict.keys())))
+    pred = tf.to_float(self._prediction_dict[pred_name])
+    if pred_name == 'logits' or pred_name.startswith('logits_'):
+      pred = tf.sigmoid(pred)
+    # multi-class probs: take positive class when rank-2
+    if pred.shape.ndims is not None and pred.shape.ndims > 1:
+      if pred.shape.ndims == 2 and int(pred.shape[-1]) == 2:
+        pred = pred[:, 1]
+      else:
+        pred = tf.reshape(pred, [-1])
+    return pred
+
+  def _build_feature_mask(self, feature_name, feature_value):
+    if feature_name not in self._feature_dict:
+      raise ValueError(
+          'summaries feature_name "%s" not found in feature_dict keys: %s' %
+          (feature_name, sorted(self._feature_dict.keys())))
+    feat = self._feature_dict[feature_name]
+    if isinstance(feat, tf.SparseTensor):
+      dense = tf.sparse_to_dense(
+          feat.indices,
+          feat.dense_shape,
+          feat.values,
+          default_value='' if feat.values.dtype == tf.string else 0)
+      feat = tf.reshape(dense, [-1])
+    else:
+      feat = tf.reshape(feat, [-1])
+    if feat.dtype in (tf.float32, tf.float64, tf.int32, tf.int64):
+      try:
+        target = float(feature_value)
+      except (TypeError, ValueError):
+        target = None
+      if target is not None:
+        return tf.equal(tf.to_float(feat), tf.constant(target, dtype=tf.float32))
+    if feat.dtype != tf.string:
+      feat = tf.as_string(feat)
+    return tf.equal(feat, str(feature_value))
+
+  def _masked_mean(self, values, mask):
+    values = tf.reshape(tf.to_float(values), [-1])
+    masked = tf.boolean_mask(values, mask)
+    count = tf.size(masked)
+    return tf.cond(
+        count > 0,
+        lambda: tf.reduce_mean(masked),
+        lambda: tf.constant(0.0, dtype=tf.float32))
+
+  def _build_summary_impl(self, summary):
+    summary_type = summary.WhichOneof('summary')
+    if summary_type == 'pcoc':
+      if self._labels is None:
+        return
+      pred_name = summary.pcoc.pred_name or 'probs'
+      preds = self._resolve_pred_tensor(pred_name)
+      label = tf.to_float(self._labels[self._label_name])
+      label = tf.reshape(label, [-1])
+      predicted_ctr = tf.reduce_mean(preds)
+      observed_ctr = tf.reduce_mean(label)
+      epsilon = summary.pcoc.epsilon
+      pcoc = predicted_ctr / (observed_ctr + epsilon)
+      tf.summary.scalar('summary/predicted_ctr', predicted_ctr)
+      tf.summary.scalar('summary/observed_ctr', observed_ctr)
+      tf.summary.scalar('summary/pcoc', pcoc)
+    elif summary_type == 'scalars':
+      cfg = summary.scalars
+      pred_name = cfg.pred_name or 'probs'
+      preds = self._resolve_pred_tensor(pred_name)
+      if cfg.HasField('feature_name') and cfg.HasField('feature_value'):
+        mask = self._build_feature_mask(cfg.feature_name, cfg.feature_value)
+        value = self._masked_mean(preds, mask)
+      else:
+        value = tf.reduce_mean(preds)
+      tf.summary.scalar('summary/%s' % cfg.name, value)
+    elif summary_type is not None:
+      raise ValueError('unsupported summary type: %s' % summary_type)
+
+  def build_summary_graph(self):
+    for summary in self._base_model_config.summaries_set:
+      self._build_summary_impl(summary)
+
   def _get_outputs_impl(self, loss_type, num_class=1, suffix=''):
     binary_loss_set = {
         LossType.F1_REWEIGHTED_LOSS,

easy_rec/python/protos/easy_rec_model.proto

@@ -30,6 +30,7 @@ import "easy_rec/python/protos/pdn.proto";
 import "easy_rec/python/protos/dssm_senet.proto";
 import "easy_rec/python/protos/simi.proto";
 import "easy_rec/python/protos/dat.proto";
+import "easy_rec/python/protos/summary.proto";
 // for input performance test
 message DummyModel {
 }
@@ -158,4 +159,9 @@ message EasyRecModel {
 
     // label name for rank_model to select one label between multiple labels
     optional string label_name = 18;
+
+    // Training TensorBoard summaries, e.g.:
+    //   summaries_set { pcoc {} }
+    //   summaries_set { scalars { name: "c1_1005_pred" feature_name: "c1" feature_value: "1005" } }
+    repeated ModelSummaries summaries_set = 19;
 }

easy_rec/python/protos/summary.proto

@@ -0,0 +1,21 @@
+syntax = "proto2";
+package protos;
+
+message PCOC {
+  optional float epsilon = 1 [default = 1e-7];
+  optional string pred_name = 2 [default = "probs"];
+}
+
+message Scalars {
+  required string name = 1;
+  optional string pred_name = 2 [default = "probs"];
+  optional string feature_name = 3;
+  optional string feature_value = 4;
+}
+
+message ModelSummaries {
+  oneof summary {
+    PCOC pcoc = 1;
+    Scalars scalars = 2;
+  }
+}

easy_rec/python/test/eval_summary_test.py

@@ -0,0 +1,90 @@
+# -*- encoding:utf-8 -*-
+# Copyright (c) Alibaba, Inc. and its affiliates.
+from __future__ import division
+
+import logging
+import unittest
+
+from google.protobuf import text_format
+
+from easy_rec.python.protos.easy_rec_model_pb2 import EasyRecModel
+from easy_rec.python.protos.summary_pb2 import ModelSummaries
+from easy_rec.python.model.rank_model import RankModel
+
+if __name__ == '__main__':
+  import tensorflow as tf
+  if tf.__version__ >= '2.0':
+    tf = tf.compat.v1
+  tf.disable_eager_execution()
+
+
+def _tf_v1():
+  import tensorflow as tf
+  return tf.compat.v1 if tf.__version__ >= '2.0' else tf
+
+
+class _FakeRankModel(object):
+  _resolve_pred_tensor = RankModel._resolve_pred_tensor
+  _build_feature_mask = RankModel._build_feature_mask
+  _masked_mean = RankModel._masked_mean
+
+
+def _run_summary_tags(model, summary_text):
+  tf = _tf_v1()
+  tf.reset_default_graph()
+  model = model()
+  summary = ModelSummaries()
+  text_format.Parse(summary_text, summary)
+  RankModel._build_summary_impl(model, summary)
+  with tf.Session() as sess:
+    summary_str = sess.run(tf.summary.merge_all())
+  summary_proto = tf.Summary()
+  summary_proto.ParseFromString(summary_str)
+  return {v.tag: v.simple_value for v in summary_proto.value}
+
+
+class SummariesSetTest(unittest.TestCase):
+
+  def setUp(self):
+    logging.info('Testing %s.%s' % (type(self).__name__, self._testMethodName))
+
+  def test_proto_and_pcoc_graph(self):
+    model = EasyRecModel()
+    text_format.Parse(
+        'model_class: "DeepFM" deepfm {} summaries_set { pcoc {} }', model)
+    self.assertEqual(model.summaries_set[0].WhichOneof('summary'), 'pcoc')
+
+    class _M(_FakeRankModel):
+
+      def __init__(self):
+        tf = _tf_v1()
+        self._labels = {'label': tf.constant([1, 0, 1, 0], dtype=tf.float32)}
+        self._prediction_dict = {
+            'probs': tf.constant([0.8, 0.4, 0.6, 0.2], dtype=tf.float32)
+        }
+        self._feature_dict = {}
+        self._label_name = 'label'
+
+    values = _run_summary_tags(_M, 'pcoc { epsilon: 1e-7 }')
+    self.assertAlmostEqual(values['summary/pcoc'], 1.0, places=5)
+
+  def test_scalars_feature_slice(self):
+    class _M(_FakeRankModel):
+
+      def __init__(self):
+        tf = _tf_v1()
+        self._labels = {'label': tf.constant([1, 0], dtype=tf.float32)}
+        self._prediction_dict = {
+            'probs': tf.constant([0.9, 0.1], dtype=tf.float32)
+        }
+        self._feature_dict = {'c1': tf.constant([1005.0, 1002.0], dtype=tf.float32)}
+        self._label_name = 'label'
+
+    values = _run_summary_tags(
+        _M,
+        'scalars { name: "c1_1005_pred" feature_name: "c1" feature_value: "1005" }')
+    self.assertAlmostEqual(values['summary/c1_1005_pred'], 0.9, places=5)
+
+
+if __name__ == '__main__':
+  unittest.main()

samples/model_config/deepfm_combo_on_avazu_ctr.config

@@ -365,6 +365,24 @@ model_config:{
     l2_regularization: 1e-5
   }
   # embedding_regularization: 1e-7
+
+  # Training TensorBoard summaries (model_config.summaries_set, not eval_config)
+  # PCOC = mean(pred)/mean(label); tags: summary/pcoc, summary/predicted_ctr, summary/observed_ctr
+  summaries_set {
+    pcoc {
+      # pred_name: "probs"   # prediction_dict key; "logits" applies sigmoid first
+      # epsilon: 1e-7        # divisor stabilizer when observed ctr is near zero
+    }
+  }
+  # Feature-sliced scalar (mean pred on c1=="1005"); tag: summary/c1_1005_pred
+  summaries_set {
+    scalars {
+      name: "c1_1005_pred"
+      pred_name: "probs"
+      feature_name: "c1"
+      feature_value: "1005"
+    }
+  }
 }
 
 export_config {