在前面的文章中,我们实现了召回和排序,接下来将进入推荐逻辑处理阶段,通常称为推荐中心,推荐中心负责接收应用系统的推荐请求,读取召回和排序的结果并进行调整,最后返回给应用系统。推荐中心的调用流程如下所示:

推荐接口设计

通常推荐接口包括 Feed 流推荐和相似文章推荐

  • Feed 流推荐:根据用户偏好,获取推荐文章列表(这里的时间戳用于区分是刷新推荐列表还是查看历史推荐列表)
    参数:用户 ID,频道 ID,推荐文章数量,请求推荐的时间戳
    结果:曝光参数,每篇文章的行为埋点参数,上一条推荐的时间戳
  • 相似文章推荐:当用户浏览某文章时,获取该文章的相似文章列表
    参数:文章 ID,推荐文章数量
    结果:文章 ID 列表

行为埋点参数:

1
2
3
4
5
6
7
8
9
10
{
"param": '{"action": "exposure", "userId": 1, "articleId": [1,2,3,4], "algorithmCombine": "c1"}',
"recommends": [
{"article_id": 1, "param": {"click": "{"action": "click", "userId": "1", "articleId": 1, "algorithmCombine": 'c1'}", "collect": "...", "share": "...","read":"..."}},
{"article_id": 2, "param": {"click": "...", "collect": "...", "share": "...", "read":"..."}},
{"article_id": 3, "param": {"click": "...", "collect": "...", "share": "...", "read":"..."}},
{"article_id": 4, "param": {"click": "...", "collect": "...", "share": "...", "read":"..."}}
]
"timestamp": 1546391572
}

这里接口采用 gRPC 框架,在 user_reco.proto 文件中定义 Protobuf 序列化协议,其中定义了 Feed 流推荐接口:rpc user_recommend(User) returns (Track) {} 和相似文章接口:rpc article_recommend(Article) returns(Similar) {}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
syntax = "proto3";

message User {
string user_id = 1;
int32 channel_id = 2;
int32 article_num = 3;
int64 time_stamp = 4;
}
// int32 ---> int64 article_id
message Article {
int64 article_id = 1;
int32 article_num = 2;

}

message param2 {
string click = 1;
string collect = 2;
string share = 3;
string read = 4;
}

message param1 {
int64 article_id = 1;
param2 params = 2;
}

message Track {
string exposure = 1;
repeated param1 recommends = 2;
int64 time_stamp = 3;
}

message Similar {
repeated int64 article_id = 1;
}

service UserRecommend {
rpc user_recommend(User) returns (Track) {}
rpc article_recommend(Article) returns(Similar) {}
}

接着,通过如下命令生成服务端文件 user_reco_pb2.py 和客户端文件 user_reco_pb2_grpc.py

1
python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. user_reco.proto

定义参数解析类,用于解析推荐请求的参数,包括用户 ID、频道 ID、文章数量、请求时间戳以及算法名称

1
2
3
4
5
6
class Temp(object):
user_id = -10
channel_id = -10
article_num = -10
time_stamp = -10
algo = ""

定义封装埋点参数方法,其中参数 res 为推荐结果,参数 temp 为用户请求参数,将推荐结果封装为在 user_reco.proto 文件中定义的 Track 结构,其中携带了文章对埋点参数,包括了事件名称、算法名称以及时间等等,方便后面解析用户对文章对行为信息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def add_track(res, temp):
"""
封装埋点参数
:param res: 推荐文章id列表
:param temp: rpc参数
:return: 埋点参数
文章列表参数
单文章参数
"""
# 添加埋点参数
track = {}

# 准备曝光参数
# 全部字符串形式提供,在hive端不会解析问题
_exposure = {"action": "exposure", "userId": temp.user_id, "articleId": json.dumps(res),
"algorithmCombine": temp.algo}

track['param'] = json.dumps(_exposure)
track['recommends'] = []

# 准备其它点击参数
for _id in res:
# 构造字典
_dic = {}
_dic['article_id'] = _id
_dic['param'] = {}

# 准备click参数
_p = {"action": "click", "userId": temp.user_id, "articleId": str(_id),
"algorithmCombine": temp.algo}

