文章推荐系统 | 十二、基于 FTRL 优化的在线排序
推荐阅读：
文章推荐系统 | 一、推荐流程设计
文章推荐系统 | 二、同步业务数据
文章推荐系统 | 三、收集用户行为数据
文章推荐系统 | 四、构建离线文章画像
文章推荐系统 | 五、计算文章相似度
文章推荐系统 | 六、构建离线用户画像
文章推荐系统 | 七、构建离线文章特征和用户特征
文章推荐系统 | 八、基于模型的离线召回
文章推荐系统 | 九、基于内容的离线及在线召回
文章推荐系统 | 十、基于热门文章和新文章的在线召回
文章推荐系统 | 十一、基于 LR 模型的离线排序
构造 TFRecord 训练集
和前面的 LR 离线模型一样，这里也是使用 LR 模型，只是选择 FTRL 优化方法，首先也是要完成训练集的构建。在上篇文章中，我们已经知道，可以通过读取用户历史行为数据，及文章特征和用户特征，构建出训练集 train，其中包括 features 和 label 两列数据，features 是文章特征和用户特征的组合。在 TensorFlow 通常使用 TFRecord 文件进行数据的存取。接下来，我们就要将 train 保存到 TFRecord 文件中。首先开启会话，将 train 中的特征和标签分别传入 write_to_tfrecords() 方法，并利用多线程执行
import tensorflow as tf
with tf.Session() as sess:
# 创建线程协调器
coord = tf.train.Coordinator()
# 开启子线程去读取数据
threads = tf.train.start_queue_runners(sess=sess, coord=coord)
# 存入数据
write_to_tfrecords(train.iloc[:, 0], train.iloc[:, 1])
# 关闭子线程，回收
coord.request_stop()
coord.join(threads)
接着，在 write_to_tfrecords() 方法中，遍历训练集数据，将每个样本构造为 tf.train.Example，其中 feature 为 BytesList 类型，label 为 Int64List 类型，并保存到 TFRecords 文件中
def write_to_tfrecords(feature_batch, click_batch):
"""将用户与文章的点击日志构造的样本写入TFRecords文件
"""
# 1、构造tfrecords的存储实例
writer = tf.python_io.TFRecordWriter("./train_ctr_20190605.tfrecords")
# 2、循环将所有样本一个个封装成example，写入文件
for i in range(len(click_batch)):
# 取出第i个样本的特征值和目标值，格式转换
click = click_batch[i]
feature = feature_batch[i].tostring()
# 构造example
example = tf.train.Example(features=tf.train.Features(feature={
"feature": tf.train.Feature(bytes_list=tf.train.BytesList(value=[feature])),
"label": tf.train.Feature(int64_list=tf.train.Int64List(value=[click]))
}))
# 序列化example,写入文件
writer.write(example.SerializeToString())
writer.close()
离线训练
FTRL（Follow The Regularized Leader）是一种获得稀疏模型的优化方法，我们利用构建好的 TFRecord 样本数据对 LR 模型进行离线训练。首先，定义 read_ctr_records() 方法来读取 TFRecord 文件，并通过调用 parse_tfrecords() 方法遍历解析每个样本，并设置了批大小和迭代次数
def read_ctr_records():
train = tf.data.TFRecordDataset(["./train_ctr_20190605.tfrecords"])
train = train.map(parse_tfrecords)
train = train.batch(64)
train = train.repeat(10000)
解析每个样本，将 TFRecord 中序列化的 feature 列，解析成 channel_id (1), article_vector (100), user_weights (10), article_weights (10)
FEATURE_COLUMNS = ['channel_id', 'article_vector', 'user_weigths', 'article_weights']
def parse_tfrecords(example):
features = {
"feature": tf.FixedLenFeature([], tf.string),
"label": tf.FixedLenFeature([], tf.int64)
}
parsed_features = tf.parse_single_example(example, features)
feature = tf.decode_raw(parsed_features['feature'], tf.float64)
feature = tf.reshape(tf.cast(feature, tf.float32), [1, 121])
channel_id = tf.cast(tf.slice(feature, [0, 0], [1, 1]), tf.int32)
article_vector = tf.reduce_sum(tf.slice(feature, [0, 1], [1, 100]), axis=1)
user_weights = tf.reduce_sum(tf.slice(feature, [0, 101], [1, 10]), axis=1)
article_weights = tf.reduce_sum(tf.slice(feature, [0, 111], [1, 10]), axis=1)
label = tf.cast(parsed_features['label'], tf.float32)
# 构造字典 名称-tensor
tensor_list = [channel_id, article_vector, user_weights, article_weights]
feature_dict = dict(zip(FEATURE_COLUMNS, tensor_list))
return feature_dict, label
指定输入特征的数据类型，并定义 LR 模型 model 及 FTRL 优化方法
# 定义离散类型特征
article_id = tf.feature_column.categorical_column_with_identity('channel_id', num_buckets=25)
# 定义连续类型特征
article_vector = tf.feature_column.numeric_column('article_vector')
user_weigths = tf.feature_column.numeric_column('user_weigths')
article_weights = tf.feature_column.numeric_column('article_weights')
feature_columns = [article_id, article_vector, user_weigths, article_weights]
model = tf.estimator.LinearClassifier(feature_columns=feature_columns,
optimizer=tf.train.FtrlOptimizer(learning_rate=0.1,
l1_regularization_strength=10,
l2_regularization_strength=10))
通过调用 read_ctr_records() 方法，来读取 TFRecod 文件中的训练数据，并设置训练步长，对定义好的 LR 模型进行训练及预估
model.train(read_ctr_records, steps=1000)
result = model.evaluate(read_ctr_records)
通常需要编写离线任务，定时读取用户行为数据作为训练集和验证集，对训练集及验证集进行 CTR 预估，并根据离线指标对结果进行分析，决定是否更新模型。
在线排序
通常在线排序是根据用户实时的推荐请求，对召回结果进行 CTR 预估，进而计算出排序结果并返回。我们需要根据召回结果构造测试集，其中每个测试样本包括用户特征和文章特征。首先，根据用户 ID 和频道 ID 读取用户特征（用户在每个频道的特征不同，所以是分频道存储的）
try:
user_feature = eval(hbu.get_table_row('ctr_feature_user',
'{}'.format(temp.user_id).encode(),
'channel:{}'.format(temp.channel_id).encode()))
except Exception as e:
user_feature = []
再根据用户 ID 读取召回结果
recall_set = read_hbase_recall('cb_recall',
'recall:user:{}'.format(temp.user_id).encode(),
'als:{}'.format(temp.channel_id).encode())
接着，遍历召回结果，获取文章特征，并将用户特征合并，作为测试样本
test = []
for article_id in recall_set:
try:
article_feature = eval(hbu.get_table_row('ctr_feature_article',
'{}'.format(article_id).encode(),
'article:{}'.format(article_id).encode()))
except Exception as e:
article_feature = []
if not article_feature:
article_feature = [0.0] * 111
feature = []
feature.extend(user_feature)
feature.extend(article_feature)
test.append(f)
加载本地 LR 模型并对测试样本进行 CTR 预估
test_array = np.array(test)
model.load_weights('/root/toutiao_project/reco_sys/offline/models/ckpt/ctr_lr_ftrl.h5')
init = tf.global_variables_initializer()
with tf.Session() as sess:
sess.run(init)
predictions = self.model.predict(sess.run(tf.constant(test_array)))
对结果进行排序并提取 CTR 最高的前 K 个文章，这样就得到了 FTRL 优化的在线排序的结果。
res = pd.DataFrame(np.concatenate((np.array(recall_set).reshape(len(recall_set), 1), predictions),
axis=1), columns=['article_id', 'prob'])
res_sort = res.sort_values(by=['prob'], ascending=True)
# 排序后，只将排名在前100个文章ID作为推荐结果返回给用户
if len(res_sort) > 100:
recall_set = list(res_sort.iloc[:100, 0])
recall_set = list(res_sort.iloc[:, 0])
参考
https://www.bilibili.com/video/av68356229
https://pan.baidu.com/s/1-uvGJ-mEskjhtaial0Xmgw（学习资源已保存至网盘， 提取码：eakp）
