该文章为转载文章，作者简介：汪剑，现在在出门问问负责推荐与个性化。曾在微软雅虎工作，从事过搜索和推荐相关工作。

TensorFlow Wide And Deep 模型详解与应用（一）
TensorFlow Wide And Deep 模型详解与应用（二）

这个系列应该是介绍WDL实现细节比较好的两篇博客了，因此进行了转载，并在格式上进行了些许修正。

TensorFlow Wide And Deep 模型详解与应用（一）

模型介绍

Wide and deep模型是TensorFlow在2016年6月左右发布的一类用于分类和回归的模型，并应用到了Google Play的应用推荐中[1]。wide and deep模型的核心思想是结合线性模型的记忆能力（memorization）和DNN模型的泛化能力（generalization），在训练过程中同时优化2个模型的参数，从而达到整体模型的预测能力最优。结合我们的产品应用场景同google play的推荐场景存在较多的类似之处，在经过调研和评估后，我们也将wide and deep模型应用到产品的推荐排序模型，并搭建了一套线下训练和线上预估的系统。鉴于网上对wide and deep模型的相关描述和讲解并不是特别多，我们将这段时间对tensorflow1.1中该模型的调研和相关应用经验分享出来，希望对相关使用人士带来帮助。

wide and deep模型的框架在原论文的图中进行了很好的概述。wide端对应的是线性模型，输入特征可以是连续特征，也可以是稀疏的离散特征，离散特征之间进行交叉后可以构成更高维的离散特征。线性模型训练中通过L1正则化，能够很快收敛到有效的特征组合中。deep端对应的是DNN模型，每个特征对应一个低维的实数向量，我们称之为特征的embedding。DNN模型通过反向传播调整隐藏层的权重，并且更新特征的embedding。wide and deep整个模型的输出是线性模型输出与DNN模型输出的叠加。如原论文中提到的，模型训练采用的是联合训练（joint training），模型的训练误差会同时反馈到线性模型和DNN模型中进行参数更新。相比于ensemble learning中单个模型进行独立训练，模型的融合仅在最终做预测阶段进行，joint training中模型的融合是在训练阶段进行的，单个模型的权重更新会受到wide端和deep端对模型训练误差的共同影响。因此在模型的特征设计阶段，wide端模型和deep端模型只需要分别专注于擅长的方面，wide端模型通过离散特征的交叉组合进行memorization，deep端模型通过特征的embedding进行generalization，这样单个模型的大小和复杂度也能得到控制，而整体模型的性能仍能得到提高。

图1 Wide and deep模型示意图

模型定义

定义wide and deep模型是比较简单的，tutorial中提供了比较完整的模型构建实例：

获取输入

模型的输入是一个python的dataframe。如tutorial的实例代码，可以通过pandas.read_csv从CSV文件中读入数据构建data frame。

定义feature columns

tf.contrib.layers中提供了一系列的函数定义不同类型的feature columns：

• tf.contrib.layers.sparse_column_with_XXX构建低维离散特征

  sparse_feature_a = sparse_column_with_hash_bucket(…)

  sparse_feature_b = sparse_column_with_hash_bucket(…)

• tf.contrib.layers.crossed_column构建离散特征的组合

  sparse_feature_a_x_sparse_feature_b = crossed_column([sparse_feature_a, sparse_feature_b], …)

• tf.contrib.layers.real_valued_column构建连续型实数特征

  real_feature_a = real_valued_column(…)

• tf.contrib.layers.embedding_column构建embedding特征

  sparse_feature_a_emb = embedding_column(sparse_id_column=sparse_feature_a, )

定义模型

定义分类模型：