_dic['param']['click'] = json.dumps(_p)
# 准备collect参数
_p["action"] = 'collect'
_dic['param']['collect'] = json.dumps(_p)
# 准备share参数
_p["action"] = 'share'
_dic['param']['share'] = json.dumps(_p)
# 准备detentionTime参数
_p["action"] = 'read'
_dic['param']['read'] = json.dumps(_p)

track['recommends'].append(_dic)

track['timestamp'] = temp.time_stamp
return track

AB Test 流量切分

由于推荐算法和策略是需要不断改进和完善等,所以 ABTest 也是推荐系统不可或缺的功能。可以根据用户 ID 将流量切分为多个桶(Bucket),每个桶对应一种排序策略,桶内流量将使用相应的策略进行排序,使用 ID 进行流量切分能够保证用户体验的一致性。通常 ABTest 过程如下所示:

通过定义 AB Test 参数,可以实现为不同的用户使用不同的推荐算法策略,其中 COMBINE 为融合方式,RECALL 为召回方式,SORT 为排序方式,CHANNEL 为频道数量,BYPASS 为分桶设置,sort_dict 为不同的排序服务对象。可以看到 Algo-1 使用 LR 进行排序,而 Algo-2 使用 Wide&Deep 进行排序

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
from collections import namedtuple

# ABTest参数信息
param = namedtuple('RecommendAlgorithm', ['COMBINE',
'RECALL',
'SORT',
'CHANNEL',
'BYPASS']
)

RAParam = param(
COMBINE={
'Algo-1': (1, [100, 101, 102, 103, 104], [200]), # 首页推荐,所有召回结果读取+LR排序
'Algo-2': (2, [100, 101, 102, 103, 104], [201]) # 首页推荐,所有召回结果读取 排序
},
RECALL={
100: ('cb_recall', 'als'), # 离线模型ALS召回,recall:user:1115629498121 column=als:18
101: ('cb_recall', 'content'), # 离线word2vec的画像内容召回 'recall:user:5', 'content:1'
102: ('cb_recall', 'online'), # 在线word2vec的画像召回 'recall:user:1', 'online:1'
103: 'new_article', # 新文章召回 redis当中 ch:18:new
104: 'popular_article', # 基于用户协同召回结果 ch:18:hot
105: ('article_similar', 'similar') # 文章相似推荐结果 '1' 'similar:2'
},
SORT={
200: 'LR',
201: 'WDL‘
},
CHANNEL=25,
BYPASS=[
{
"Bucket": ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd'],
"Strategy": "Algo-1"
},
{
"BeginBucket": ['e', 'f'],
"Strategy": "Algo-2"
}
]
)

sort_dict = {
"LR": lr_sort_service,
"WDL": wdl_sort_service
}

流量切分,将用户 ID 进行哈希,然后取哈希结果的第一个字符,将包含该字符的策略桶所对应的算法编号赋值到此用户请求参数的 algo 属性中,后面将调用该编号对应的算法策略为此用户计算推荐数据

1
2
3
4
5
6
7
8
9
import hashlib
from setting.default import DefaultConfig, RAParam

# 进行分桶实现分流,制定不同的实验策略
bucket = hashlib.md5(user_id.encode()).hexdigest()[:1]
if bucket in RAParam.BYPASS[0]['Bucket']:
temp.algo = RAParam.BYPASS[0]['Strategy']
else:
temp.algo = RAParam.BYPASS[1]['Strategy']

推荐中心逻辑

推荐中心逻辑主要包括:

  • 接收应用系统发送的推荐请求,解析请求参数
  • 进行 ABTest 分流,为用户分配推荐策略
  • 根据分配的算法调用召回服务和排序服务,读取推荐结果
  • 根据业务进行调整,如过滤、补足、合并信息等
  • 封装埋点参数,返回推荐结果

首先,在 Hbase 中创建历史推荐结果表 history_recommend,用于存储用户历史推荐结果

1
2
3
create 'history_recommend', {NAME=>'channel', TTL=>7776000, VERSIONS=>999999}   86400
# 每次指定一个时间戳,可以达到不同版本的效果
put 'history_recommend', 'reco:his:1', 'channel:18', [17283, 140357, 14668, 15182, 17999, 13648, 12884,18135]

继续在 Hbase 中创建待推荐结果表 wait_recommend,用于存储经过多路召回并且排序之后的待推荐结果,当 wait_recommend 没有数据时,才再次调用排序服务计算出新的待推荐结果并写入到 wait_recommend,所以不需设置多个版本。注意该表与 cb_recall 的区别,cb_recall 存储的是还未经排序的召回结果。

