new_org/fun-rec
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加打散策略

Browse Source
This commit is contained in:
RuyiLuo
2022-05-04 13:04:07 +08:00
parent e096d45dac
commit 4230b337b4
4 changed files with 350 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/README.md
+1 -2
View File
@@ -115,8 +115,7 @@
- [YoutubeDNN召回](ch03/ch3.2/3.2.4.2)
- [DSSM召回](ch03/ch3.2/3.2.4.3)
- [DeepFM排序模型](ch03/ch3.2/3.2.5)
- 规则与重排
- 任务监控与调度
- [重排(打散策略)](ch03/ch3.2/3.2.6)
- **当前问题汇总**
- [熟悉推荐系统基本流程问答整理](ch03/ch3.2/3.2.8.1)
- [数据库的基本使用问答整理](ch03/ch3.2/3.2.8.2)
docs/_sidebar.md
+18 -19
View File
@@ -2,9 +2,9 @@
- 目录
- 第一章 推荐系统概述
- [推荐系统的意义](/ch01/ch1.1)
- [推荐系统架构](ch01/ch1.2)
- [推荐系统技术栈](ch01/ch1.3)
- [1.1 推荐系统的意义](/ch01/ch1.1)
- [1.2 推荐系统架构](ch01/ch1.2)
- [1.3 推荐系统技术栈](ch01/ch1.3)
- 第二章 推荐系统算法基础
- 2.1 经典召回模型
- 2.1.1 基于协同过滤的召回
@@ -31,7 +31,7 @@
- 2.1.5 基于树模型的召回
- [TDM](ch02/ch2.1/ch2.1.5/TDM)
- 2.2 经典排序模型
- [GBDT+LR](ch02/ch2.2/ch2.2.1)
- [2.2.1 GBDT+LR](ch02/ch2.2/ch2.2.1)
- 2.2.2 特征交叉
- [FM](ch02/ch2.2/ch2.2.2/FM)
- [PNN](ch02/ch2.2/ch2.2.2/PNN)
@@ -55,13 +55,13 @@
- [PLE](ch02/ch2.2/ch2.2.5/PLE)
- 第三章 推荐系统实战
- 3.1 天池入门赛-新闻推荐
- [赛题理解&Baseline](ch03/ch3.1/markdown/ch3.1.1)
- [数据分析](ch03/ch3.1/markdown/ch3.1.2)
- [多路召回](ch03/ch3.1/markdown/ch3.1.3)
- [特征工程](ch03/ch3.1/markdown/ch3.1.4)
- [排序模型&模型融合](ch03/ch3.1/markdown/ch3.1.5)
- [3.1.1 赛题理解&Baseline](ch03/ch3.1/markdown/ch3.1.1)
- [3.1.2 数据分析](ch03/ch3.1/markdown/ch3.1.2)
- [3.1.3 多路召回](ch03/ch3.1/markdown/ch3.1.3)
- [3.1.4 特征工程](ch03/ch3.1/markdown/ch3.1.4)
- [3.1.5 排序模型&模型融合](ch03/ch3.1/markdown/ch3.1.5)
- 3.2 新闻推荐系统的实践
- [特别说明(必看)](ch03/ch3.2/3.2)
- [3.2.1 特别说明(必看)](ch03/ch3.2/3.2)
- 3.2.1 离线物料系统的构建
- [Mysql](ch03/ch3.2/3.2.1.1)
- [MongoDB](ch03/ch3.2/3.2.1.2)
@@ -72,21 +72,20 @@
- [前端基础及Vue实战](ch03/ch3.2/3.2.2.1)
- [flask简介及基础](ch03/ch3.2/3.2.2.2)
- [前后端交互](ch03/ch3.2/3.2.2.3)
- [推荐系统流程的构建](ch03/ch3.2/3.2.3)
- [3.2.3 推荐系统流程的构建](ch03/ch3.2/3.2.3)
- 3.2.4 召回
- [规则类召回](ch03/ch3.2/3.2.4.1)
- [YouTubeDNN召回](ch03/ch3.2/3.2.4.2)
- [DSSM召回](ch03/ch3.2/3.2.4.3)
- [DeepFM排序](ch03/ch3.2/3.2.5)
- 3.2.6 规则与重排
- 3.2.7 任务调度与监控
- [3.2.5 DeepFM排序](ch03/ch3.2/3.2.5)
- [3.2.6 重排(打散策略)](ch03/ch3.2/3.2.6)
- 3.2.8 当前问题汇总
- [熟悉推荐系统基本流程问答整理](ch03/ch3.2/3.2.8.1)
- [数据库的基本使用问答整理](ch03/ch3.2/3.2.8.2)
- [离线物料系统的构建问答整理](ch03/ch3.2/3.2.8.3)
- 第四章 推荐系统算法面经
- [ML与DL基础](ch04/ch4.1)
- [推荐模型相关](ch04/ch4.2)
- [热门技术相关](ch04/ch4.3)
- [业务场景相关](ch04/ch4.4)
- [HR及其他](ch04/ch4.5)
- [4.1 ML与DL基础](ch04/ch4.1)
- [4.2 推荐模型相关](ch04/ch4.2)
- [4.3 热门技术相关](ch04/ch4.3)
- [4.4 业务场景相关](ch04/ch4.4)
- [4.5 HR及其他](ch04/ch4.5)
docs/ch03/ch3.2/3.2.6.md
+330
View File
@@ -0,0 +1,330 @@
# 重排(打散策略）
在精排打完分之后，一般还会有个重排阶段，这个阶段里用户的展示是最近的，所以需要考虑的问题也很多，大致分为以下三类问题：
1. 用户体验：包括打散、多样性等
2. 算法效率：多任务融合、实时性等
3. 流量调控：流量扶持、生态建设等
这里主要是给出了在用户体验方面的打散策略，下面的代码实现是基于类别分桶打散的，通过不断的轮询每个类别的内容，达到内容上的打散。下面的代码打散策略还需要配合输入灌入redis中时，key的设置格式，这样便于后续打散的实现，详细的代码如下：
```python
import sys
sys.path.append("../../")
sys.path.append("../")
import json
import time
import threading
from conf.dao_config import cate_dict
from conf.proj_path import bad_case_news_log_path
from dao.redis_server import RedisServer
from dao.mysql_server import MysqlServer
from dao.entity.register_user import RegisterUser
from controller.user_action_controller import UserAction
from collections import defaultdict
redis_server = RedisServer()
class OnlineServer(object):
"""单例模式推荐服务类
"""
_instance_lock = threading.Lock()
def __init__(self,):
self.reclist_redis_db = redis_server.get_reclist_redis()
self.static_news_info_redis_db = redis_server.get_static_news_info_redis()
self.dynamic_news_info_redis_db = redis_server.get_dynamic_news_info_redis()
self.exposure_redis_db = redis_server.get_exposure_redis()
self.register_sql_sess = MysqlServer().get_register_user_session()
self.cate_dict = cate_dict
self.cate_id_list = list(self.cate_dict.keys())
self.bad_case_news_log_path = bad_case_news_log_path
self.name2id_cate_dict = {v: k for k, v in self.cate_dict.items()}
self._set_user_group()
def __new__(cls, *args, **kwargs):
if not hasattr(OnlineServer, "_instance"):
with OnlineServer._instance_lock:
if not hasattr(OnlineServer, "_instance"):
OnlineServer._instance = object.__new__(cls)
return OnlineServer._instance
def _get_register_user_cold_start_redis_key(self, userid):
"""通过查sql表得到用户的redis key进而确定当前新用户使用哪一个新的模板
"""
user_info = self.register_sql_sess.query(RegisterUser).filter(RegisterUser.userid == userid).first()
print(user_info)
if int(user_info.age) < 23 and user_info.gender == "female":
redis_key = "cold_start_group:{}".format(str(1))
elif int(user_info.age) >= 23 and user_info.gender == "female":
redis_key = "cold_start_group:{}".format(str(2))
elif int(user_info.age) < 23 and user_info.gender == "male":
redis_key = "cold_start_group:{}".format(str(3))
elif int(user_info.age) >= 23 and user_info.gender == "male":
redis_key = "cold_start_group:{}".format(str(4))
else:
pass
return redis_key
def _set_user_group(self):
"""将用户进行分组
1. age < 23 && gender == female
2. age >= 23 && gender == female
3. age < 23 && gender == male
4. age >= 23 && gender == male
"""
self.user_group = {
"1": ["国内","娱乐","体育","科技"],
"2": ["国内","社会","美股","财经","股市"],
"3": ["国内","股市","体育","科技"],
"4": ["国际", "国内","军事","社会","美股","财经","股市"]
}
self.group_to_cate_id_dict = defaultdict(list)
for k, cate_list in self.user_group.items():
for cate in cate_list:
self.group_to_cate_id_dict[k].append(self.name2id_cate_dict[cate])
def _get_register_user_group_id(self, age, gender):
"""获取注册用户的分组,
bug: 新用户注册可能会有延迟
"""
if int(age) < 23 and gender == "female":
return "1"
elif int(age) >= 23 and gender == "female":
return "2"
elif int(age) < 23 and gender == "male":
return "3"
elif int(age) >= 23 and gender == "male":
return "4"
else:
return "error"
def _copy_cold_start_list_to_redis(self, user_id, group_id):
"""将确定分组后的用户的物料添加到redis中，并记录当前用户的所有新闻类别id
"""
# 遍历当前分组的新闻类别
for cate_id in self.group_to_cate_id_dict[group_id]:
group_redis_key = "cold_start_group:{}:{}".format(group_id, cate_id)
user_redis_key = "cold_start_user:{}:{}".format(user_id, cate_id)
self.reclist_redis_db.zunionstore(user_redis_key, [group_redis_key])
# 将用户的类别集合添加到redis中
cate_id_set_redis_key = "cold_start_user_cate_set:{}".format(user_id)
self.reclist_redis_db.sadd(cate_id_set_redis_key, *self.group_to_cate_id_dict[group_id])
def _judge_and_get_user_reverse_index(self, user_id, rec_type, age=None, gender=None):
"""判断当前用户是否存在倒排索引, 如果没有的话拷贝一份
"""
if rec_type == 'hot_list':
# 判断用户是否存在热门列表
cate_id = self.cate_id_list[0] # 随机选择一个就行
hot_list_user_key = "user_id_hot_list:{}:{}".format(str(user_id), cate_id)
if self.reclist_redis_db.exists(hot_list_user_key) == 0:
# 给用户拷贝一份每个类别的倒排索引
for cate_id in self.cate_id_list:
cate_id_news_templete_key = "hot_list_news_cate:{}".format(cate_id)
hot_list_user_key = "user_id_hot_list:{}:{}".format(str(user_id), cate_id)
self.reclist_redis_db.zunionstore(hot_list_user_key, [cate_id_news_templete_key])
elif rec_type == "cold_start":
# 判断用户是否在冷启动列表中
cate_id_set_redis_key = "cold_start_user_cate_set:{}".format(user_id)
print("判断用户是否在冷启动列表中 {}".format(self.reclist_redis_db.exists(cate_id_set_redis_key)))
if self.reclist_redis_db.exists(cate_id_set_redis_key) == 0:
# 如果系统中没有当前用户的冷启动倒排索引, 那么就需要从冷启动模板中复制一份
# 确定用户分组
try:
group_id = self._get_register_user_group_id(age, gender)
except:
return False
print("group_id : {}".format(group_id))
self._copy_cold_start_list_to_redis(user_id, group_id)
else:
pass
return True
def _get_user_expose_set(self, user_id):
"""获取用户曝光列表
"""
user_exposure_prefix = "user_exposure:"
user_exposure_key = user_exposure_prefix + str(user_id)
# 获取用户当前曝光列表
if self.exposure_redis_db.exists(user_exposure_key) > 0:
exposure_list = self.exposure_redis_db.smembers(user_exposure_key)
news_expose_set = set(map(lambda x: x.split(':')[0], exposure_list))
else:
news_expose_set = set()
return news_expose_set
def _save_user_exposure(self, user_id, newslist):
"""记录用户曝光到redis"""
if len(newslist) == 0: return False # 无曝光数目
ctime = str(round(time.time()*1000)) # 曝光时间戳
key = "user_exposure:" + str(user_id) # 为key拼接
# 将历史曝光记录与newlist(最新曝光)的交集新闻提出来 并将该部分删除，防止重复存储曝光新闻
exposure_news_set = self.exposure_redis_db.smembers(key) # 历史曝光记录
del_exposure_news = [] # 历史曝光记录与newlist(最新曝光)的交集新闻,需要删除
if exposure_news_set.__len__() != 0:
del_exposure_news = [item for item in exposure_news_set if item.split(":")[0] in newslist]
# 为曝光过的新闻拼接时间
news_save = []
for news_id in newslist:
val = news_id+":"+ctime
val = val.replace('"', "'" ) # 将双引号都替换成单引号
news_save.append(val)
# 存储redis
try:
if del_exposure_news.__len__() != 0:
self.exposure_redis_db.srem(key,*del_exposure_news)
self.exposure_redis_db.sadd(key,*news_save)
except Exception as e:
print(str(e))
return False
return True
def _get_polling_rec_list(self, user_id, news_expose_set, cate_id_list, rec_type, one_page_news_cnt=10):
"""获取轮询的打散新闻列表
"""
# 候选曝光列表
exposure_news_list = []
# 用户展示新闻列表
user_news_list = []
iter_cnt = 0
# 给每个用户轮询每个类别的新闻，获取打散之后的新闻列表
while len(user_news_list) != one_page_news_cnt:
cate_id_index = iter_cnt % len(cate_id_list)
cate_id = cate_id_list[cate_id_index]
if rec_type == "hot_list":
user_redis_key = "user_id_hot_list:{}:{}".format(str(user_id), cate_id)
elif rec_type == "cold_start":
user_redis_key = "cold_start_user:{}:{}".format(str(user_id), cate_id)
else:
pass
cur_cate_cnt = 0
while self.reclist_redis_db.zcard(user_redis_key) > 0:
# 摘取排名第一的新闻
news_id_and_cate = self.reclist_redis_db.zrevrange(user_redis_key, 0, 0)[0]
news_id = news_id_and_cate.split('_')[1] # 将新闻id切分出来
if news_id in news_expose_set:
# 将当前新闻id添加到待删除的新闻列表中
self.reclist_redis_db.zrem(user_redis_key, news_id_and_cate)
continue
# TODO 在数据入库的时候离线处理无法成功加载json的问题
# 获取新闻详细信息, 由于爬取的新闻没有做清理，导致有些新闻无法转化成json的形式
# 所以这里如果转化失败的内容也直接删除掉
try:
news_info_dict = self.get_news_detail(news_id)
cur_cate_cnt += 1
except Exception as e:
# 删除无效的新闻
self.reclist_redis_db.zrem(user_redis_key, news_id_and_cate)
# 记录无效的新闻的id
with open(self.bad_case_news_log_path, "a+") as f:
f.write(news_id + "\n")
print("there are not news detail info for {}".format(news_id))
continue
# 删除当前key
self.reclist_redis_db.zrem(user_redis_key, news_id_and_cate)
# 判断当前类别的新闻是否摘取成功, 摘取成功的话就推出当前循环
if cur_cate_cnt == 1:
# 将摘取成功的新闻信息添加到用户新闻列表中
user_news_list.append(news_info_dict)
exposure_news_list.append(news_id)
break
iter_cnt += 1
return user_news_list, exposure_news_list
def get_cold_start_rec_list_v2(self, user_id, age=None, gender=None):
"""推荐页展示列表，使用轮询的方式进行打散
"""
# 获取用户曝光列表
news_expose_set = self._get_user_expose_set(user_id)
# 判断用户是否存在冷启动列表中
flag = self._judge_and_get_user_reverse_index(user_id, "cold_start", age, gender)
if not flag:
print("_judge_and_get_user_reverse_index fail")
return []
# 获取用户的cate id列表
cate_id_set_redis_key = "cold_start_user_cate_set:{}".format(user_id)
cate_id_list = list(self.reclist_redis_db.smembers(cate_id_set_redis_key))
# 通过轮询的方式
user_news_list, exposure_news_list = self._get_polling_rec_list(user_id, news_expose_set, cate_id_list, rec_type="cold_start")
# 添加曝光内容
self._save_user_exposure(user_id, exposure_news_list)
return user_news_list
def get_hot_list_v2(self, user_id):
"""热门页展示列表，使用轮询的方式进行打散
"""
# 获取用户曝光列表
news_expose_set = self._get_user_expose_set(user_id)
# 判断用户是否存在热门列表
self._judge_and_get_user_reverse_index(user_id, "hot_list")
# 通过轮询的方式获取用户的展示列表
user_news_list, exposure_news_list = self._get_polling_rec_list(user_id, news_expose_set, self.cate_id_list, rec_type="hot_list")
# 添加曝光内容
self._save_user_exposure(user_id, exposure_news_list)
return user_news_list
def get_news_detail(self, news_id):
"""获取新闻展示的详细信息
"""
news_info_str = self.static_news_info_redis_db.get("static_news_detail:" + news_id)
news_info_str = news_info_str.replace('\'', '\"' ) # 将单引号都替换成双引号
news_info_dit = json.loads(news_info_str)
news_dynamic_info_str = self.dynamic_news_info_redis_db.get("dynamic_news_detail:" + news_id)
news_dynamic_info_str = news_dynamic_info_str.replace("'", '"' ) # 将单引号都替换成双引号
news_dynamic_info_dit = json.loads(news_dynamic_info_str)
for k in news_dynamic_info_dit.keys():
news_info_dit[k] = news_dynamic_info_dit[k]
return news_info_dit
def update_news_dynamic_info(self, news_id,action_type):
"""更新新闻展示的详细信息
"""
news_dynamic_info_str = self.dynamic_news_info_redis_db.get("dynamic_news_detail:" + news_id)
news_dynamic_info_str = news_dynamic_info_str.replace("'", '"' ) # 将单引号都替换成双引号
news_dynamic_info_dict = json.loads(news_dynamic_info_str)
if len(action_type) == 2:
if action_type[1] == "true":
news_dynamic_info_dict[action_type[0]] +=1
elif action_type[1] == "false":
news_dynamic_info_dict[action_type[0]] -=1
else:
news_dynamic_info_dict["read_num"] +=1
news_dynamic_info_str = json.dumps(news_dynamic_info_dict)
news_dynamic_info_str = news_dynamic_info_str.replace('"', "'" )
res = self.dynamic_news_info_redis_db.set("dynamic_news_detail:" + news_id, news_dynamic_info_str)
return res
def test(self):
user_info = self.register_sql_sess.query(RegisterUser).filter(RegisterUser.userid == "4566566568405766145").first()
print(user_info.age)
if __name__ == "__main__":
# 测试单例模式
oneline_server = OnlineServer()
# oneline_server.get_hot_list("4563333734895456257")
oneline_server.test()
```
**参考资料**
[推荐算法架构4：重排](https://xieyangyi.blog.csdn.net/article/details/123095982)