m = tf.contrib.learn.DNNLinearCombinedClassifier( n_classes = n_classes, // 分类数目 weight_column_name = weight_column_name, // 训练实例的权重 model_dir = model_dir, // 模型目录 linear_feature_columns = wide_columns, // 输入线性模型的feature columns linear_optimizer = tf.train.FtrlOptimizer( ...), // 线性模型权重更新的optimizer dnn_feature_columns = deep_columns, // 输入DNN模型的feature columns dnn_hidden_units=[ 100, 50]， // DNN模型的隐藏层单元数目 dnn_optimizer=tf.train.AdagradOptimizer( ...) // DNN模型权重更新的optimizer )

需要指出的是：模型的model_dir同下面会提到的export模型的目录是2个不同的目录，model_dir存放模型的graph和summary数据，如果model_dir存放了上一次训练的模型数据，训练时会从model_dir恢复上一次训练的模型并在此基础上进行训练。我们用tensorboard加载显示的模型数据也是从该目录下生成的。模型export的目录则主要是用于tensorflow server启动时加载模型的servable实例，用于线上预测服务。

如果要使用回归模型，可以如下定义：

m = tf.contrib.learn.DNNLinearCombinedRegressor( weight_column_name = weight_column_name, linear_feature_columns = wide_columns, linear_optimizer = tf.train.FtrlOptimizer( ...), dnn_feature_columns = deep_columns, dnn_hidden_units=[ 100, 50]， dnn_optimizer=tf.train.AdagradOptimizer( ...) ) 训练评测

训练模型可以使用fit函数：m.fit(input_fn=input_fn(df_train))，评测使用evaluate函数：m.evaluate(input_fn=input_fn(df_test))。Input_fn函数定义如何从输入的dataframe构建特征和标记:

def input_fn(df) // tf.constant构建constant tensor，df[k].values是对应feature column的值构成的listcontinuous_cols = {k: tf.constant(df[k].values) fork inCONTINUOUS_COLUMNS} // tf.SparseTensor构建sparse tensor，SparseTensor由indices,values, dense_shape三// 个dense tensor构成，indices中记录非零元素在sparse tensor的位置，values是// indices中每个位置的元素的值，dense_shape指定sparse tensor中每个维度的大小// 以下代码为每个category column构建一个[df[k].size，1]的二维的SparseTensor。categorical_cols = { k: tf.SparseTensor( indices=[[i, 0] fori inrange(df[k]. size)], values=df[k].values, dense_shape=[df[k]. size, 1]) fork inCATEGORICAL_COLUMNS } // 可以用以下示意图来表示以上代码构建的sparse tensor// label是一个 constanttensor，记录每个实例的 labellabel= tf. constant(df[LABEL_COLUMN].values) // features是continuous_cols和categorical_cols的union构成的dict // dict中每个entry的key是feature column的name，value是feature column值的tensor returnfeatures, label输出

模型通过export输出到一个指定目录，tensorflow serving从该目录加载模型提供在线预测服务：m.export(export_dir=export_dir,input_fn = export._default_input_fn

use_deprecated_input_fn=True,signature_fn=signature_fn)

input_fn函数定义生成模型servable实例的特征，signature_fn函数定义模型输入输出的signature。

由于在tensorflow1.0之后export已经deprecate，需要用export_savedmodel来替代，所以本文就不对export进行更多讲解，只在文末给出我们是如何使用它的，建议所有使用者以后切换到最新的API。

模型详解

wide and deep模型是基于TF.learn API来实现的，其源代码实现主要在tensorflow.contrib.learn.python.learn.estimators中。以分类模型为例，wide与deep结合的分类模型对应的类是DNNLinearCombinedClassifier，实现在源文件dnn_linear_combined.py。 我们先看看DNNLinearCombinedClassifier的初始化函数的完整定义，看构造一个wide and deep模型可以输入哪些参数：

def init( self, model_dir =None, n_classes =2, weight_column_name =None, linear_feature_columns =None, linear_optimizer =None, joint_linear_weights =False, dnn_feature_columns =None, dnn_optimizer =None, dnn_hidden_units =None, dnn_activation_fn =nn .relu, dnn_dropout =None, gradient_clip_norm =None, enable_centered_bias =False, config =None, feature_engineering_fn =None, embedding_lr_multipliers =None):

我们可以将类的构造函数中的参数分为以下几组

基础参数

• model_dir

  我们训练的模型存放到model_dir指定的目录中。如果我们需要用tensorboard来DEBUG模型，将tensorboard的logdir指向该目录即可：tensorboard –logdir=$model_dir

• n_classes

  分类数。默认是二分类，>2则进行多分类。

• weight_column_name

  定义每个训练样本的权重。训练时每个训练样本的训练误差乘以该样本的权重然后用于权重更新梯度的计算。如果需要为每个样本指定权重，input_fn返回的features里需要包含一个以weight_column_name为列名的列，该列的长度为训练样本的数目，列中每个元素对应一个样本的权重，数据类型是float，如以下伪代码：

weight = tf .constant(df[WEIGHT_COLUMN_NAME] .values, dtype=float32);features[weight_column_name] = weight

• config

  指定运行时配置参数

• eature_engineering_fn

  对输入函数input_fn输出的(features, label)进行后处理生成新的(features’, label’)然后输入给模型训练函数model_fn使用。

call_model_fn(): feature, labels = self._feature_engineering_fn(feature, labels) 线性模型相关参数

• linear_feature_columns

  线性模型的输入特征

• linear_optimizer

  线性模型的优化函数，定义权重的梯度更新算法，默认采用FTRL。所有默认支持的linear_optimizer和dnn_optimizer可以在optimizer.py的OPTIMIZER_CLS_NAMES变量中找到相关定义。

• join_linear_weights

  按照代码中的注释，如果join_linear_weights= true， 线性模型的权重会存放在一个tf.Variable中，可以加快训练，但是linear_feature_columns中的特征列必须都是sparse feature column并且每个feature column的combiner必须是“sum”。经过自己线下的对比试验，对模型的预测能力似乎没有太大影响，对训练速度有所提升，最终训练模型时我们保持了默认值。

DNN模型相关参数

• dnn_feature_columns

  DNN模型的输入特征

• dnn_optimizer

  DNN模型的优化函数，定义各层权重的梯度更新算法，默认采用Adagrad。

• dnn_hidden_units

  每个隐藏层的神经元数目

• dnn_activation_fn

  隐藏层的激活函数，默认采用RELU

• dnn_dropout

  模型训练中隐藏层单元的drop_out比例

• gradient_clip_norm

  定义gradient clipping，对梯度的变化范围做出限制，防止gradient vanishing 或gradient explosion。wide and deep中默认采用tf.clip_by_global_norm。

• embedding_lr_multipliers

  embedding_feature_column到float的一个mapping。对指定的embedding feature column在计算梯度时乘以一个常数因子，调整梯度的变化速率。

看完模型的构造函数后，我们大概知道wide和deep端的模型各对应什么样的模型，模型需要输入什么样的参数。为了更深入了解模型，以下我们对wide and deep模型的相关代码进行了分析，力求解决如下疑问： (1) 分别用于线性模型和DNN模型训练的特征是如何定义的，其内部如何实现；(2) 训练中线性模型和DNN模型如何进行联合训练，训练误差如何反馈给wide模型和deep模型？下面我们重点针对特征和模型训练这两方面进行解读。

特征

wide and deep模型训练一般是以多个训练样本作为1个批次(batch)进行训练，训练样本在行维度上定义，每一行对应一个训练样本实例，包括特征（feature column），标注（label）以及权重（weight），如图2。特征在列维度上定义，每个特征对应1个feature column，feature column由在列维度上的1个或者若干个张量(tensor)组成，tensor中的每个元素对应一个样本在该feature column上某个维度的值。feature column的定义在可以在源代码的feature_column.py文件中找到，对应类为_FeatureColumn，该类定义了基本接口，是wide and deep模型中所有特征类的抽象父类。

image.png

wide and deep模型中使用的特征包括两大类： 一类是连续型特征，主要用于deep模型的训练，包括real value类型的特征以及embedding类型的特征等；一类是离散型特征，主要用于wide模型的训练，包括sparse类型的特征以及cross类型的特征等。以下是所有特征的一个汇总图

image

图3 wide and deep模型特征类图

图中类与类的关系除了inherit（继承）之外，同时我们也标出了特征类之间的构成关系：_BucketizedColumn由_RealValueColumn通过对连续值域进行分桶构成，_CrossedColumn由若干_SparseColumn或者_BucketizedColumn或者_CrossedColumn经过交叉组合构成。图中左边部分特征属于离散型特征，右边部分特征属于连续型特征。

我们在实际使用的时候，通常情况下是调用tensorflow提供的接口来构建特征的。以下是构建各类特征的接口：

sparse_column_with_integerized_feature()--> _SparseColumnIntegerized sparse_column_with_hash_bucket()--> _SparseColumnHashed sparse_column_with_keys()--> _SparseColumnKeys sparse_column_with_vocabulary_file()--> _SparseColumnVocabulary weighted_sparse_column()--> _WeightedSparseColumn one_hot_column()--> _OneHotColumn embedding_column()--> _EmbeddingColumn shared_embedding_columns()--> List [_EmbeddingColumn]scattered_embedding_column()--> _ScatteredEmbeddingColumn real_valued_column()--> _RealValuedColumn bucketized_column()-->_BucketizedColumn crossed_column()--> _CrossedColumn

FeatureColumn为模型训练定义了几个基本接口用于提取和转换特征，在后面讲解具体feature时会有具体描述：

（1） def insert_transformed_feature(self, columns_to_tensors):

“”“Apply transformation and inserts it into columns_to_tensors.

FeatureColumn的特征输出和转换函数。columns_to_tensor是FeatureColumn到tensors的映射。

（2）def _to_dnn_input_layer(self, input_tensor, weight_collection=None, trainable=True, output_rank=2):

“”“Returns a Tensor as an input to the first layer of neural network.”“”

构建DNN的float tensor输入，参见后面对RealValuedColumn的讲解。

（3）def _deep_embedding_lookup_arguments(self, input_tensor):

“”“Returns arguments to embedding lookup to build an input layer.”“”

构建DNN的embedding输入，参见后面对EmbeddingColumn的讲解。

（4）def _wide_embedding_lookup_arguments(self, input_tensor):

“”“Returns arguments to look up embeddings for this column.”“”

构建线性模型的输入，参见后面对SparseColumn的讲解。

我们从离散型的特征（sparse特征）开始分析。离散型特征可以看做由若干键值构成的特征，比如用户的性别。在实际实现中，每一个键值在sparse column内部对应一个整数id。离散特征的基类是_SparseColumn：

class_SparseColumn(_FeatureColumn, collections.namedtuple("_SparseColumn", ["column_name", "is_integerized", "bucket_size", "lookup_config", "combiner", "dtype"])):

collections.namedtuple中的字符串数组是_SparseColumn从对应的创建接口函数中接收的输入参数的名称。

def__new__(cls, column_name, is_integerized=False, bucket_size=None, lookup_config=None, combiner="sum", dtype=dtypes.string):

SparseFeature是如何存放这些离散取值的呢？这个跟bucket_size和lookup_config这两个参数相关。在实际定义中，有且只定义其中一个参数。通过使用哪一个参数我们可以把sparse feature分成两类，定义lookup_config参数的特征使用一个in memory的字典存储feature的所有取值，包括后面会讲到的_SparseColumnKeys，_SparseColumnVocabulary；定义bucket_size参数的特征使用一个哈希表来存储特征值，特征值通过哈希函数散列到各个桶，包括_SparseColumnHashed和_SparseColumnIntegerized(is_integerized = True)。

dtype指定特征值的类型，除了字符串类型(dtypes.string）之外，spare feature column还支持64位整数类型（dtypes.int64），默认我们认为输入的离散特征是字符串，如果我们定义了is_integerized = True，那么我们认为特征是一个整型的id型特征，我们可以直接用特征的取值作为特征的id，而不需要建立一个专门的映射。

combiner参数对应的是样本维度特征的归一化，如果特征列在单个样本上有多个取值，combiner参数指定如何对单个样本上特征的多个取值进行归一化。源代码注释中是这样写的： “combiner： A string specifying how to reduce if the sparse column is multivalent”，multivalent的具体含义在crossed feature column的定义中有一个稍微清楚的解释（combiner: A string specifying how to reduce if there are multiple entries in a single row）。combiner可以指定3种归一化方式：sum对应无归一化，sqrtn对应L2归一化，mean对应L1归一化。通常情况下采用L2归一化，模型的准确度相对会更高。

SparseColumn不能直接作为DNN的输入，它只能用于直接构建线性模型的输入：

def_wide_embedding_lookup_arguments(self, input_tensor):return_LinearEmbeddingLookupArguments( input_tensor=self.id_tensor(input_tensor), weight_tensor=self.weight_tensor(input_tensor), vocab_size=self.length, initializer=init_ops.zeros_initializer(), combiner=self.combiner)

_LinearEmbeddingLookupArguments是一个namedtuple（A new subclass of tuple with named fields）。input_tensor是训练样本集中特征的id构成的数组，weight_tensor中每个元素对应一个样本中该特征的权重，vocab_size是特征取值的个数， intiializer是特征初始化的函数，默认初始化为0。

不过看源代码中_SparseColumn及其子类并没有使用特征权重：

defweight_tensor(self, input_tensor):"""Returns the weight tensor from the given transformed input_tensor."""returnNone

如果需要为_SparseColumn的特征赋予权重，可以使用_WeightedSparseColumn，构造接口函数为weighted_sparse_column（Create a _SparseColumn by combing sparse_id_column and weight_column）

class_WeightedSparseColumn(FeatureColumn, collections.namedtuple( "WeightedSparseColumn",["sparse_id_column", "weight_column_name", "dtype"])):def__new(cls, sparse_id_column, weight_column_name, dtype):returnsuper(_WeightedSparseColumn, cls).new(cls, sparse_id_column, weight_column_name, dtype)

_WeightedSparseColumn需要3个参数：sparse_id_column对应sparse feature column， 是_SparseColumn类型的对象，weight_column_name为输入中对应sparse_id_column的weight column（input_fn返回的features dict中需要有一个weight_column_name的tensor）dtype是weight column中每个元素的数据类型。这里有几个隐含要求：

（1）dtype需要能够转换成浮点数类型，否则会抛TypeError；

（2）weight_column_name对应的weight column可以是一个SparseTensor，也可以是一个常规的dense tensor，程序会将dense tensor转换成SparseTensor，但是要求weight column最终对应的SparseTensor与sparse_id_column的SparseTensor有相同的索引(indices)和维度(dense_shape)。

_WeightedSparseColumn输出特征的id tensor和weight tensor的函数如下：

definsert_transformed_feature(self, columns_to_tensors):"""Inserts a tuple with the id and weight tensors."""ifself.sparse_id_column notincolumns_to_tensors: self.sparse_id_column.insert_transformed_feature(columns_to_tensors) weight_tensor = columns_to_tensors[self.weight_column_name] ifnotisinstance(weight_tensor, sparse_tensor_py.SparseTensor): # The weight tensor can be a regular Tensor. In such case, sparsify it.// 我们输入的weight tensor可以是一个常规的Tensor， 如通过tf.Constants构建的tensor， // 这种情况下，会调用dense_to_sparse_tensor将weight_tensor转换成SparseTensor。 weight_tensor = contrib_sparse_ops.dense_to_sparse_tensor(weight_tensor) // 最终使用的weight_tensor的数据类型是float ifnotself.dtype.is_floating: weight_tensor = math_ops.to_float(weight_tensor) // 返回中对应该WeightedSparseColumn的一个二元组，二元组的第一个元素是SparseFeatureColumn调用 // insert_transformed_feature后的id_tensor，第二个元素是weight tensor。 columns_to_tensors[self] = tuple([columns_to_tensors[self.sparse_id_column],weight_tensor]) defid_tensor(self, input_tensor):"""Returns the id tensor from the given transformed input_tensor."""returninput_tensor[ 0] defweight_tensor(self, input_tensor):"""Returns the weight tensor from the given transformed input_tensor."""returninput_tensor[ 1] sparse column from keys

这个是最简单的离散特征，类比于枚举类型，一般用于枚举的值不是太多的情况。创建基于keys的sparse特征的接口是sparse_column_with_keys(column_name, keys, default_value=-1, combiner=None)，对应类是SparseColumnKeys，构造函数为：

def__new__(cls, column_name, keys, default_value=-1, combiner="sum"):returnsuper(_SparseColumnKeys, cls).new(cls, column_name, combiner=combiner, lookup_config=_SparseIdLookupConfig(keys=keys, vocab_size=len(keys), default_value=default_value), dtype=dtypes.string)

keys为一个字符串列表，定义了所有的枚举值。构造特征输入的keys最后存储在lookup_config里面，每个key的类型是string，并且对应1个id，id是该key在输入的keys数组中的下标。在模型实际训练中使用的是每个key对应的id。

SparseColumnKeys输入到模型前需要将枚举值的key转换到相应的id，这个转换工作在函数insert_transformed_feature中实现：

definsert_transformed_feature(self, columns_to_tensors):"""Handles sparse column to id conversion."""input_tensor = self._get_input_sparse_tensor(columns_to_tensors) """"Returns a lookup table that converts a string tensor into int64 IDs.This operation constructs a lookup table to convert tensor of strings into int64 IDs. The mapping can be initialized from a string mapping 1-D tensor where each element is a key and corresponding index within the tensor is the value. """table = lookup.index_table_from_tensor(mapping=tuple(self.lookup_config.keys), default_value=self.lookup_config.default_value, dtype=self.dtype, name= "lookup") columns_to_tensors[self] = table.lookup(input_tensor) sparse column from vocabulary file

sparse column with keys一般枚举都能满足，如果枚举的值多了就不合适了，所以提供了一个从文件加载枚举变量的接口：

sparse_column_with_vocabulary_file((column_name, vocabulary_file, num_oov_buckets =0, vocab_size =None, default_value =-1, combiner ="sum",dtype =dtypes .string)

对应的构造函数为：

def__new__(cls, column_name, vocabulary_file, num_oov_buckets=0, vocab_size=None, default_value=-1, combiner="sum", dtype=dtypes.string):

那么从文件中读入的特征值是存哪里呢？看看这个构造函数最后返回的类实例：

returnsuper(_SparseColumnVocabulary, cls)._new _(cls, column_name,combiner=combiner, lookup_config=_SparseIdLookupConfig(vocabulary_file=vocabulary_file,num_oov_buckets=num_oov_buckets, vocab_size=vocab_size,default_value=default_value), dtype=dtype)

如同_SparseColumnKeys，这个特征也使用了_SparseIdLookupConfig来存储特征值，vocabulary_file指向定义枚举值的文件，vocabulary_file每一行对应一个枚举值，每个枚举值的id是该枚举值所在行号（注意，行号是从0开始的），vocab_size定义枚举值的个数。_SparseIdLookupConfig从特征文件中构建一个特征值到id的哈希表，我们看看SparseColumnVocabulary进行vocabulary到id的转换时如何使用_SparseIdLookupConfig对象。

definsert_transformed_feature(self, columns_to_tensors):"""Handles sparse column to id conversion."""st = self._get_input_sparse_tensor(columns_to_tensors) ifself.dtype.is_integer: // 输入的整数数值型特征转换成字符串形式 sparse_string_values = string_ops.as_string(st.values) sparse_string_tensor = sparse_tensor_py.SparseTensor(st.indices,sparse_string_values, st.dense_shape) else: sparse_string_tensor = st """Returns a lookup table that converts a string tensor into int64 IDs.This operation constructs a lookup table to convert tensor of strings into int64 IDs. The mapping can be initialized from a vocabulary file specified in vocabulary_file, where the whole line is the key and the zero-based line number is the ID. table = lookup.index_table_from_file(vocabulary_file=self.lookup_config.vocabulary_file, num_oov_buckets=self.lookup_config.num_oov_buckets,vocab_size=self.lookup_config.vocab_size, default_value=self.lookup_config.default_value, name=self.name + "_lookup") columns_to_tensors[self] = table.lookup(sparse_string_tensor)

index_table_from_file函数从lookup_config的字典文件中构建table。Table变量是一个string到int64的HashTable，如果定义了num_oov_buckets，table是IdTableWithHashBuckets对象（a string to id wrapper that assigns out-of-vocabulary keys to buckets）。

sparse column with hash bucket

如果没有vocab文件定义枚举特征，我们可以使用hash bucket特征，使用该特征的接口是

sparse_column_with_hash_bucket(column_name, hash_bucket_size, combiner=None,dtype=dtypes.string)

对应类_SparseColumnHashed的构造函数为：def new(cls, column_name, hash_bucket_size, combiner=”sum”, dtype=dtypes.string):

ash_bucket_size定义哈希桶的个数，用于哈希值取模。dtype支持整数和字符串。实际计算哈希值的时候是将整数转换成对应的字符串表示形式，用字符串计算哈希值然后取模，转换后的特征值是0到hash_bucket_size的一个整数。

definsert_transformed_feature(self, columns_to_tensors):"""Handles sparse column to id conversion."""input_tensor = self._get_input_sparse_tensor(columns_to_tensors) ifself.dtype.is_integer: // 整数类型的输入转换成字符串类型 sparse_values = string_ops.as_string(input_tensor.values) else: sparse_values = input_tensor.values sparse_id_values = string_ops.string_to_hash_bucket_fast(sparse_values, self.bucket_size, name= "lookup") // Sparse特征的哈希值作为特征值对应的id返回 columns_to_tensors[self] = sparse_tensor_py.SparseTensor(input_tensor.indices, sparse_id_values, input_tensor.dense_shape) integerized sparse column

hash bucket的sparse特征取哈希值的时候是将整数看做字符串处理的，如果我们希望用整数本身的数值作为哈希值，可以使用_SparseColumnIntegerized，对应的接口是

sparse_column_with_integerized_feature： defsparse_column_with_integerized_feature(column_name,hash_bucket_size,combiner= "sum", dtype=dtypes.int64) 对应的类是SparseColumnIntegerized： def__new_(cls, column_name, bucket_size, combiner= "sum", dtype=dtypes.int64) 特征的转换函数定义： definsert_transformed_feature(self, columns_to_tensors): """Handles sparse column to id conversion."""input_tensor = self._get_input_sparse_tensor(columns_to_tensors) // 直接对特征值取模，取模后的值作为特征值的idsparse_id_values = math_ops.mod(input_tensor.values, self.bucket_size, name= "mod") columns_to_tensors[self] = sparse_tensor_py.SparseTensor( input_tensor.indices, sparse_id_values, input_tensor.dense_shape) crossed column

Crossed column支持1个以上的离散型feature column进行笛卡尔积，组成高维度的交叉特征。特征之间进行交叉，可以将特征之间的相关性引入模型，增强模型的表达能力。crossed column仅支持以下3种离散特征的交叉组合： _SparsedColumn, _BucketizedColumn和_CrossedColumn，其接口定义为：

def crossed_column(columns, hash_bucket_size, combiner=”sum”,ckpt_to_load_from=None, tensor_name_ in_ckpt=None, hash_key=None) 对应类为CrossedColumn： def new(cls, columns, hash_bucket_size, hash_key, combiner= "sum",ckpt_to_load_from=None, tensor_name in_ckpt=None):

columns对应一个feature column的集合，如tutorial中的例子：[age_buckets, education, occupation]；hash_bucket_size参数指定hash bucket的桶个数，特征交叉的组合个数越多，hash_bucket_size也应相应增加，从而减小哈希冲突。

交叉特征生成模型输入的逻辑可以分为如下两步：

definsert_transformed_feature(self, columns_to_tensors):"""Handles cross transformation."""def_collect_leaf_level_columns(cross):"""Collects base columns contained in the cross."""leaf_level_columns = [] forc incross.columns: // 对CrossedColumn类型的feature column进行递归展开 ifisinstance(c, _CrossedColumn): leaf_level_columns.extend(_collect_leaf_level_columns(c)) else: // SparseColumn和BucketizedColumn作为叶子节点 leaf_level_columns.append(c) returnleaf_level_columns // 步骤 1： 将crossed column中的所有特征进行递归展开，展开后的特征值存放在feature_tensors数组中 feature_tensors = [] forc in_collect_leaf_level_columns(self): ifisinstance(c, _SparseColumn): feature_tensors.append(columns_to_tensors[c.name]) else: ifc notincolumns_to_tensors: c.insert_transformed_feature(columns_to_tensors) ifisinstance(c, _BucketizedColumn): feature_tensors.append(c.to_sparse_tensor(columns_to_tensors[c])) else: feature_tensors.append(columns_to_tensors[c]) // 步骤 2: 生成cross feature的tensor，sparse_feature_cross通过动态库调用SparseFeatureCross函数，函数接 //口可参见sparse_feature_cross_op.cc columns_to_tensors[self] = sparse_feature_cross_op.sparse_feature_cross(feature_tensors, hashed_output= True,num_buckets=self.hash_bucket_size,hash_key=self.hash_key, name= "cross")

在源代码该部分的注释中有一个例子说明feature column进行cross后的效果，我们用1个图来将这部分注释展示的更明确点：

image

需要指出的一点是：交叉特征是没有权重定义的。

对离散特征进行交叉组合在预测模型中使用比较广泛，但是该类特征的一个局限性是它对训练数据中没有见过的特征组合泛化能力有限，后面我们谈到的embedding column则是通过构建离散特征的低维向量表示，强化离散特征的泛化能力。

real valued column

real valued feature column对应连续型数值特征，接口为

real_valued_column(column_name, dimension= 1, default_value= None, dtype=dtypes.float32,normalizer= None):

对应类为_RealValuedColumn：

_ RealValuedColumn(column_name, dimension, default_value, dtype,normalizer)

dimension指定feature column的维度，默认值为1，即1维浮点数数组。dimension也可以取大于1的整数，对应多维数组。rea valued column的特征取值类型可以是float32或者int，int类型在输入到模型之前会转换成float类型。normalizer定义在一批训练样本实例中，特征在列维度的归一化，相当于column-level normalization。这个同sparse feature column的combiner不同，combiner定义的是离散特征在单个样本维度的归一化（example-level normalization），以下示意图举了个例子来说明两者的区别：

image

图5 combiner与normalizer的区别

normalizer在real valued feature column输入DNN时调用：

definsert_transformed_feature(self, columns_to_tensors):# Transform the input tensor according to the normalizer function.// _normalized_input_tensor调用的是构造real valued colum时传入的normalizer函数 input_tensor = self._normalized_input_tensor(columns_to_tensors[self.name]) columns_to_tensors[self] = math_ops.to_float(input_tensor)

real valued column调用_to_dnn_input_layer转换为DNN的输入。_to_dnn_input_layer生成一个二维数组，数组的每一行是一个训练样本的real valued column的特征值，该特征值与其他连续型特征拼接后构成DNN的输入层。

def_to_dnn_input_layer(self,input_tensor,weight_collections=None,trainable=True,output_rank=2):// DNN的输入必须是dense tensor，sparse tensor需要调用to_dense_tensor转换成dense tensor input_tensor = self._to_dense_tensor(input_tensor) ifinput_tensor.dtype != dtypes.float32: input_tensor = math_ops.to_float(input_tensor) // 调用dense_inner_flatten(input_tensor, output_rank)。 // output_rank = 2，输出 [batch_size, real value column’s input dimension] return_reshape_real_valued_tensor(input_tensor, output_rank, self.name) def_to_dense_tensor(self, input_tensor):ifisinstance(input_tensor, sparse_tensor_py.SparseTensor): default_value = (self.default_value[ 0] ifself.default_value isnotNoneelse0) // Sparse tensor转换成dense tensor returnsparse_ops.sparse_tensor_to_dense(input_tensor, default_value=default_value) // real valued column直接返回input tensor returninput_tensor bucketized column

连续型特征通过bucketization生成离散特征，连续特征离散化的优点在网上有一些相关讨论，比如餐馆的距离对用户选择的影响，我们通常会将距离划分为若干个区间，如100米以内，1公里以内等，这样小幅度的距离差异不会对我们最终模型的预测造成太大影响，除非距离差异跨域了区间边界。bucketized column的接口定义为：def bucketized_column(source_column, boundaries)对应类为_BucketizedColumn，构造函数定义：def new(cls, source_column, boundaries):source_column必须是real_valued_column，boundaries是一个浮点数的列表，而且列表必须是递增序的，比如boundaries = [0, 100, 200] 定义了以下一组区间： （-INF， 0），[0，100），[100，200），[200, INF)。

definsert_transformed_feature(self, columns_to_tensors):# Bucketize the source column.ifself.source_column notincolumns_to_tensors: self.source_column.insert_transformed_feature(columns_to_tensors) columns_to_tensors[self] = bucketization_op.bucketize(columns_to_tensors[self.source_column], boundaries=list(self.boundaries), name= "bucketize")

bucketize函数调用tensorflow c++ core library中的BucketizeOp类完成feature的bucketization功能。

embedding column

sparse feature column通过embedding转换成连续型向量后可以作为deep model的输入，前面谈到了cross column的一个不足之处是在测试集合的泛化能力，通过embedding column将离散特征连续化，根据标注学习特征的向量形式，如同矩阵分解中学习物品的隐含因子向量或者词向量模型中单词的词向量。embedding column的接口形式是：

def embedding_column(sparse_id_column, dimension, combiner= None, initializer= None, ckpt_to_load_from= None,tensor_name_in_ckpt= None, max_norm= None, trainable= True) 对应类为_EmbeddingColumn： def new(cls,sparse_id_column,dimension,combiner= "mean",initializer= None, ckpt_to_load_from= None, tensor_name_in_ckpt= None,shared_embedding_name= None, shared_vocab_size= None,max_norm= None, trainable = True):

sparse_id_column是SparseColumn对象或者WeightedSparseColumn对象，dimension是embedding column的向量维度。SparseColumn的每个特征取值对应一个整数id，该整数id在embedding column中对应一个dimension维度的浮点数向量。combiner参数指定在单个样本上对特征向量归一化的方式，initializer参数指定特征向量的初始化函数，默认按truncated normal distribution初始化(mean = 0, stddev = 1/ sqrt(length of sparse id column))。max_norm限定每个样本特征向量做L2归一化后的最大值：embedding_vector = embedding_vector * max_norm / L2_norm(embedding_vector)。

为了进一步理解embedding column，我们可以画一个简易图：

image

图6 embedding feature column示意图

如上图，以sparse_column_with_keys(column_name = ‘gender’, keys = [‘female’, ‘male’])为例，假设female对应id = 0, male对应id = 1，每个id在embedding feature中对应1个6维的浮点数向量。 在实际训练数据中，当gender特征取值为’female’时，给到DNN输入层的将是id = 0对应的向量（tf.embedding_lookup_sparse）。 embedding_column设置了一个trainable参数，指定是否根据模型训练误差更新特征对应的embedding。

embedding特征的变换函数：

definsert_transformed_feature(self, columns_to_tensors):ifself.sparse_id_column notincolumns_to_tensors: self.sparse_id_column.insert_transformed_feature(columns_to_tensors) columns_to_tensors[self] = columns_to_tensors[self.sparse_id_column] def_deep_embedding_lookup_arguments(self, input_tensor):return_DeepEmbeddingLookupArguments( input_tensor=self.sparse_id_column.id_tensor(input_tensor), // sparse_id_column为_SparseColumn类型的对象时，weight_tensor = None// sparse_id_column为_WeightedSparseColumn类型对象时，weight_tensor = WeihgtedSparseColumn的 // weight tensor，weight_tensor须满足： // 1）weight_tensor.indices = input_tensor.indices // 2）weight_tensor.shape = input_tensor.shape weight_tensor=self.sparse_id_column.weight_tensor(input_tensor), // sparse feature column的元素个数 vocab_size=self.length, // embedding的维度 dimension=self.dimension, // embedding的初始化函数 initializer=self.initializer, // embedding的行归一化方法 combiner=self.combiner, shared_embedding_name=self.shared_embedding_name, hash_key= None, max_norm=self.max_norm, trainable=self.trainable)

从_DeepEmbeddingLookupArguments产生sparse feature的embedding的逻辑在函数_embeddings_from_arguments实现:

def_embeddings_from_arguments(column, args, weight_collections,trainable, output_rank=2):// column对应embedding feature column的name， args是feature column对应的 // _DeepEmbeddingLookupArguments对象，weight_collections存储embedding的权重， // output_rank指定输出embedding的tensor的rank。 input_tensor = layers._inner_flatten(args.input_tensor, output_rank) weight_tensor = layers._inner_flatten(args.weight_tensor, output_rank) // 考虑默认情况下构建embedding: args.hash_key isNone, args.shared_embedding_name isNone// 获取或创建embedding的model variable // embeddings是[number of sparse feature id, embedding dimension]的浮点数二维数组 // 每行对应一个sparse feature id的embedding embeddings = contrib_variables.model_variable( name= 'weights'，shape=[args.vocab_size, args.dimension], dtype=dtypes.float32,initializer=args.initializer, // If trainable, embedding vector作为一个model variable添加到GraphKeys.TRAINABLE_VARIABLES trainable=(trainable andargs.trainable), collections=weight_collections // weight_collections存储每个feature id的weight ) // 获取每个sparse feature id的embedding returnembedding_ops.safe_embedding_lookup_sparse(embeddings, input_tensor, sparse_weights=weight_tensor, combiner=args.combiner, name=column.name + 'weights', max_norm=args.max_norm)

safe_embedding_lookup_sparse调用tf.embedding_lookup_sparse获取每个sparse feature id的embedding。

tf.embedding_lookup_sparse首先调用tf.embedding_lookup获取sparse feature id的embedding vector:

// sp_ids是input_tensor的id tensorids = sp_ids.values embeddings = embedding_lookup ( // params对应embeddings矩阵，每个元素是embedding_dimension的float tensor，可以将params看// 做一个embedding tensor的partitions，partition的策略由partition_strategy指定params, // ids对应input_tensor的values数组ids, // id分配到params的分配策略，有mod和div两种，默认mod，具体定义可参见tf.embedding_lookup的说明partition_strategy=partition_strategy, // 限制embedding的最大L2-Normmax_norm=max_norm )

如果sparse_weights不是None，embedding的值乘以weights，

weights = sparse_weights.values

embeddings *= weights

根据combiner，对embedding进行归一化

segment_id = sp_ids .indices[ ;0]if combiner == "sum": // No normalization embeddings = math_ops .segment_sum(embeddings, segment_ids, name=name) elif combiner == "mean": // L1 normlization: embeddings = SUM(embeddings * weight) / SUM(weight) embeddings = math_ops .segment_sum(embeddings, segment_ids) weight_sum = math_ops .segment_sum(weights, segment_ids) embeddings = math_ops .div(embeddings, weight_sum, name=name) elif combiner == "sqrtn": // L2 normalization: embeddings = SUM(embeddings * weight^ 2) / SQRT(SUM(weight^ 2)) embeddings = math_ops .segment_sum(embeddings, segment_ids) weights_squared = math_ops .pow(weights, 2) weight_sum = math_ops .segment_sum(weights_squared, segment_ids) weight_sum_sqrt = math_ops .sqrt(weight_sum) embeddings = math_ops .div(embeddings, weight_sum_sqrt, name=name)其他feature columns

除了以上列举的几个feature column，tensorflow还支持one hot column，shared embedding column和scattered embedding column。one hot column对sparse feature column进行one-hot编码，如果离散特征的取值较少，可以用one hot feature column进行编码用于DNN的训练。不同于embedding column，one hot feature column不支持通过模型训练来更新其特征的embedding。shared embedding column和scattered embedding column由于篇幅原因就不多谈了。

前面讲了模型输入的特征，下面谈谈模型本身。关于wide and deep模型官方教程中有一段描述：The wide models and deep models are combined by summing up their final output log odds as the prediction, then feeding the prediction to a logistic loss function。从这里大概看出，线性模型与DNN模型进行结合的方式是对两者的预测结果进行相加，然后把相加过后的值拿去计算分类的loss。下面我们在源代码级别把上面这段描述展开，详细分析wide端模型和deep端模型是如何实现结合的。

由于我们运用的只是分类模型，所以就不对回归模型进行分析。 DNNLinearCombinedClassifier类继承于类Estimator，Estimator类继承于类BaseEstimator。BaseEstimator是一个抽象类，定义了通用的模型训练以及评测的函数接口(train_model, evaluate_model, infer_model)，Estimator类中用一个统一函数call_model_fn来实现train_model, evaluate_model, infer_model。

image

图7 estimator的类关系图

为了更好了解整个过程，我们看看内部函数的调用过程（代码可以参见estimator/estimator.py）：

image

图8 Estmiator类的函数调用图

模型训练通过调用BaseEstimator的fit()接口开始，其调用栈是：fit -> _train_model -> _get_train_ops ->_call_model_fn(ModelKeys.TRAIN) -> _model_fn，最终_model_fn()产生模型并通过export函数将模型输出到model_dir对应目录中。

我们把训练模型的调用过程在代码级别展开，标出关键的几个函数和数据结构，省略不关键的代码，希望能让读者看到训练模型的大致过程：

image

图9 模型训练的调用栈

评测（evaluate）和预测（predict）的过程与训练（train）大致相同，读者可以通过源代码文件找到对应函数了解。可以看出，整个函数调用栈中最关键的2个函数是： input_fn和model_fn。input_fn从输入数据中生成features和labels，features是一个Tensor或者是一个从特征名到Tensor的字典，如果features是一个Tensor，程序会给这个Tensor一个空字符串的键值，转换成特征名到Tensor的字典。labels是样本的label构成的tensor。input_fn由应用程序调用者提供实现，返回（features, labels）二元组，要求tf.get_shape(features)[0] == tf.get_shape(labels)[0]，也就是两个tensor的行数目得保持一致。model_fn定义训练和评测模型的具体逻辑，如模型训练产生的误差(model_fn_ops.loss)以及训练算子（model_fn_ops.train_op）通过封装在EstmiatorSpec的对象中由training的Session进行调用。每个具体模型需要实现的是自定义的model_fn。

DNNLinearCombinedClassifier是如何实现自己的model_fn的呢？本文开头我们给出了它的初始化函数原型，进入初始化函数的实现中我们定位到代码行 model_fn=_dnn_linear_combined_model_fn。

这个就是DNNLinearCombinedClassifier的model_fn。这个函数的定义如下：

def_dnn_linear_combined_model_fn(features, labels, mode, params, config= None)

features和labels大家都已经知道，mode指定model_fn的操作模式，目前支持3个值：训练模型(model_fn.ModeKeys.TRAIN)，对模型进行评测(model_fn.ModeKeys.EVAL)，根据输入特征进行预测(model_fn.ModeKeys.PREDICT)，mode的定义可参见文件estimator/model_fn.py。params和config参数分别定义模型训练的参数以及模型运行的配置。

模型训练的整个过程可以分为如下三步：

（1）从params获取参数，定义模型优化函数

// 将输入的特征统一为字符串到tensor的dictfeatures = _get_feature_dict(features) // 定义线性模型的优化函数linear_optimizer = params. get( "linear_optimizer") or"Ftrl"linear_optimizer = _get_optimizer(linear_optimizer) // 定义DNN模型的优化函数dnn_optimizer = params. get( "dnn_optimizer") or"Adagrad"dnn_optimizer = _get_optimizer(dnn_optimizer)

（2）构建DNN模型

首先从dnn_feature_columns构造DNN模型输入层：

dnn_feature_columns = params. get( "dnn_feature_columns") // dnn_feature_columns中包括DNN模型需要的所有连续特征和embedding特征net = layers.input_from_feature_columns(columns_to_tensors=features, feature_columns=dnn_feature_columns, weight_collections=[dnn_parent_scope], scope=dnn_input_scope)

这里的features是feature column key到tensor的dict，在本文前面我们对每个特征都谈了它的insert_transformed_feature(columns_to_tensors)，在构建DNN模型输入层之前会调用每个FeatureColumn的insert_transformed_feature函数生成一个浮点数类型的tensor作为DNN输入层的一部分。

构建完输入层后，然后至底向上进行隐藏层构建，每层的隐藏单元的个数在参数

dnn_hidden_units中定义： dnn_hidden_units = params. get( "dnn_hidden_units") forlayer_id, num_hidden_units inenumerate(dnn_hidden_units): // 从输入层或者是下面一层隐藏层构建新的隐藏层// dnn_activation_fn定义隐藏层的激活函数，默认使用RELUdnn_activation_fn = params. get( "dnn_activation_fn") ornn.relu。 net = layers.fully_connected(net，num_hidden_units，activation_fn=dnn_activation_fn， variables_collections=[dnn_parent_scope], scope=dnn_hidden_layer_scope) // 模型训练中对隐藏层进行drop-out// dnn_dropout是舍弃隐藏单元输出的概率dnn_dropout = params. get( "dnn_dropout") ifdnn_dropout is notNone andmode == model_fn.ModeKeys.TRAIN: net = layers.dropout(net, keep_prob=( 1.0- dnn_dropout))

从最后一层隐藏层到输出层建立一个全连接就完成DNN模型的构建了, dnn_logits是模型输出，它是一个分类数个数的Tensor，Tensor的每个元素对应1个分类的线性激活值（linear activation: w * net + bias）：

withvariable_scope.variable_scope( "logits", values=(net,)) asdnn_logits_scope: dnn_logits = layers.fully_connected( net, head.logits_dimension, // 分类个数activation_fn=None, // 输出层不做非线性变换variables_collections=[dnn_parent_scope], scope=dnn_logits_scope)

从dnn feature columns中构建DNN输入层的工作主要由函数layers._input_from_feature_columns完成，每个训练样本上的dnn feature column的值拼接成一个dense tensor作为DNN的输入。

该函数的定义如下：

def _input_from_feature_columns( columns_to_tensors, feature_columns, weight_collections, trainable, scope, output_rank, default_name) transformer = _Transformer(columns_to_tensors) forcolumn insorted( set(feature_columns), key=lambda x: x.key): // 调用feature column的特征变换函数insert_transformed_feature()// 返回feature column变换后的tensor： columns_to_tensors[column]。transformed_tensor = transformer.transform(column) try: // 构建_EmbeddingColumn的embeddingarguments = column._deep_embedding_lookup_arguments(transformed_tensor) output_tensors.append(_embeddings_from_arguments(column, arguments, weight_collections, trainable, output_rank=output_rank)) exceptNotImplementedError asee: // 构建_RealValuedColumn的tensortry: output_tensors.append(column._to_dnn_input_layer(transformed_tensor,weight_collections, trainable,output_rank=output_rank)) exceptValueError ase: // output_tensors数组每个元素对应一个feature column在该批次训练数据中每个训练实例上生成的tensor，// 根据output_rank - 1对output_tensor中的每个tensor在指定维度上进行拼接return array_ops. concat(output_tensors, output_rank - 1)

举一个例子说明下拼接的结果吧，假设该批次有2个训练实例，real_val_col对应一个real_valued_column，deep_emb_col对应一个embedding_column，real_val_col和deep_emb_col的第一个元素分别对应第一个训练实例上的feature column的tensor，第二个元素对应第二个训练实例上的feature column的tensor，可以参看以下演示代码的执行结果：

real_val_col = [[0.1,0.2], [0.3,0.4]]deep_emb_col = [[0.01,0.02,0.03], [0.04,0.05,0.06]]output_tensors = [] output_tensors.append(real_val_col) output_tensors.append(deep_emb_col) output_rank = 2dnn_input = array_ops.concat(output_tensors， output_rank - 1) tf.InteractiveSession().run(c) array([ [ 0.1, 0.2, 0.01, 0.02, 0.03], [ 0.30000001, 0.40000001, 0.04, 0.05, 0.06]], dtype=float32) 可以看到最终拼接的结果是[ [real_val_col[ 0], deep_emb_col[ 0]] [real_val_col[ 1], deep_emb_col[ 1]] ]

（3）构建线性模型

线性模型的公式是：y = W * x + b，x从linear_feature_columns中构建，W是各个特征的权重，b是模型的偏置。每个分类的预测值存储在linear_logits变量中。

linear_logits, _, _ = layers.weighted_sum_from_feature_columns(columns_to_tensors=features, feature_columns=linear_feature_columns, num_outputs=head.logits_dimension, // 分类数weight_collections=[linear_parent_scope], scope= scope)

weighted_sum_feature_columns()函数执行过程同构建DNN输入层的_input_from_feature_columns函数执行过程比较相似，以下是它的源代码：

// 每个特征与权重矩阵相乘得到每个分类上该特征的预测分数，将每个分类上特征的预测分数累加后加上偏置得到 // 对每个分类的最终预测值， 如图10forcolumn insorted( set(feature_columns), key=lambda x: x.key): transformed_tensor = transformer.transform(column) try: // 线性模型使用_wide_embedding_lookup_arguments返回的LinearEmbeddingArguments构建权重矩阵// SparsColumn, WeightedSparseColumn或Crossed Column定义了wide_embedding_lookup_argumentsembedding_lookup_arguments = column. _wide_embedding_lookup_arguments( transformed_tensor) // create_embedding_lookup()的实现同embedding_lookup_arguments()比较类似，variable中返回权重矩// 阵W，predictions返回 W * Xvariable, predictions = _create_embedding_lookup(column, columns_to_tensors, embedding_lookup_arguments, num_outputs, trainable, weight_collections) except NotImplementedError: // 其他特征直接作为线性模型的输入 // 如果输入是Sparse tensor， 需要转换为dense tensor。tensor = column. _to_dense_tensor(transformed_tensor) // 展开成二维数组[batch_size, dimension of feature column]tensor = _maybe_reshape_input_tensor( tensor, column.name, output_rank= 2) // variable定义线性模型的权重W// W是二维矩阵，行是特征的取值个数，列是分类个数// W初始化为零矩阵// 设置trainable=True, W加入GraphKeys.TRAINABLE_VARIABLESvariable= [ contrib_variables.model_variable(name= 'weight', shape=[tensor.get_shape()[ 1], num_outputs], initializer=init_ops.zeros_initializer(), trainable=trainable, collections=weight_collections)] // y’ = W * xpredictions = math_ops.matmul(tensor, variable[ 0], name= 'matmul') output_tensors.append(array_ops.reshape(predictions, shape=(- 1, num_outputs))) column_to_variable[column] = variable//在每个分类上，将所有feature column对该分类的预测值相加predictions_no_bias = math_ops.add_n(output_tensors) // bias是模型的偏置项，是一个一维向量，向量大小是输出分类的个数，初始化为0// 设置Trainable=True, bias加入GraphKeys.TRAINABLE_VARIABLESbias = contrib_variables.model_variable( 'bias_weight',shape=[num_outputs], initializer=init_ops.zeros_initializer(),trainable=trainable, collections= _add_variable_collection(weight_collections)) // predictions: W*x+bpredictions = nn_ops.bias_add(predictions_no_bias, bias) returnpredictions, column_to_variable, bias

image

图10 线性模型训练

（4）combine

接下来将两个模型结合成一个模型：

ifdnn_logits isnotNoneandlinear_logits isnotNone: logits = dnn_logits + linear_logits

在原论文中给了logistic regression二分类情况下融合模型的预测公式：P(Y = 1|x) = σ(w^T * wide[x, φ(x)] + w‘ ^T * deep[final activation] + b)，其中w^T * wide[x, φ(x)] + b 对应linear_logits，w‘ ^T * deep[final activation]对应dnn_logits, logits是wide和deep模型预测结果的叠加。σ函数将模型预测结果进行变换生成各个分类的概率。

模型产生的误差分别反馈给DNN和线性模型进行参数更新：

def_make_training_op(training_loss):… // training_loss反馈到DNN模型 // DNNLinearCombinedClassifier中定义的dnn_optimizer用在这里 ifdnn_logits isnotNone: train_ops.append( optimizers.optimize_loss( loss=training_loss,learning_rate=_DNN_LEARNING_RATE, optimizer=dnn_optimizer, …) // training_loss反馈到线性模型，DNNLinearCombinedClassifier中定义的linear_optimizer用在这 iflinear_logits isnotNone: train_ops.append( optimizers.optimize_loss( loss=training_loss, learning_rate=_linear_learning_rate(len(linear_feature_columns)), optimizer=linear_optimizer, ... ) returnhead.create_model_fn_ops(features=features,mode=mode,labels=labels, train_op_fn=_make_training_op, logits=logits)

head.create_model_fn_ops完成模型算子的定义，包括损失函数计算，每个分类概率的计算，模型评测指标的计算等。DNNLinearCombinedClassifier在初始化时根据分类个数创建head成员变量，如果是二分类，创建_BinaryLogisticHead对象，如果是多分类，创建_MultiClassHead对象。head变量的类型定义可参考源代码文件：head.py。

head = head_lib.multi_class_head( n_classes=n_classes,weight_column_name=weight_column_name, enable_centered_bias=enable_centered_bias)

我们看看二分类的_BinaryLogisticHead类如何生成model_fn：

def create_model_fn_ops( self, features,mode,labels =None, train_op_fn =None,logits =None,logits_input =None, scope =None): withvariable_scope .variable_scope(scope, default_name =self.head_name or"binary_logistic_head", values =(tuple(six .itervalues(features)) +(labels, logits, logits_input))): // 生成label tensor，维度是1个batch中训练样本的个数， 每个元素是一个样本的标注labels =self._transform_labels(mode =mode, labels =labels) // 生成logits tensor，维度与label tensor保持一致，每个元素是一个训练样本的模型输出值logits =_logits(logits_input, logits, self.logits_dimension) // 构建模型训练算子return_create_model_fn_ops(features =features, mode =mode, loss_fn =self._loss_fn, // 损失函数，默认使用交叉熵损失(_log_loss_with_two_classes) logits_to_predictions_fn =self._logits_to_predictions, // 计算每个分类的概率，对应上面提到的σ函数metrics_fn =self._metrics, // 模型指标计算，包括AUC，accuracy等，见MetricKey的定义...labels =labels, train_op_fn =train_op_fn, // 根据训练误差进行模型参数的更新函数logits =logits, // 模型的预测值...)

二分类默认使用sigmoid cross entropy计算损失函数，cross entropy乘以训练样本的权重得到训练样本的损失函数值。在批次训练中，一批训练样本的损失函数值定义为：weighted_loss = ( sum { weight[i] * loss[i]} ) / N， N是该批训练样本的个数。

def_log_loss_with_two_classes(labels, logits, weights=None):withops.name_scope( None, "log_loss_with_two_classes", (logits, labels)) asname: logits = ops.convert_to_tensor(logits) labels = math_ops.to_float(labels) // label转换为float类型的数组 ... loss = nn.sigmoid_cross_entropy_with_logits(labels=labels, logits=logits, name=name) return_compute_weighted_loss(loss, weights)

_logits_to_predictions函数根据PredictionKey计算每个分类的概率或者输出最大概率的分类，logits是模型未经变换的输出：

def _logits_to_predictions(self, logits): with ops .name_scope(None, "predictions", (logits,)): two_class_logits = _one_class_to_two_class_logits(logits) return { // LOGITS：直接输出模型对每个分类的预测值 prediction_key .PredictionKey.LOGITS: logits, // LOGISITC: simgoid变换 prediction_key .PredictionKey.LOGISTIC: math_ops .sigmoid(logits, name=prediction_key .PredictionKey.LOGISTIC), // SOFTMAX: softmax变换 prediction_key .PredictionKey.PROBABILITIES: nn .softmax( two_class_logits, name=prediction_key .PredictionKey.PROBABILITIES), // CLASSES: 预测值最大的类别 prediction_key .PredictionKey.CLASSES: math_ops .argmax( two_class_logits, 1, name=prediction_key .PredictionKey.CLASSES) }

最终构建模型训练算子

def _create_model_fn_ops(features, mode, loss_fn, logits_to_predictions_fn, metrics_fn, create_output_alternatives_fn, labels=None, train_op_fn=None, logits=None, logits_dimension=None, head_name=None, weight_column_name=None,enable_centered_bias= False): // 源代码中对enable_centered_bias的解释：// enable_centered_bias: A bool. If True, estimator will learn a centered bias variable for each class. Rest of // the model structure learns the residual after centered bias.// 融合模型中，centered bias 相当于在每个分类的输出中加上bias variable:// logits = (dnn_logits + linear_logits) + centered_bias，// 分别为(dnn_logits+linear_logits) 以及centered_bias计算loss生成training opifenable_centered_bias: centered_bias = _centered_bias(logits_dimension, head_name) { // centered_bias的第一个维度是分类个数，初始化为0，trainable设置为Truecentered_bias = variable_scope.variable(name= "centered_bias_weight", initial_value=array_ops.zeros(shape=(logits_dimension,)), trainable= True) returncentered_bias } logits = nn.bias_add(logits, centered_bias) predictions = logits_to_predictions_fn(logits) // 计算模型的损失if(mode != model_fn.ModeKeys.INFER) and(labels is not None): // input_fn返回的features dictionary中weight_column_name列对应的是weight tensorweight_tensor = _weight_tensor(features, weight_column_name) // logits对应dnn_logits + linear_logits + centered_biasloss, weighted_average_loss = loss_fn(labels, logits, weight_tensor) //根据loss进行参数更新，计算评测指标ifmode == model_fn.ModeKeys.TRAIN: batch_size = array_ops.shape(logits)[ 0] // 生成training optrain_op = _train_op(loss, labels, train_op_fn, centered_bias, batch_size, loss_fn, weight_tensor) // 生成metric op eval_metric_ops = metrics_fn(weighted_average_loss, predictions, labels, weight_tensor) // 返回模型算子returnmodel_fn.ModelFnOps( mode=mode, predictions=predictions, // 分类预测值loss=loss, // 模型的误差损失train_op=train_op, // 模型训练算子，定义模型参数的更新函数，包括DNN和线// 性模型的参数以及centered_bias参数eval_metric_ops=eval_metric_ops, // 模型指标评测算子output_alternatives=create_output_alternatives_fn(predictions)) }

_train_op函数生成training的算子， 如果centered_bias没有定义(enable_centered_bias=False)，返回train_op_fn创建的training op； 如果定义了centered_bias，为centered_bias创建training op，并与train_op_fn创建的training_op通过control_flow_ops.group组成组合算子返回。centered_bias的training op采用AdagradOptimizer作为优化函数。下图描述了enable_centered_bias=True的情况下training op的创建过程：

image

图11 enable_centered_bias情况下训练算子创建

def_train_op(loss, labels, train_op_fn, centered_bias, batch_size, loss_fn, weights) { ifcentered_bias is not None: centered_bias_step = _centered_bias_step(centered_bias=centered_bias, batch_size=batch_size, labels=labels, loss_fn=loss_fn, weights=weights) else: centered_bias_step = None withops.name_scope(None, "train_op", (loss, labels)): train_op = train_op_fn(loss) ifcentered_bias_step is not None: train_op = control_flow_ops.group(train_op, centered_bias_step) returntrain_op } def_centered_bias_step(centered_bias, batch_size, labels, loss_fn, weights): """Creates and returns training op for centered bias."""withops.name_scope(None, "centered_bias_step", (labels,)) as name: logits_dimension = array_ops.shape(centered_bias)[ 0] // logits是[batch_size，num of output classes]的二维数组，数组中每个元素的值是centered_biaslogits = array_ops.reshape(array_ops.tile(centered_bias, (batch_size,)),(batch_size, logits_dimension)) withops.name_scope(None, "centered_bias", (labels, logits)): // 对centered bias variable计算loss， centered_bias_loss是一个batch中loss的平均值centered_bias_loss = math_ops.reduce_mean(loss_fn(labels, logits, weights), name= "training_loss") # Learn central bias by an optimizer. 0.1is a convervative learning rate fora single variable. // centered_bias_loss是centered_bias同label比较后产生的loss，同training模型产生的loss是独立的returntraining.AdagradOptimizer( 0.1).minimize(centered_bias_loss, var_list=(centered_bias,), name=name) }

模型的评价指标的计算在metrics_fn函数中完成，在BinaryLogisticHead中metrics_fn对应_metrics函数：

// eval_loss: _compute_weighted_loss()返回一个batch中per example loss的平均值 // prediction: _logits_to_prediction()返回一个batch中per example的prediction // labels: 样本标注，weights: 样本权重 def_metrics(self, eval_loss, predictions, labels, weights):"""Returns a dict of metrics keyed by name."""withops.name_scope( "metrics", values=([eval_loss, labels, weights] + list(six.itervalues(predictions)))): classes = predictions[prediction_key.PredictionKey.CLASSES] logistic = predictions[prediction_key.PredictionKey.LOGISTIC] metrics = {_summary_key(self.head_name, mkey.LOSS): metrics_lib.streaming_mean(eval_loss)} metrics[_summary_key(self.head_name, mkey.ACCURACY)] = (metrics_lib.streaming_accuracy(classes, labels, weights)) metrics[_summary_key(self.head_name, mkey.AUC)] = ( _streaming_auc(logistic, labels, weights)) metrics[_summary_key(self.head_name, mkey.AUC_PR)] = ( _streaming_auc(logistic, labels, weights, curve= "PR")) returnmetrics

到此wide and deep分类模型的具体细节已讲述完毕。下图将wide and deep模型的训练过程进行简单总结，希望对大家理解整个训练过程有所帮助。

image

图12 模型训练总结

模型应用

接下来考虑将模型应用到产品具体场景。我们通过DNNLinearCombinedClassifier构建二分类模型以最大化用户下载应用的概率。

训练和测试数据来自于半年中用户的行为日志，包括下载，使用和删除等操作行为，每条操作中记录用户ID，应用ID，操作行为标识，操作时间等参数，我们在用户行为基础上基于应用的标签和类别属性构建用户画像，用户的行为日志最终分别通过用户ID和应用ID关联用户画像和应用属性生成训练的输入数据：

image

准备好输入数据后，下一步定义每条数据的标记，特征以及权重。用户下载了应用标记为1（action = click），用户浏览了应用但是没有下载行为，则标记为0（action = skip）。标记为0的训练实例的权重设定为1.0，标记为1的实例权重通过计算用户下载该应用后的使用时间以及用户后来是否删除该应用来获取：1.0 + min(2, log(使用时间) * exp(-用户是否删除了该应用))。特征部分主要使用三种类型的特征：用户特征，应用特征以及用户和应用的交互特征。如论文中建议，我们将embedding特征和连续型特征作为DNN的输入，将交叉特征作为线性模型的输入。下表中列举了我们使用的部分特征：

每条训练实例中的应用与用户历史操作过的应用集合进行交叉构建交叉特征，包括：

（训练实例中用户看到的应用） 交叉 （用户之前的历史下载应用）

（训练实例中用户看到的应用） 交叉 （用户之前的历史删除应用）

在构建交叉特征时，原论文中提到的是user impressioned app同user installed app进行交叉，实际中我们需要的是将用户下载历史中的多个app与impressioned app一一进行交叉，通过一个crossed column是无法完成的，我们采取的方式是将用户下载历史中的应用按照下载时间排序，取最近下载的20个应用，标记为1到20，然后用每个应用的ID构造一个sparse integerized feature column，与用户的impressioned app ID创建的sparse integerized feature column进行交叉，生成20个crossed feature column:

impress_col = sparse_integerized_feat_column(impressioned_app .ID) for i in( 1, 20): wide_col = sparse_integerized_feat_column(download_app[i] .ID) cross_col = crossed_column(impress_col, wide_col)

输入到DNN的连续型特征我们通过定义real valued feature column的normalizer进行L2归一化，normalizer定义和使用方式如下：

def l2_real_val_col_normalizer( x) return tf .nn.l2_normalize( x, 0) real_valued_col = tf .contrib.layers.real_valued_column( "real_feat_col", normalizer = l2_real_val_col_normalizer)

L1归一化可以定义如下normalizer：

def l1_real_val_col_normalizer(x) sum= math_ops.reduce_sum(math_ops. abs(x), 0, keep_dims=True) inv_norm = 1.0/ (math_ops.maximum( sum, epsilon)) returnmath_ops. multiply(x, inv_norm)

同时为用户下载和删除历史中的每一个应用单独构建一个embedding column作为DNN输入。

标记和特征构建好后，下一步是切分训练和测试数据集，我们按照事件的时间戳对数据按照7：3的比例切分成训练和测试集，并且从测试集中去掉所有在训练集中出现过的（用户，应用） 对。 在训练之前我们还做了3件事情：（1）在训练集上对部分特征做了相关性分析，看特征之间是否存在比较强的线性相关性；（2） 统计训练数据上的正负样本比例，通过对正样本进行repeated boostrap sampling增加正样本比例。

最终模型训练中我们还尝试了：（1）enable centered bias； （2）enable layer batch normalization；（3）尝试DNN的其他优化函数。 在我们的测试集合上模型的AUC如下：

可以看到： 结合模型比单个模型有一定提高，但是相对wide模型提高不大，主要是DNN模型的预测能力相比不是特别理想，分析原因可能是由于：1）训练数据量不够；2）需要为DNN挖掘更多更有区分能力的特征。

模型训练评测后下一步准备上线，模型需要export后才能被tensorflow serving加载使用。目前我们没有使用上tensorflow1.0以后的export_savedmodel，仍然使用的是export函数：

m.export( export_dir= export_dir, use_deprecated_input_fn=True,signature_fn=signature_fn, exports_to_keep= 3)

export中通过signature_fn定义模型的输入和输出，signature_fn我们是仿照estimator.classification_signature_fn_with_prob进行定义的，不同点是我们增加了对named_graph_sigatures的定义：

defsignature_fn(examples, unused_features, predictions):ifisinstance(predictions, dict): default_signature = exporter.classification_signature( examples, scores_tensor=predictions[ 'probabilities']) named_graph_signatures = { 'inputs': exporter.generic_signature({ 'values': examples}), 'outputs': exporter.generic_signature({ 'preds': predictions[ 'probabilities']})} else: default_signature = exporter.classification_signature( examples, scores_tensor=predictions) named_graph_signatures = { 'inputs': exporter.generic_signature({ 'values': examples}), 'outputs': exporter.generic_signature({ 'preds': predictions})} returndefault_signature, named_graph_signatures

线上的模型服务我们启动时需要传入参数–use_saved_model=false

，这样tensorflow serving做prediction时会使用SessionBundlePredict，输入从’input’字段获取，输出从’output’字段获取，具体可看TensorflowPredictor的代码。业务服务用C++编写client构造PredictRequest来访问模型服务：

using tensorflow::DataType; using tensorflow::Example; using tensorflow::Feature; using tensorflow::TensorProto; using tensorflow:: serving::PredictRequest; using tensorflow:: serving::PredictResponse; using tensorflow:: serving::PredictionService; Example example; google:: protobuf::Map<string, Feature>& feature = example.mutable_features ()->mutable_feature(); Feature feature; feature.mutable_float_list ()->add_value( 1.0); feature[“key”] = feature; string serialized_example; example.SerializeToString(&serialized_example); TensorProto tensor_proto; tensor_proto.set_dtype( DataType::DT_STRING); tensor_proto.add_string_val(serialized_example); tensor_proto.mutable_tensor_shape ()->add_dim ()->set_size( 1); PredictRequest request; request.mutable_model_spec ()->set_name(FLAGS_model_spec); (request.mutable_inputs())[ "values"] = tensor_proto; PredictResponse resp; client.Predict(request, &resp);

以上是关于wide and deep模型我们的分享，随着tensorflow的逐渐演进，模型代码有可能会发生比较大的改变，希望借这个分享让各位读者对tensorflow wide and deep模型本身的思想和tensorflow的编程方式有一定了解。

以下是wide and deep模型的参考资料：

[1] wide and deep模型的原论文：https://arxiv.org/pdf/1606.07792.pdf

[2] wide and deep模型的tutorial: https://www.tensorflow.org/tutorials/wide_and_deep

TensorFlow Wide And Deep 模型详解与应用（二）

前面讲了模型输入的特征，下面谈谈模型本身。关于 wide and deep 模型官方教程中有一段描述：The wide models and deep models are combined by summing up their final output log odds as the prediction, then feeding the prediction to a logistic loss function。从这里大概看出，线性模型与 DNN 模型进行结合的方式是对两者的预测结果进行相加，然后把相加过后的值拿去计算分类的 loss。下面我们在源代码级别把上面这段描述展开，详细分析 wide 端模型和 deep 端模型是如何实现结合的。

由于我们运用的只是分类模型，所以就不对回归模型进行分析。

DNNLinearCombinedClassifier 类继承于类 Estimator，Estimator 类继承于类 BaseEstimator。BaseEstimator 是一个抽象类，定义了通用的模型训练以及评测的函数接口 (train_model, evaluate_model, infer_model)，Estimator 类中用一个统一函数 call_model_fn 来实现 train_model, evaluate_model, infer_model。

image

图 7 estimator 的类关系图

为了更好了解整个过程，我们看看内部函数的调用过程（代码可以参见 estimator/estimator.py）：

image

图 8 Estmiator 类的函数调用图

模型训练通过调用 BaseEstimator 的 fit() 接口开始，其调用栈是：fit -> _train_model -> _get_train_ops ->_call_model_fn(ModelKeys.TRAIN) -> _model_fn，最终_model_fn() 产生模型并通过 export 函数将模型输出到 model_dir 对应目录中。

我们把训练模型的调用过程在代码级别展开，标出关键的几个函数和数据结构，省略不关键的代码，希望能让读者看到训练模型的大致过程：

image

图 9 模型训练的调用栈

评测（evaluate）和预测（predict）的过程与训练（train）大致相同，读者可以通过源代码文件找到对应函数了解。可以看出，整个函数调用栈中最关键的 2 个函数是： input_fn 和 model_fn。input_fn 从输入数据中生成 features 和 labels，features 是一个 Tensor 或者是一个从特征名到 Tensor 的字典，如果 features 是一个 Tensor，程序会给这个 Tensor 一个空字符串的键值，转换成特征名到 Tensor 的字典。labels 是样本的 label 构成的 tensor。input_fn 由应用程序调用者提供实现，返回（features, labels）二元组，要求 tf.get_shape(features)[0] == tf.get_shape(labels)[0]，也就是两个 tensor 的行数目得保持一致。model_fn 定义训练和评测模型的具体逻辑，如模型训练产生的误差 (model_fn_ops.loss) 以及训练算子（model_fn_ops.train_op）通过封装在 EstmiatorSpec 的对象中由 training 的 Session 进行调用。每个具体模型需要实现的是自定义的 model_fn。

DNNLinearCombinedClassifier 是如何实现自己的 model_fn 的呢？本文开头我们给出了它的初始化函数原型，进入初始化函数的实现中我们定位到代码行 model_fn=_dnn_linear_combined_model_fn。

这个就是 DNNLinearCombinedClassifier 的 model_fn。这个函数的定义如下：

def_dnn_linear_combined_model_fn(features, labels, mode, params, config= None)

features 和 labels 大家都已经知道，mode 指定 model_fn 的操作模式，目前支持 3 个值：训练模型 (model_fn.ModeKeys.TRAIN)，对模型进行评测 (model_fn.ModeKeys.EVAL)，根据输入特征进行预测 (model_fn.ModeKeys.PREDICT)，mode 的定义可参见文件 estimator/model_fn.py。params 和 config 参数分别定义模型训练的参数以及模型运行的配置。

模型训练的整个过程可以分为如下三步：

（1）从 params 获取参数，定义模型优化函数

// 将输入的特征统一为字符串到 tensor 的 dictfeatures = _get_feature_dict(features) // 定义线性模型的优化函数linear_optimizer = params. get( "linear_optimizer") or"Ftrl"linear_optimizer = _get_optimizer(linear_optimizer) // 定义 DNN 模型的优化函数dnn_optimizer = params. get( "dnn_optimizer") or"Adagrad"dnn_optimizer = _get_optimizer(dnn_optimizer)

（2）构建 DNN 模型

首先从 dnn_feature_columns 构造 DNN 模型输入层：

dnn_feature_columns = params. get( "dnn_feature_columns") // dnn_feature_columns 中包括 DNN 模型需要的所有连续特征和 embedding 特征net = layers.input_from_feature_columns(columns_to_tensors=features, feature_columns=dnn_feature_columns, weight_collections=[dnn_parent_scope], scope=dnn_input_scope)

这里的 features 是 feature column key 到 tensor 的 dict，在本文前面我们对每个特征都谈了它的 insert_transformed_feature(columns_to_tensors)，在构建 DNN 模型输入层之前会调用每个 FeatureColumn 的 insert_transformed_feature 函数生成一个浮点数类型的 tensor 作为 DNN 输入层的一部分。

构建完输入层后，然后至底向上进行隐藏层构建，每层的隐藏单元的个数在参数

dnn_hidden_units 中定义： dnn_hidden_units = params. get( "dnn_hidden_units") forlayer_id, num_hidden_units inenumerate(dnn_hidden_units): // 从输入层或者是下面一层隐藏层构建新的隐藏层// dnn_activation_fn 定义隐藏层的激活函数，默认使用 RELUdnn_activation_fn = params. get( "dnn_activation_fn") ornn.relu。 net = layers.fully_connected(net，num_hidden_units，activation_fn=dnn_activation_fn， variables_collections=[dnn_parent_scope], scope=dnn_hidden_layer_scope) // 模型训练中对隐藏层进行 drop-out// dnn_dropout 是舍弃隐藏单元输出的概率dnn_dropout = params. get( "dnn_dropout") ifdnn_dropout is notNone andmode == model_fn.ModeKeys.TRAIN: net = layers.dropout(net, keep_prob=( 1.0- dnn_dropout))

从最后一层隐藏层到输出层建立一个全连接就完成 DNN 模型的构建了, dnn_logits 是模型输出，它是一个分类数个数的 Tensor，Tensor 的每个元素对应 1 个分类的线性激活值（linear activation: w * net + bias）：

withvariable_scope.variable_scope( "logits", values=(net,)) asdnn_logits_scope: dnn_logits = layers.fully_connected( net, head.logits_dimension, // 分类个数activation_fn=None, // 输出层不做非线性变换variables_collections=[dnn_parent_scope], scope=dnn_logits_scope)

从 dnn feature columns 中构建 DNN 输入层的工作主要由函数 layers._input_from_feature_columns 完成，每个训练样本上的 dnn feature column 的值拼接成一个 dense tensor 作为 DNN 的输入。

该函数的定义如下：

def _input_from_feature_columns( columns_to_tensors, feature_columns, weight_collections, trainable, scope, output_rank, default_name) transformer = _Transformer(columns_to_tensors) forcolumn insorted( set(feature_columns), key=lambda x: x.key): // 调用 feature column 的特征变换函数 insert_transformed_feature()// 返回 feature column 变换后的 tensor： columns_to_tensors[column]。transformed_tensor = transformer.transform(column) try: // 构建_EmbeddingColumn 的 embeddingarguments = column._deep_embedding_lookup_arguments(transformed_tensor) output_tensors.append(_embeddings_from_arguments(column, arguments, weight_collections, trainable, output_rank=output_rank)) exceptNotImplementedError asee: // 构建_RealValuedColumn 的 tensortry: output_tensors.append(column._to_dnn_input_layer(transformed_tensor,weight_collections, trainable,output_rank=output_rank)) exceptValueError ase: // output_tensors 数组每个元素对应一个 feature column 在该批次训练数据中每个训练实例上生成的 tensor，// 根据 output_rank - 1 对 output_tensor 中的每个 tensor 在指定维度上进行拼接return array_ops. concat(output_tensors, output_rank - 1)

举一个例子说明下拼接的结果吧，假设该批次有 2 个训练实例，real_val_col 对应一个 real_valued_column，deep_emb_col 对应一个 embedding_column，real_val_col 和 deep_emb_col 的第一个元素分别对应第一个训练实例上的 feature column 的 tensor，第二个元素对应第二个训练实例上的 feature column 的 tensor，可以参看以下演示代码的执行结果：

real_val_col = [[0.1,0.2], [0.3,0.4]]deep_emb_col = [[0.01,0.02,0.03], [0.04,0.05,0.06]]output_tensors = [] output_tensors.append(real_val_col) output_tensors.append(deep_emb_col) output_rank = 2dnn_input = array_ops.concat(output_tensors，output_rank - 1) tf.InteractiveSession().run(c) array([ [ 0.1, 0.2, 0.01, 0.02, 0.03], [ 0.30000001, 0.40000001, 0.04, 0.05, 0.06]], dtype=float32) 可以看到最终拼接的结果是 [ [real_val_col[ 0], deep_emb_col[ 0]] [real_val_col[ 1], deep_emb_col[ 1]] ]

（3）构建线性模型

线性模型的公式是：y = W * x + b，x 从 linear_feature_columns 中构建，W 是各个特征的权重，b 是模型的偏置。每个分类的预测值存储在 linear_logits 变量中。

linear_logits, _, _ = layers.weighted_sum_from_feature_columns(columns_to_tensors=features, feature_columns=linear_feature_columns, num_outputs=head.logits_dimension, // 分类数weight_collections=[linear_parent_scope], scope= scope)

weighted_sum_feature_columns() 函数执行过程同构建 DNN 输入层的_input_from_feature_columns 函数执行过程比较相似，以下是它的源代码：

// 每个特征与权重矩阵相乘得到每个分类上该特征的预测分数，将每个分类上特征的预测分数累加后加上偏置得到对每个分类的最终预测值，如图 10
for column insorted( set(feature_columns), key=lambda x: x.key):
transformed_tensor = transformer.transform(column)
try: // 线性模型使用_wide_embedding_lookup_arguments 返回的 LinearEmbeddingArguments 构建权重矩阵;SparsColumn, WeightedSparseColumn 或 Crossed Column 定义了 wide_embedding_lookup_arguments;
embedding_lookup_arguments = column. _wide_embedding_lookup_arguments( transformed_tensor) // create_embedding_lookup() 的实现同 embedding_lookup_arguments() 比较类似，variable 中返回权重矩阵 W，predictions 返回 W * X
variable, predictions = _create_embedding_lookup(column, columns_to_tensors, embedding_lookup_arguments, num_outputs, trainable, weight_collections)
except NotImplementedError: // 其他特征直接作为线性模型的输入 // 如果输入是 Sparse tensor，需要转换为 dense tensor。
tensor = column. _to_dense_tensor(transformed_tensor) // 展开成二维数组 [batch_size, dimension of feature column]
tensor = _maybe_reshape_input_tensor( tensor, column.name, output_rank= 2)
// variable 定义线性模型的权重 W// W 是二维矩阵，行是特征的取值个数，列是分类个数// W 初始化为零矩阵// 设置 trainable=True, W 加入 GraphKeys.TRAINABLE_VARIABLES
variable= [ contrib_variables.model_variable(name= 'weight', shape=[tensor.get_shape()[ 1], num_outputs], initializer=init_ops.zeros_initializer(), trainable=trainable, collections=weight_collections)] // y』 = W * xpredictions = math_ops.matmul(tensor, variable[ 0], name= 'matmul') output_tensors.append(array_ops.reshape(predictions, shape=(- 1, num_outputs))) column_to_variable[column] = variable//在每个分类上，将所有 feature column 对该分类的预测值相加predictions_no_bias = math_ops.add_n(output_tensors) // bias 是模型的偏置项，是一个一维向量，向量大小是输出分类的个数，初始化为 0// 设置 Trainable=True, bias 加入 GraphKeys.TRAINABLE_VARIABLESbias = contrib_variables.model_variable( 'bias_weight',shape=[num_outputs], initializer=init_ops.zeros_initializer(),trainable=trainable, collections= _add_variable_collection(weight_collections)) // predictions: W*x+bpredictions = nn_ops.bias_add(predictions_no_bias, bias) returnpredictions, column_to_variable, bias

image

图 10 线性模型训练

（4）combine

接下来将两个模型结合成一个模型：

ifdnn_logits isnotNoneandlinear_logits isnotNone: logits = dnn_logits + linear_logits

在原论文中给了 logistic regression 二分类情况下融合模型的预测公式：P(Y = 1|x) = σ(w^T * wide[x, φ(x)] + w『 ^T * deep[final activation] + b)，其中 w^T * wide[x, φ(x)] + b 对应 linear_logits，w『 ^T * deep[final activation] 对应 dnn_logits, logits 是 wide 和 deep 模型预测结果的叠加。σ函数将模型预测结果进行变换生成各个分类的概率。

模型产生的误差分别反馈给 DNN 和线性模型进行参数更新：

def_make_training_op(training_loss):… // training_loss 反馈到 DNN 模型 // DNNLinearCombinedClassifier 中定义的 dnn_optimizer 用在这里 ifdnn_logits isnotNone: train_ops.append( optimizers.optimize_loss( loss=training_loss,learning_rate=_DNN_LEARNING_RATE, optimizer=dnn_optimizer, …) // training_loss 反馈到线性模型，DNNLinearCombinedClassifier 中定义的 linear_optimizer 用在这 iflinear_logits isnotNone: train_ops.append( optimizers.optimize_loss( loss=training_loss, learning_rate=_linear_learning_rate(len(linear_feature_columns)), optimizer=linear_optimizer, ... ) returnhead.create_model_fn_ops(features=features,mode=mode,labels=labels, train_op_fn=_make_training_op, logits=logits)

head.create_model_fn_ops 完成模型算子的定义，包括损失函数计算，每个分类概率的计算，模型评测指标的计算等。DNNLinearCombinedClassifier 在初始化时根据分类个数创建 head 成员变量，如果是二分类，创建_BinaryLogisticHead 对象，如果是多分类，创建_MultiClassHead 对象。head 变量的类型定义可参考源代码文件：head.py。

head = head_lib.multi_class_head( n_classes=n_classes,weight_column_name=weight_column_name, enable_centered_bias=enable_centered_bias)

我们看看二分类的_BinaryLogisticHead 类如何生成 model_fn：

def create_model_fn_ops( self, features,mode,labels =None, train_op_fn =None,logits =None,logits_input =None, scope =None): withvariable_scope .variable_scope(scope, default_name =self.head_name or"binary_logistic_head", values =(tuple(six .itervalues(features)) +(labels, logits, logits_input))): // 生成 label tensor，维度是 1 个 batch 中训练样本的个数，每个元素是一个样本的标注labels =self._transform_labels(mode =mode, labels =labels) // 生成 logits tensor，维度与 label tensor 保持一致，每个元素是一个训练样本的模型输出值logits =_logits(logits_input, logits, self.logits_dimension) // 构建模型训练算子return_create_model_fn_ops(features =features, mode =mode, loss_fn =self._loss_fn, // 损失函数，默认使用交叉熵损失 (_log_loss_with_two_classes) logits_to_predictions_fn =self._logits_to_predictions, // 计算每个分类的概率，对应上面提到的σ函数metrics_fn =self._metrics, // 模型指标计算，包括 AUC，accuracy 等，见 MetricKey 的定义...labels =labels, train_op_fn =train_op_fn, // 根据训练误差进行模型参数的更新函数logits =logits, // 模型的预测值...)

二分类默认使用 sigmoid cross entropy 计算损失函数，cross entropy 乘以训练样本的权重得到训练样本的损失函数值。在批次训练中，一批训练样本的损失函数值定义为：weighted_loss = ( sum { weight[i] * loss[i]} ) / N，N 是该批训练样本的个数。

def_log_loss_with_two_classes(labels, logits, weights=None):withops.name_scope( None, "log_loss_with_two_classes", (logits, labels)) asname: logits = ops.convert_to_tensor(logits) labels = math_ops.to_float(labels) // label 转换为 float 类型的数组 ... loss = nn.sigmoid_cross_entropy_with_logits(labels=labels, logits=logits, name=name) return_compute_weighted_loss(loss, weights)

_logits_to_predictions 函数根据 PredictionKey 计算每个分类的概率或者输出最大概率的分类，logits 是模型未经变换的输出：

def _logits_to_predictions(self, logits): with ops .name_scope(None, "predictions", (logits,)): two_class_logits = _one_class_to_two_class_logits(logits) return { // LOGITS：直接输出模型对每个分类的预测值 prediction_key .PredictionKey.LOGITS: logits, // LOGISITC: simgoid 变换 prediction_key .PredictionKey.LOGISTIC: math_ops .sigmoid(logits, name=prediction_key .PredictionKey.LOGISTIC), // SOFTMAX: softmax 变换 prediction_key .PredictionKey.PROBABILITIES: nn .softmax( two_class_logits, name=prediction_key .PredictionKey.PROBABILITIES), // CLASSES: 预测值最大的类别 prediction_key .PredictionKey.CLASSES: math_ops .argmax( two_class_logits, 1, name=prediction_key .PredictionKey.CLASSES) }

最终构建模型训练算子

def _create_model_fn_ops(features, mode, loss_fn, logits_to_predictions_fn, metrics_fn, create_output_alternatives_fn, labels=None, train_op_fn=None, logits=None, logits_dimension=None, head_name=None, weight_column_name=None,enable_centered_bias= False): // 源代码中对 enable_centered_bias 的解释：// enable_centered_bias: A bool. If True, estimator will learn a centered bias variable for each class. Rest of // the model structure learns the residual after centered bias.// 融合模型中，centered bias 相当于在每个分类的输出中加上 bias variable:// logits = (dnn_logits + linear_logits) + centered_bias，// 分别为 (dnn_logits+linear_logits) 以及 centered_bias 计算 loss 生成 training opifenable_centered_bias: centered_bias = _centered_bias(logits_dimension, head_name) { // centered_bias 的第一个维度是分类个数，初始化为 0，trainable 设置为 Truecentered_bias = variable_scope.variable(name= "centered_bias_weight", initial_value=array_ops.zeros(shape=(logits_dimension,)), trainable= True) returncentered_bias } logits = nn.bias_add(logits, centered_bias) predictions = logits_to_predictions_fn(logits) // 计算模型的损失if(mode != model_fn.ModeKeys.INFER) and(labels is not None): // input_fn 返回的 features dictionary 中 weight_column_name 列对应的是 weight tensorweight_tensor = _weight_tensor(features, weight_column_name) // logits 对应 dnn_logits + linear_logits + centered_biasloss, weighted_average_loss = loss_fn(labels, logits, weight_tensor) //根据 loss 进行参数更新，计算评测指标ifmode == model_fn.ModeKeys.TRAIN: batch_size = array_ops.shape(logits)[ 0] // 生成 training optrain_op = _train_op(loss, labels, train_op_fn, centered_bias, batch_size, loss_fn, weight_tensor) // 生成 metric op eval_metric_ops = metrics_fn(weighted_average_loss, predictions, labels, weight_tensor) // 返回模型算子returnmodel_fn.ModelFnOps( mode=mode, predictions=predictions, // 分类预测值loss=loss, // 模型的误差损失train_op=train_op, // 模型训练算子，定义模型参数的更新函数，包括 DNN 和线// 性模型的参数以及 centered_bias 参数eval_metric_ops=eval_metric_ops, // 模型指标评测算子output_alternatives=create_output_alternatives_fn(predictions)) }

_train_op 函数生成 training 的算子，如果 centered_bias 没有定义 (enable_centered_bias=False)，返回 train_op_fn 创建的 training op； 如果定义了 centered_bias，为 centered_bias 创建 training op，并与 train_op_fn 创建的 training_op 通过 control_flow_ops.group 组成组合算子返回。centered_bias 的 training op 采用 AdagradOptimizer 作为优化函数。下图描述了 enable_centered_bias=True 的情况下 training op 的创建过程：

image

图 11 enable_centered_bias 情况下训练算子创建 def_train_op(loss, labels, train_op_fn, centered_bias, batch_size, loss_fn, weights) { ifcentered_bias is not None: centered_bias_step = _centered_bias_step(centered_bias=centered_bias, batch_size=batch_size, labels=labels, loss_fn=loss_fn, weights=weights) else: centered_bias_step = None withops.name_scope(None, "train_op", (loss, labels)): train_op = train_op_fn(loss) ifcentered_bias_step is not None: train_op = control_flow_ops.group(train_op, centered_bias_step) returntrain_op } def_centered_bias_step(centered_bias, batch_size, labels, loss_fn, weights): """Creates and returns training op for centered bias."""withops.name_scope(None, "centered_bias_step", (labels,)) as name: logits_dimension = array_ops.shape(centered_bias)[ 0] // logits 是 [batch_size，num of output classes] 的二维数组，数组中每个元素的值是 centered_biaslogits = array_ops.reshape(array_ops.tile(centered_bias, (batch_size,)),(batch_size, logits_dimension)) withops.name_scope(None, "centered_bias", (labels, logits)): // 对 centered bias variable 计算 loss，centered_bias_loss 是一个 batch 中 loss 的平均值centered_bias_loss = math_ops.reduce_mean(loss_fn(labels, logits, weights), name= "training_loss") # Learn central bias by an optimizer. 0.1is a convervative learning rate fora single variable. // centered_bias_loss 是 centered_bias 同 label 比较后产生的 loss，同 training 模型产生的 loss 是独立的returntraining.AdagradOptimizer( 0.1).minimize(centered_bias_loss, var_list=(centered_bias,), name=name) }

模型的评价指标的计算在 metrics_fn 函数中完成，在 BinaryLogisticHead 中 metrics_fn 对应_metrics 函数：

// eval_loss: _compute_weighted_loss() 返回一个 batch 中 per example loss 的平均值 // prediction: _logits_to_prediction() 返回一个 batch 中 per example 的 prediction // labels: 样本标注，weights: 样本权重 def_metrics(self, eval_loss, predictions, labels, weights):"""Returns a dict of metrics keyed by name."""withops.name_scope( "metrics", values=([eval_loss, labels, weights] + list(six.itervalues(predictions)))): classes = predictions[prediction_key.PredictionKey.CLASSES] logistic = predictions[prediction_key.PredictionKey.LOGISTIC] metrics = {_summary_key(self.head_name, mkey.LOSS): metrics_lib.streaming_mean(eval_loss)} metrics[_summary_key(self.head_name, mkey.ACCURACY)] = (metrics_lib.streaming_accuracy(classes, labels, weights)) metrics[_summary_key(self.head_name, mkey.AUC)] = ( _streaming_auc(logistic, labels, weights)) metrics[_summary_key(self.head_name, mkey.AUC_PR)] = ( _streaming_auc(logistic, labels, weights, curve= "PR")) returnmetrics

到此 wide and deep 分类模型的具体细节已讲述完毕。下图将 wide and deep 模型的训练过程进行简单总结，希望对大家理解整个训练过程有所帮助。

image

图 12 模型训练总结 Wide And Deep 模型应用

接下来考虑将模型应用到产品具体场景。我们通过 DNNLinearCombinedClassifier 构建二分类模型以最大化用户下载应用的概率。

训练和测试数据来自于半年中用户的行为日志，包括下载，使用和删除等操作行为，每条操作中记录用户 ID，应用 ID，操作行为标识，操作时间等参数，我们在用户行为基础上基于应用的标签和类别属性构建用户画像，用户的行为日志最终分别通过用户 ID 和应用 ID 关联用户画像和应用属性生成训练的输入数据：

image

准备好输入数据后，下一步定义每条数据的标记，特征以及权重。用户下载了应用标记为 1（action = click），用户浏览了应用但是没有下载行为，则标记为 0（action = skip）。标记为 0 的训练实例的权重设定为 1.0，标记为 1 的实例权重通过计算用户下载该应用后的使用时间以及用户后来是否删除该应用来获取：1.0 + min(2, log(使用时间) * exp(-用户是否删除了该应用))。特征部分主要使用三种类型的特征：用户特征，应用特征以及用户和应用的交互特征。如论文中建议，我们将 embedding 特征和连续型特征作为 DNN 的输入，将交叉特征作为线性模型的输入。下表中列举了我们使用的部分特征：

图 13 模型构建过程

每条训练实例中的应用与用户历史操作过的应用集合进行交叉构建交叉特征，包括：

（训练实例中用户看到的应用）交叉（用户之前的历史下载应用）

（训练实例中用户看到的应用）交叉（用户之前的历史删除应用）

在构建交叉特征时，原论文中提到的是 user impressioned app 同 user installed app 进行交叉，实际中我们需要的是将用户下载历史中的多个 app 与 impressioned app 一一进行交叉，通过一个 crossed column 是无法完成的，我们采取的方式是将用户下载历史中的应用按照下载时间排序，取最近下载的 20 个应用，标记为 1 到 20，然后用每个应用的 ID 构造一个 sparse integerized feature column，与用户的 impressioned app ID 创建的 sparse integerized feature column 进行交叉，生成 20 个 crossed feature column:

impress_col = sparse_integerized_feat_column(impressioned_app .ID) for i in( 1, 20): wide_col = sparse_integerized_feat_column(download_app[i] .ID) cross_col = crossed_column(impress_col, wide_col)

输入到 DNN 的连续型特征我们通过定义 real valued feature column 的 normalizer 进行 L2 归一化，normalizer 定义和使用方式如下：

def l2_real_val_col_normalizer( x) return tf .nn.l2_normalize( x, 0) real_valued_col = tf .contrib.layers.real_valued_column( "real_feat_col", normalizer = l2_real_val_col_normalizer) L1 归一化可以定义如下 normalizer： def l1_real_val_col_normalizer( x) sum = math_ops .reduce_sum(math_ops .abs( x), 0, keep_dims=True) inv_norm = 1.0/ (math_ops .maximum(sum, epsilon)) return math_ops .multiply( x, inv_norm)

同时为用户下载和删除历史中的每一个应用单独构建一个 embedding column 作为 DNN 输入。

标记和特征构建好后，下一步是切分训练和测试数据集，我们按照事件的时间戳对数据按照 7：3 的比例切分成训练和测试集，并且从测试集中去掉所有在训练集中出现过的（用户，应用）对。在训练之前我们还做了 3 件事情：（1）在训练集上对部分特征做了相关性分析，看特征之间是否存在比较强的线性相关性；（2）统计训练数据上的正负样本比例，通过对正样本进行 repeated boostrap sampling 增加正样本比例。

最终模型训练中我们还尝试了：（1）enable centered bias；（2）enable layer batch normalization；（3）尝试 DNN 的其他优化函数。在我们的测试集合上模型的 AUC 如下：

可以看到： 结合模型比单个模型有一定提高，但是相对 wide 模型提高不大，主要是 DNN 模型的预测能力相比不是特别理想，分析原因可能是由于：1）训练数据量不够；2）需要为 DNN 挖掘更多更有区分能力的特征。

模型训练评测后下一步准备上线，模型需要 export 后才能被 tensorflow serving 加载使用。目前我们没有使用上 TensorFlow1.0 以后的 export_savedmodel，仍然使用的是 export 函数：

m.export( export_dir= export_dir, use_deprecated_input_fn=True,signature_fn=signature_fn, exports_to_keep= 3)

export 中通过 signature_fn 定义模型的输入和输出，signature_fn 我们是仿照 estimator.classification_signature_fn_with_prob 进行定义的，不同点是我们增加了对 named_graph_sigatures 的定义：

defsignature_fn(examples, unused_features, predictions):ifisinstance(predictions, dict): default_signature = exporter.classification_signature( examples, scores_tensor=predictions[ 'probabilities']) named_graph_signatures = { 'inputs': exporter.generic_signature({ 'values': examples}), 'outputs': exporter.generic_signature({ 'preds': predictions[ 'probabilities']})} else: default_signature = exporter.classification_signature( examples, scores_tensor=predictions) named_graph_signatures = { 'inputs': exporter.generic_signature({ 'values': examples}), 'outputs': exporter.generic_signature({ 'preds': predictions})} returndefault_signature, named_graph_signatures

线上的模型服务我们启动时需要传入参数–use_saved_model=false

，这样 tensorflow serving 做 prediction 时会使用 SessionBundlePredict，输入从』input』字段获取，输出从』output』字段获取，具体可看 TensorflowPredictor 的代码。业务服务用 C++编写 client 构造 PredictRequest 来访问模型服务：

using tensorflow::DataType; using tensorflow::Example; using tensorflow::Feature; using tensorflow::TensorProto; using tensorflow:: serving::PredictRequest; using tensorflow:: serving::PredictResponse; using tensorflow:: serving::PredictionService; Example example; google:: protobuf::Map<string, Feature>& feature = example.mutable_features ()->mutable_feature(); Feature feature; feature.mutable_float_list ()->add_value( 1.0); feature[「key」] = feature; string serialized_example; example.SerializeToString(&serialized_example); TensorProto tensor_proto; tensor_proto.set_dtype( DataType::DT_STRING); tensor_proto.add_string_val(serialized_example); tensor_proto.mutable_tensor_shape ()->add_dim ()->set_size( 1); PredictRequest request; request.mutable_model_spec ()->set_name(FLAGS_model_spec); (request.mutable_inputs())[ "values"] = tensor_proto; PredictResponse resp; client.Predict(request, &resp);

以上是关于 wide and deep 模型我们的分享，随着 TensorFlow 的逐渐演进，模型代码有可能会发生比较大的改变，希望借这个分享让各位读者对 TensorFlow wide and deep 模型本身的思想和 TensorFlow 的编程方式有一定了解。

以下是 wide and deep 模型的参考资料：

[1] wide and deep 模型的原论文：https://arxiv.org/pdf/1606.07792.pdf

[2] wide and deep 模型的 tutorial: https://www.tensorflow.org/tutorials/wide_and_deep