1
2
3
4
create 'wait_recommend', 'channel'

put 'wait_recommend', 'reco:1', 'channel:18', [17283, 140357, 14668, 15182, 17999, 13648, 12884,18135]
put 'wait_recommend', 'reco:1', 'channel:0', [17283, 140357, 14668, 15182, 17999, 13648, 12884, 17302, 13846]

用户获取 Feed 流推荐数据时,如果用户向下滑动,发出的是刷新推荐列表的请求,需要传入当前时间作为请求时间戳参数,该请求时间戳必然大于 Hbase 历史推荐结果表中的请求时间戳,那么程序将获取新的推荐列表,并返回 Hbase 历史推荐结果表中最近一次推荐的请求时间戳,用于查询历史推荐结果;如果用户向上滑动,发出的是查看历史推荐结果的请求,需要传入前面刷新推荐列表时返回的最近一次推荐的请求时间戳,该请求时间戳必然小于等于 Hbase 历史推荐结果中最近一次推荐的时间戳,那么程序将获取小于等于该请求时间戳的最近一次历史推荐结果,并返回小于该推荐结果最近一次推荐的时间戳,也就是上一次推荐的时间戳,下面是具体实现。

在获取推荐列表时,首先获取用户的历史数据库中最近一次时间戳 last_stamp,没有则将 last_stamp 置为 0

1
2
3
4
5
6
7
try:
last_stamp = self.hbu.get_table_row('history_recommend',
'reco:his:{}'.format(temp.user_id).encode(),
'channel:{}'.format(temp.channel_id).encode(),
include_timestamp=True)[1]
except Exception as e:
last_stamp = 0

  • 如果用户请求的时间戳小于历史推荐结果中最近一次请求的时间戳 last_stamp,那么该请求为用户获取历史推荐结果
    1.如果没有历史推荐结果,则返回时间戳 0 以及空列表 []
    2.如果历史推荐结果只有一条,则返回这一条历史推荐结果并返回时间戳 0,表示已经没有历史推荐结果(APP 可以显示已经没有历史推荐记录了)
    3.如果历史推荐结果有多条,则返回历史推荐结果中第一条推荐结果(最近一次),然后返回历史推荐结果中第二条推荐结果的时间戳
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
if temp.time_stamp < last_stamp:
try:
row = self.hbu.get_table_cells('history_recommend',
'reco:his:{}'.format(temp.user_id).encode(),
'channel:{}'.format(temp.channel_id).encode(),
timestamp=temp.time_stamp + 1,
include_timestamp=True)
except Exception as e:
row = []
res = []

if not row:
temp.time_stamp = 0
res = []
elif len(row) == 1 and row[0][1] == temp.time_stamp:
res = eval(row[0][0])
temp.time_stamp = 0
elif len(row) >= 2:
res = eval(row[0][0])
temp.time_stamp = int(row[1][1])

res = list(map(int, res))
# 封装推荐结果
track = add_track(res, temp)
# 曝光参数设置为空
track['param'] = ''

(注意:这里将用户请求的时间戳 +1,因为 Hbase 只能获取小于该时间戳的历史推荐结果)

  • 如果用户请求的时间戳大于 Hbase 历史推荐结果中最近一次请求的时间戳 last_stamp,那么该请求为用户刷新推荐列表,需要读取推荐结果并返回。如果结果为空,需要调用 user_reco_list() 方法,再次计算推荐结果,再返回。
    1
    2
    3
    4
    5
    6
    7
    8
    9
    if temp.time_stamp > last_stamp:
    # 获取缓存
    res = redis_cache.get_reco_from_cache(temp, self.hbu)
    # 如果结果为空,需要再次计算推荐结果 进行召回+排序,同时写入到hbase待推荐结果列表
    if not res:
    res = self.user_reco_list(temp)

    temp.time_stamp = int(last_stamp)
    track = add_track(res, temp)

