在(1)中介绍了FM模型的理论和python实现二分类模型。作为用于CTR预估的模型之一，FM重点在于实现ctr。
一、数据集
电商数据中的用户行为日志数据。召回完成，在排序阶段，需要考虑用户特征和物品特征，用户特征来源于用户画像，物品特征来源于物品自身固有属性；用户画像一部分是通过物品画像得到。
1、物品画像
在电商领域，以脐橙为例，物品画像通常包含如下维度：
1、关键词：商品标题和详情页的文字部分提取关键词(topN)，其数据格式为 keyword = [(光滑,0.32),(饱满,0.34),...]，
2、实体词：商品标题和详情页的文字部分提起实体词(topN)，其数据格式为：entity=["云南省","赣南",...]，
3、价格：商品本身的价格是一个连续型特征，进行分桶处理为类别特征,如果划分10个区间，则商品的价格为10个特征之一：price=[0,0,1,0,...]，
4、分类：脐橙本身又分为了多个种类，每个脐橙属于一个分类。如果区分大分类和小分类，则可得到两个分类特征。分类为类别特征：category=[0,1,0,...]，
5、产地: 商品产地为类别特征：area=[0,1,0,...]，
如果商品有和时间或者节气、节日等强相关的特征，可以将其加入物品画像。
2、用户画像
很明显，用户画像是基于物品画像的。用户购买、收藏、点击等行为日志通过ios端或者android客户端埋点获得，过滤清洗之后存入hdfs,供后续推荐算法使用。
1、关于日志
问题1：需要考虑用户哪些行为是有价值的。很明显，用户购买、收藏了某个商品，他是喜欢这个商品的。那么构建用户画像只使用这类日志可以么？答案是No，因为对于一个日活有限的电商平台而言，这类日志很少。一般要加上浏览日志。
问题2：用户浏览、购买、收藏了一个商品，产生了2+N(代表多次浏览该商品)条日志，日志如何处理。一个方案，只保留购买行为的日志，因为购买和用户喜欢是最强相关的。
问题3：日志时间。如果喜欢浏览了商品A2秒，商品B20秒，如何确定其喜好。设定阈值，用户点进去一个商品然后快速返回，不能表示其喜好该商品。
2、用户画像字段
这里，用户画像直接使用物品画像进行构建，放弃了用户自身属性(年龄、性别等)。因为这些属性大部分用户都为空。实际场景中，很多属性是app不能获得的。
1、关键词：用户操作过的商品的关键词，按照权重加权求和。数据格式为 keyword = [(光滑,1.32),(饱满,2.34),...]，
2、实体词：用户操作过的商品的实体词，按照实体词总数取分数。数据格式为：entity=[("云南省,0.1"),("赣南",0.2),...]，
3、价格：用户的价格画像为每个价格区间的比例：price=[("0-100",0.2),("100-500",0.4),...]，
4、分类：用户分类为操作过的商品的每个分类的比例：category=[(1,0.2),(2,0.1),...]，
5、产地: 用户操作过的商品的产地的分布：area=[("云南省",0.1),("安徽省",0.3),...]。
3、模型输入向量生成
用户向量+物品向量
假设有2000个关键词，1000个实体词，10种价格区间，10个分类，50个产地，则最终的用户向量维度为：(2000+1000+10+10+50=3070)，物品向量维度为3070。
ps:这里没有使用物品之外的特征，比如时间信息、app相关信息，行为信息等数据。
故，最终的模型输入特征向量:input_vector = np.zeros(6140+1,dtype=np.float)，然后在对应特征位置赋值。
生成的numpy数组保存为xxx.npy
二、torch实现FM
用于CTR时，模型输出为sigmoid之后的概率值:[0,1]。
分为几个模块
1、数据集加载
import torch
from torch.utils.data import Dataset
import numpy as np
from dataprocess import DataLoad # 自定义的npy数据读取类
class CtrDataset(Dataset):
"""
Custom dataset class for dataset in order to use efficient
dataloader tool provided by PyTorch.
"""
def __init__(self, train=True,split_=0.8):
"""
Initialize file path and train/test mode.
Inputs:
- train: bool.是否为训练阶段
- split_: 训练数据比例。
"""
self.train = train
train_data,test_data = DataLoad().split_sample(split_)
if self.train:
self.train_x = train_data[:, :-1]
self.train_y = train_data[:, -1]
else:
self.test_x = test_data[:,:-1]
self.test_y = test_data[:,-1]
def __getitem__(self, idx):
'''
self.train_data的值：[[0,1,...],[],...]，y要修改为：[[1],[0],...]的格式。
'''
if self.train:
dataI, targetI = self.train_x[idx, :], self.train_y[idx]
targetI = np.array(targetI)
targetI = torch.from_numpy(targetI)
targetI = torch.unsqueeze(targetI,-1)
return dataI,targetI
else:
dataI, targetI = self.test_x[idx, :], self.test_y[idx]
targetI = np.array(targetI)
targetI = torch.from_numpy(targetI)
targetI = torch.unsqueeze(targetI, -1)
return dataI, targetI
def __len__(self):
if self.train:
return len(self.train_x)
else:
return len(self.test_x)
2、 DataLoader加载数据
train_data = CtrDataset( train=True,split_=split_)
test_data = CtrDataset( train=True,split_=split_)
loader_train = DataLoader(train_data, batch_size=50,
shuffle=True)
常用操作有：batch_size(每个batch的大小), shuffle(是否进行shuffle操作), num_workers(加载数据的时候使用几个子进程)。
3、选择使用CPU还是GPU进行训练
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
4、FM模型
class FMLayer(nn.Module):
def __init__(self, n=10, k=5):
"""
:param n: 特征维度
:param k: 隐向量维度
"""
super(FMLayer, self).__init__()
self.dtype = torch.float
self.n = n
self.k = k
self.linear = nn.Linear(self.n, 1) # 前两项线性层
'''
torch.nn.Parameter是继承自torch.Tensor的子类，其主要作用是作为nn.Module中的可训练参数使用。它与torch.Tensor的区别就是nn.Parameter会自动被认为是module的可训练参数，即加入到parameter()这个迭代器中去；而module中非nn.Parameter()的普通tensor是不在parameter中的。
注意到，nn.Parameter的对象的requires_grad属性的默认值是True，即是可被训练的，这与torth.Tensor对象的默认值相反。
在nn.Module类中，pytorch也是使用nn.Parameter来对每一个module的参数进行初始化的。'''
self.v = nn.Parameter(torch.randn(self.n, self.k)) # 交互矩阵
nn.init.uniform_(self.v, -0.1, 0.1)
def fm_layer(self, x):
# x 属于 R^{batch*n}
linear_part = self.linear(x)
#print("linear_part",linear_part.shape)
# linear_part = torch.unsqueeze(linear_part, 1)
# print(linear_part.shape)
# 矩阵相乘 (batch*p) * (p*k)
inter_part1 = torch.mm(x, self.v) # out_size = (batch, k) # 矩阵a和b矩阵相乘。 vi,f * xi
# 矩阵相乘 (batch*p)^2 * (p*k)^2
inter_part2 = torch.mm(torch.pow(x, 2), torch.pow(self.v, 2)) # out_size = (batch, k)
# 这里torch求和一定要用sum
inter = 0.5 * torch.sum(torch.sub(torch.pow(inter_part1, 2), inter_part2),1,keepdim=True)
#print("inter",inter.shape)
output = linear_part + inter
output = torch.sigmoid(output)
#print(output.shape) # out_size = (batch, 1)
return output
def forward(self, x):
return self.fm_layer(x)
上述为FM公式的torch版本。作为网络模型，还需要定义损失函数和训练过程。
模型输出已经是经过sigmoid的概率值，直接使用交叉熵作为损失函数。
def fit(self,data,optimizer,epochs=100):
"""
Training a model and valid accuracy.
Inputs:
- loader_train: I
- optimizer: Abstraction of optimizer used in training process, e.g., "torch.optim.Adam()""torch.optim.SGD()".
- epochs: Integer, number of epochs.
"""
criterion = F.binary_cross_entropy
for epoch in range(epochs):
for t, (batch_x, batch_y) in enumerate(data):
batch_x = batch_x.to(device)
batch_y = batch_y.to(device)
total = self.forward(batch_x)
loss = criterion(total, batch_y)
optimizer.zero_grad()
loss.backward()
optimizer.step()
loader_test = DataLoader(test_data, batch_size=50,
shuffle=True)
r = self.test(loader_test)
print('Epoch %d , loss = %.4f' % (epoch, r))
def test(self,data):
'''
测试集测试
:return:
'''
criterion = F.binary_cross_entropy
all_loss = 0
i = 0
for t, (batch_x, batch_y) in enumerate(data):
batch_x = batch_x.to(device)
batch_y = batch_y.to(device)
total = self.forward(batch_x)
loss = criterion(total, batch_y)
all_loss += loss.item()
i += 1
return all_loss/i
三、模型训练和保存
使用flask作为web服务框架。
为了线上部署，使用torchscript进行模型的保存。
https://pytorch.org/tutorials/beginner/Intro_to_TorchScript_tutorial.html
https://discuss.pytorch.org/t/infer-torch-model-via-gunicorn-wsgi/60437
fm = FMLayer(n=features,k=30)
fm = fm.to(device)
optimizer = optim.Adam(fm.parameters(), lr=1e-4, weight_decay=0.0)
fm.fit(loader_train, optimizer, epochs=100)
fm = fm.to("cpu")
temp = torch.zeros((1,6140))
traced_model = torch.jit.trace(fm,temp)
torch.jit.save(traced_model, 'model.pt')
使用torch.save(model,path)进行保存的模型，在加载的使用，要求可以找到原始的FMLayer类，直接xx.py没有问题。但是，如果web服务使用gunicorn进行启动，就会报错：
AttributerError:Can't get attribute 'FMLayer' on <module '__main__' from '/usr/local/bin/gunicorn'
因为：torch.load(model_path)的时候，需要在当前位置有模型类。而使用gunicorn的时候，它会在gunicorn那里寻找模型类。
使用torch.jit.load(model_path, map_location='cpu')可以不用在当前位置有对应的模型类。
四、线上部署
对外提供api接口，接收输入数据:用户id和召回算法得到的物品id,返回排序后的物品id列表。
使用docker部署注意事项：
1、完整的requirements.txt
2、gunicorn 的配置 daemon = "false"
3、时区改变：RUN cp /usr/share/zoneinfo/Asia/Shanghai /etc/localtime && echo 'Asia/Shanghai' >/etc/timezone
4、主机和容器数据同步，日志和新的模型文件.
volumes: #映射的数据卷
- ./app:/www/web
- ./nginx/conf:/etc/nginx
- ./nginx/logs:/www/web_logs
五、模型更新和线上服务更新
使用每天的日志训练模型并实时更新线上模型，通过flask_apscheduler模块在web服务中执行定时任务。