|
多路召回是什么
采用不同的策略、特征、或简单模型，分别召回一部分候选集，然后把候选集混合在一起供后续排序模型使用，可以明显看出，"多路召回“策略是在”计算速度“和”召回率“之间进行权衡的结果。
召回常用的策略
1、YoutubeDNN召回
def youtubednn_u2i_dict(data, topk=20):
sparse_features = ["click_article_id", "user_id"]
SEQ_LEN = 30 # 用户点击序列的长度，短的填充，长的截断
user_profile_ = data[["user_id"]].drop_duplicates('user_id')
item_profile_ = data[["click_article_id"]].drop_duplicates('click_article_id')
# 类别编码
features = ["click_article_id", "user_id"]
feature_max_idx = {}
for feature in features:
lbe = LabelEncoder()
data[feature] = lbe.fit_transform(data[feature])
feature_max_idx[feature] = data[feature].max() + 1
# 提取user和item的画像，这里具体选择哪些特征还需要进一步的分析和考虑
user_profile = data[["user_id"]].drop_duplicates('user_id')
item_profile = data[["click_article_id"]].drop_duplicates('click_article_id')
user_index_2_rawid = dict(zip(user_profile['user_id'], user_profile_['user_id']))
item_index_2_rawid = dict(zip(item_profile['click_article_id'], item_profile_['click_article_id']))
# 划分训练和测试集
# 由于深度学习需要的数据量通常都是非常大的，所以为了保证召回的效果，往往会通过滑窗的形式扩充训练样本
train_set, test_set = gen_data_set(data, 0)
# 整理输入数据，具体的操作可以看上面的函数
train_model_input, train_label = gen_model_input(train_set, user_profile, SEQ_LEN)
test_model_input, test_label = gen_model_input(test_set, user_profile, SEQ_LEN)
# 确定Embedding的维度
embedding_dim = 16
# 将数据整理成模型可以直接输入的形式
user_feature_columns = [SparseFeat('user_id', feature_max_idx['user_id'], embedding_dim),
VarLenSparseFeat(SparseFeat('hist_article_id', feature_max_idx['click_article_id'], embedding_dim,
embedding_name="click_article_id"), SEQ_LEN, 'mean', 'hist_len'),]
item_feature_columns = [SparseFeat('click_article_id', feature_max_idx['click_article_id'], embedding_dim)]
# 模型的定义
# num_sampled: 负采样时的样本数量
model = YoutubeDNN(user_feature_columns, item_feature_columns, num_sampled=5, user_dnn_hidden_units=(64, embedding_dim))
# 模型编译
model.compile(optimizer="adam", loss=sampledsoftmaxloss)
# 模型训练，这里可以定义验证集的比例，如果设置为0的话就是全量数据直接进行训练
history = model.fit(train_model_input, train_label, batch_size=256, epochs=1, verbose=1, validation_split=0.0)
# 训练完模型之后,提取训练的Embedding，包括user端和item端
test_user_model_input = test_model_input
all_item_model_input = {"click_article_id": item_profile['click_article_id'].values}
user_embedding_model = Model(inputs=model.user_input, outputs=model.user_embedding)
item_embedding_model = Model(inputs=model.item_input, outputs=model.item_embedding)
# 保存当前的item_embedding 和 user_embedding 排序的时候可能能够用到，但是需要注意保存的时候需要和原始的id对应
user_embs = user_embedding_model.predict(test_user_model_input, batch_size=2 ** 12)
item_embs = item_embedding_model.predict(all_item_model_input, batch_size=2 ** 12)
# embedding保存之前归一化一下
user_embs = user_embs / np.linalg.norm(user_embs, axis=1, keepdims=True)
item_embs = item_embs / np.linalg.norm(item_embs, axis=1, keepdims=True)
# 将Embedding转换成字典的形式方便查询
raw_user_id_emb_dict = {user_index_2_rawid[k]: \
v for k, v in zip(user_profile['user_id'], user_embs)}
raw_item_id_emb_dict = {item_index_2_rawid[k]: \
v for k, v in zip(item_profile['click_article_id'], item_embs)}
# 将Embedding保存到本地
pickle.dump(raw_user_id_emb_dict, open(save_path + 'user_youtube_emb.pkl', 'wb'))
pickle.dump(raw_item_id_emb_dict, open(save_path + 'item_youtube_emb.pkl', 'wb'))
# faiss紧邻搜索，通过user_embedding 搜索与其相似性最高的topk个item
index = faiss.IndexFlatIP(embedding_dim)
# 上面已经进行了归一化，这里可以不进行归一化了
# faiss.normalize_L2(user_embs)
# faiss.normalize_L2(item_embs)
index.add(item_embs) # 将item向量构建索引
sim, idx = index.search(np.ascontiguousarray(user_embs), topk) # 通过user去查询最相似的topk个item
user_recall_items_dict = collections.defaultdict(dict)
for target_idx, sim_value_list, rele_idx_list in tqdm(zip(test_user_model_input['user_id'], sim, idx)):
target_raw_id = user_index_2_rawid[target_idx]
# 从1开始是为了去掉商品本身, 所以最终获得的相似商品只有topk-1
for rele_idx, sim_value in zip(rele_idx_list[1:], sim_value_list[1:]):
rele_raw_id = item_index_2_rawid[rele_idx]
user_recall_items_dict[target_raw_id][rele_raw_id] = user_recall_items_dict.get(target_raw_id, {})\
.get(rele_raw_id, 0) + sim_value
user_recall_items_dict = {k: sorted(v.items(), key=lambda x: x[1], reverse=True) for k, v in user_recall_items_dict.items()}
# 将召回的结果进行排序
# 保存召回的结果
# 这里是直接通过向量的方式得到了召回结果，相比于上面的召回方法，上面的只是得到了i2i及u2u的相似性矩阵，还需要进行协同过滤召回才能得到召回结果
# 可以直接对这个召回结果进行评估，为了方便可以统一写一个评估函数对所有的召回结果进行评估
pickle.dump(user_recall_items_dict, open(save_path + 'youtube_u2i_dict.pkl', 'wb'))
return user_recall_items_dict
2、基于文章的召回
和基于用户的协同过滤类似，只不过这时我们转向找到物品和物品之间的相似度，只有找到了目标用户对某些物品的评分，那么我们就可以对相似度高的类似物品进行预测，将评分最高的若干个相似物品推荐给用户。比如你在网上买了一本机器学习相关的书，网站马上会推荐一堆机器学习，大数据相关的书给你，这里就明显用到了基于项目的协同过滤思想。
# 基于商品的召回i2i
def item_based_recommend(user_id, user_item_time_dict, i2i_sim, sim_item_topk, recall_item_num, item_topk_click, item_created_time_dict, emb_i2i_sim):
"""
基于文章协同过滤的召回
:param user_id: 用户id
:param user_item_time_dict: 字典, 根据点击时间获取用户的点击文章序列 {user1: {item1: time1, item2: time2..}...}
:param i2i_sim: 字典，文章相似性矩阵
:param sim_item_topk: 整数， 选择与当前文章最相似的前k篇文章
:param recall_item_num: 整数， 最后的召回文章数量
:param item_topk_click: 列表，点击次数最多的文章列表，用户召回补全
:param emb_i2i_sim: 字典基于内容embedding算的文章相似矩阵
return: 召回的文章列表 {item1:score1, item2: score2...}
"""
# 获取用户历史交互的文章
user_hist_items = user_item_time_dict[user_id]
item_rank = {}
for loc, (i, click_time) in enumerate(user_hist_items):
for j, wij in sorted(i2i_sim[i].items(), key=lambda x: x[1], reverse=True)[:sim_item_topk]:
if j in user_hist_items:
continue
# 文章创建时间差权重
created_time_weight = np.exp(0.8 ** np.abs(item_created_time_dict[i] - item_created_time_dict[j]))
# 相似文章和历史点击文章序列中历史文章所在的位置权重
loc_weight = (0.9 ** (len(user_hist_items) - loc))
content_weight = 1.0
if emb_i2i_sim.get(i, {}).get(j, None) is not None:
content_weight += emb_i2i_sim[i][j]
if emb_i2i_sim.get(j, {}).get(i, None) is not None:
content_weight += emb_i2i_sim[j][i]
item_rank.setdefault(j, 0)
item_rank[j] += created_time_weight * loc_weight * content_weight * wij
# 不足10个，用热门商品补全
if len(item_rank) }).get(j, None) is not None:
content_weight += emb_i2i_sim[i][j]
if emb_i2i_sim.get(j, {}).get(i, None) is not None:
content_weight += emb_i2i_sim[j][i]
# 创建时间差权重
created_time_weight += np.exp(0.8 * np.abs(item_created_time_dict[i] - item_created_time_dict[j]))
items_rank[i] += loc_weight * content_weight * created_time_weight * wuv
# 热度补全
if len(items_rank) k: sorted(v, key=lambda x:x[1], reverse=True)[:recall_item_num] \
for k, v in cold_start_user_items_dict.items()}
pickle.dump(cold_start_user_items_dict, open(save_path + 'cold_start_user_items_dict.pkl', 'wb'))
return cold_start_user_items_dict
多路召回合并
将前面所有的召回策略得到的用户文章合并起来`
def combine_recall_results(user_multi_recall_dict, weight_dict=None, topk=25):
final_recall_items_dict = {}
# 对每一种召回结果按照用户进行归一化，方便后面多种召回结果，相同用户的物品之间权重相加
def norm_user_recall_items_sim(sorted_item_list):
# 如果冷启动中没有文章或者只有一篇文章，直接返回，出现这种情况的原因可能是冷启动召回的文章数量太少了，
# 基于规则筛选之后就没有文章了, 这里还可以做一些其他的策略性的筛选
if len(sorted_item_list) 0:
norm_score = 1.0 * (score - min_sim) / (max_sim - min_sim) if max_sim > min_sim else 1.0
else:
norm_score = 0.0
norm_sorted_item_list.append((item, norm_score))
return norm_sorted_item_list
print('多路召回合并...')
for method, user_recall_items in tqdm(user_multi_recall_dict.items()):
print(method + '...')
# 在计算最终召回结果的时候，也可以为每一种召回结果设置一个权重
if weight_dict == None:
recall_method_weight = 1
else:
recall_method_weight = weight_dict[method]
for user_id, sorted_item_list in user_recall_items.items(): # 进行归一化
user_recall_items[user_id] = norm_user_recall_items_sim(sorted_item_list)
for user_id, sorted_item_list in user_recall_items.items():
# print('user_id')
final_recall_items_dict.setdefault(user_id, {})
for item, score in sorted_item_list:
final_recall_items_dict[user_id].setdefault(item, 0)
final_recall_items_dict[user_id][item] += recall_method_weight * score
final_recall_items_dict_rank = {}
# 多路召回时也可以控制最终的召回数量
for user, recall_item_dict in final_recall_items_dict.items():
final_recall_items_dict_rank[user] = sorted(recall_item_dict.items(), key=lambda x: x[1], reverse=True)[:topk]
# 将多路召回后的最终结果字典保存到本地
pickle.dump(final_recall_items_dict, open(os.path.join(save_path, 'final_recall_items_dict.pkl'),'wb'))
return final_recall_items_dict_rank
|