定义 user_reco_list() 方法,首先要读取多路召回结果,根据为用户分配的算法策略,读取相应路径的召回结果,并进行重后合并

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
reco_set = []
# (1, [100, 101, 102, 103, 104], [])
for number in RAParam.COMBINE[temp.algo][1]:
if number == 103:
_res = self.recall_service.read_redis_new_article(temp.channel_id)
reco_set = list(set(reco_set).union(set(_res)))
elif number == 104:
_res = self.recall_service.read_redis_hot_article(temp.channel_id)
reco_set = list(set(reco_set).union(set(_res)))
else:
# 100, 101, 102召回结果读取
_res = self.recall_service.read_hbase_recall(RAParam.RECALL[number][0],
'recall:user:{}'.format(temp.user_id).encode(),
'{}:{}'.format(RAParam.RECALL[number][1],
temp.channel_id).encode())
reco_set = list(set(reco_set).union(set(_res)))

接着,过滤当前该请求频道的历史推荐结果,如果不是 0 频道还需过滤 0 频道的历史推荐结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
history_list = []
data = self.hbu.get_table_cells('history_recommend',
'reco:his:{}'.format(temp.user_id).encode(),
'channel:{}'.format(temp.channel_id).encode())

for _ in data:
history_list = list(set(history_list).union(set(eval(_))))

data = self.hbu.get_table_cells('history_recommend',
'reco:his:{}'.format(temp.user_id).encode(),
'channel:{}'.format(0).encode())

for _ in data:
history_list = list(set(history_list).union(set(eval(_))))

reco_set = list(set(reco_set).difference(set(history_list)))

最后,根据分配的算法策略,调用排序服务,将分数最高的 N 个推荐结果返回,并写入历史推荐结果表,如果还有剩余的排序结果,将其余写入待推荐结果表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# 使用指定模型对召回结果进行排序
# temp.user_id, reco_set
_sort_num = RAParam.COMBINE[temp.algo][2][0]
# 'LR'
reco_set = sort_dict[RAParam.SORT[_sort_num]](reco_set, temp, self.hbu)

if not reco_set:
return reco_set
else:

# 如果reco_set小于用户需要推荐的文章
if len(reco_set) <= temp.article_num:
res = reco_set
else:
# 大于要推荐的文章结果
res = reco_set[:temp.article_num]

# 将剩下的文章列表写入待推荐的结果
self.hbu.get_table_put('wait_recommend',
'reco:{}'.format(temp.user_id).encode(),
'channel:{}'.format(temp.channel_id).encode(),
str(reco_set[temp.article_num:]).encode(),
timestamp=temp.time_stamp)

# 直接写入历史记录当中,表示这次又成功推荐一次
self.hbu.get_table_put('history_recommend',
'reco:his:{}'.format(temp.user_id).encode(),
'channel:{}'.format(temp.channel_id).encode(),
str(res).encode(),
timestamp=temp.time_stamp)

return res

到这里,推荐中心的基本逻辑已经结束。下面是读取多路召回结果的实现细节:通过指定列族,读取基于模型、离线内容以及在线的召回结果,并删除 cb_recall 的召回结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def read_hbase_recall_data(self, table_name, key_format, column_format):
"""
读取cb_recall当中的推荐数据
读取的时候可以选择列族进行读取als, online, content

:return:
"""
recall_list = []
data = self.hbu.get_table_cells(table_name, key_format, column_format)
# data是多个版本的推荐结果[[],[],[],]
for _ in data:
recall_list = list(set(recall_list).union(set(eval(_))))
self.hbu.get_table_delete(table_name, key_format, column_format)
return recall_list

读取 redis 中的新文章

1
2
3
4
5
6
7
8
9
10
11
12
13
def read_redis_new_article(self, channel_id):
"""
读取新文章召回结果
:param channel_id: 提供频道
:return:
"""
_key = "ch:{}:new".format(channel_id)
try:
res = self.client.zrevrange(_key, 0, -1)
except Exception as e:
res = []

return list(map(int, res))

读取 redis 中的热门文章,并选取热度最高的前 K 个文章

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def read_redis_hot_article(self, channel_id):
"""
读取热门文章召回结果
:param channel_id: 提供频道
:return:
"""
_key = "ch:{}:hot".format(channel_id)
try:
res = self.client.zrevrange(_key, 0, -1)
except Exception as e:

# 由于每个频道的热门文章有很多,因为 保留文章点击次数
res = list(map(int, res))
if len(res) > self.hot_num:
res = res[:self.hot_num]
return res

读取相似文章

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def read_hbase_article_similar(self, table_name, key_format, article_num):
"""获取文章相似结果
:param article_id: 文章id
:param article_num: 文章数量
:return:
"""
try:
_dic = self.hbu.get_table_row(table_name, key_format)

res = []
_srt = sorted(_dic.items(), key=lambda obj: obj[1], reverse=True)
if len(_srt) > article_num:
_srt = _srt[:article_num]
for _ in _srt:
res.append(int(_[0].decode().split(':')[1]))
except Exception as e:
res = []
return res

使用缓存策略

  • 如果 redis 缓存中存在数据,就直接从 redis 缓存中获取推荐结果
  • 如果 redis 缓存为空而 Hbase 的待推荐结果表 wait_recommend 不为空,则从 wait_recommend 中获取推荐结果,并将一定数量的待推荐结果放入 redis 缓存中
  • 若 redis 和 wait_recommend 都为空,则需读取召回结果并进行排序,将排序结果写入 Hbase 的待推荐结果表 wait_recommend 中及 redis 中

(每次读取的推荐结果都要将其写入 Hbase 的历史推荐结果表 history_recommend 中)

读取 redis 缓存

1
2
3
4
5
6
7
8
9
10
#读取redis对应的键
key = 'reco:{}:{}:art'.format(temp.user_id, temp.channel_id)
# 读取,删除,返回结果
pl = cache_client.pipeline()

# 读取redis数据
res = cache_client.zrevrange(key, 0, temp.article_num - 1)
if res:
# 手动删除读取出来的缓存结果
pl.zrem(key, *res)

如果 redis 缓存为空

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
else:
# 删除键
cache_client.delete(key)
try:
# 从wait_recommend中读取
wait_cache = eval(hbu.get_table_row('wait_recommend',
'reco:{}'.format(temp.user_id).encode(),
'channel:{}'.format(temp.channel_id).encode()))
except Exception as e:
wait_cache = []

# 如果为空则直接返回空
if not wait_cache:
return wait_cache

# 如果wait_recommend中有数据
if len(wait_cache) > 100:
cache_redis = wait_cache[:100]

# 前100个数据放入redis
pl.zadd(key, dict(zip(cache_redis, range(len(cache_redis)))))

# 100个后面的数据,在放回wait_recommend
hbu.get_table_put('wait_recommend',
'reco:{}'.format(temp.user_id).encode(),
'channel:{}'.format(temp.channel_id).encode(),
str(wait_cache[100:]).encode())

else:
# 清空wait_recommend数据
hbu.get_table_put('wait_recommend',
'reco:{}'.format(temp.user_id).encode(),
'channel:{}'.format(temp.channel_id).encode(),
str([]).encode())

# 所有不足100个数据,放入redis
pl.zadd(key, dict(zip(wait_cache, range(len(wait_cache)))))

res = cache_client.zrange(key, 0, temp.article_num - 1)

最后,在 Supervisor 中配置 gRPC 实时推荐程序

1
2
3
4
5
6
7
8
9
10
11
12
[program:online]
environment=JAVA_HOME=/root/bigdata/jdk,SPARK_HOME=/root/bigdata/spark,HADOOP_HOME=/root/bigdata/hadoop,PYSPARK_PYTHON=/miniconda2/envs/reco_sys/bin/python ,PYSPARK_DRIVER_PYTHON=/miniconda2/envs/reco_sys/bin/python
command=/miniconda2/envs/reco_sys/bin/python /root/toutiao_project/reco_sys/abtest/routing.py
directory=/root/toutiao_project/reco_sys/abtest
user=root
autorestart=true
redirect_stderr=true
stdout_logfile=/root/logs/recommendsuper.log
loglevel=info
stopsignal=KILL
stopasgroup=true
killasgroup=true

想说的话

文章推荐系统系列到此就完结啦~ 撒花 🎉🎉🎉
若有疏漏的地方,欢迎各位多多指正,感谢关注,love & peace. 🙏🙏🙏

参考

https://www.bilibili.com/video/av68356229
https://pan.baidu.com/s/1-uvGJ-mEskjhtaial0Xmgw(学习资源已保存至网盘, 提取码:eakp)

【技术服务】,详情点击查看: https://mp.weixin.qq.com/s/PtX9ukKRBmazAWARprGIAg


扫一扫 关注微信公众号!号主 专注于搜索和推荐系统,尝试使用算法去更好的服务于用户,包括但不局限于机器学习,深度学习,强化学习,自然语言理解,知识图谱,还不定时分享技术,资料,思考等文章!