utils.py

import torch
import os
import random
import scipy.sparse as sp
import numpy as np
import torch.nn.functional as F
import torch.nn as nn
from parses import parse_args

def save_model(model, save_path, optimizer=None):
    os.makedirs(os.path.dirname(save_path), exist_ok=True)
    data2save = {
        'state_dict': model.state_dict(),
        'optimizer': optimizer.state_dict(),
    }
    torch.save(data2save, save_path)


def load_model(model, load_path, optimizer=None):
    data2load = torch.load(load_path, map_location='cpu')
    model.load_state_dict(data2load['state_dict'])
    if optimizer is not None and data2load['optimizer'] is not None:
        optimizer = data2load['optimizer']

def cosine_similarity(matrix1, matrix2) -> sp.csr_matrix:
    # 计算矩阵的点积
    dot_product = matrix1.dot(matrix2.T)

    # 计算矩阵的范数
    norm_product = np.multiply(np.sqrt(matrix1.power(2).sum(axis=1)), np.sqrt(matrix2.power(2).sum(axis=1)).T)

    # 计算余弦相似度
    similarity = dot_product / norm_product

    return similarity

args = parse_args()
device = args.device

class KDLoss(nn.Module):
    def __init__(self, kd_temp):
        super(KDLoss, self).__init__()
        self.kd_temp = kd_temp
        self.bce_loss = nn.BCELoss()
        
    def _generate_label(self, pos_preds, neg_preds):
        return torch.concat([torch.ones_like(pos_preds), torch.zeros_like(neg_preds)])

    def _knowledge_distill_loss(self, preds_a, preds_b):
        preds_a = preds_a / self.kd_temp
        preds_b = preds_b / self.kd_temp
        pos_entropy = torch.sigmoid(preds_b).detach() * torch.log(torch.sigmoid(preds_a) + 10e-6)
        neg_entropy = (1 - torch.sigmoid(preds_b).detach()) * torch.log(1 - torch.sigmoid(preds_a) - 10e-6)
        neg_entropy[np.isnan(neg_entropy.detach().cpu().numpy())] = 0.0
        distill_loss = -torch.mean(pos_entropy + neg_entropy)
        return distill_loss
    
    def forward(self, pos_preds_S, pos_preds_T, neg_preds_S):
        # L𝑆 (𝜃𝑆 ) = L𝐶𝐹 (𝜃𝑆 ) + 𝜆𝑇→𝑆 · L𝐵𝐷 (𝜃𝑆 ; 𝜃𝑇 )
        pre_labels = self._generate_label(pos_preds_S, neg_preds_S)
        pre_norm = torch.concat([torch.sigmoid(pos_preds_S), torch.sigmoid(neg_preds_S)])
        cf_bceloss = self.bce_loss(pre_norm, pre_labels)
        distill_loss = self._knowledge_distill_loss(pos_preds_S, pos_preds_T)
        kd_loss = 0.0 * cf_bceloss + 1.0 * distill_loss
        return kd_loss #*self.kd_reg
    

def innerProduct(usrEmbeds, itmEmbeds):
    return torch.sum(usrEmbeds * itmEmbeds, dim=-1)

def pairPredict(ancEmbeds, posEmbeds, negEmbeds):
    pos_preds = torch.mul(ancEmbeds, posEmbeds).sum(dim=1)
    neg_preds = torch.mul(ancEmbeds, negEmbeds).sum(dim=1)
    mf_loss = torch.mean(-torch.log(10e-6 + torch.sigmoid(pos_preds - neg_preds)))
    return mf_loss

def calcRegLoss(parameters):
    ret = 0
    for W in parameters:
        ret += W.norm(2).square()
    return ret

def calcReward(bprLossDiff, keepRate):
    _, posLocs = torch.topk(bprLossDiff, int(bprLossDiff.shape[0] * (1 - keepRate)))
    reward = torch.zeros_like(bprLossDiff).to(device)
    reward[posLocs] = 1.0
    return reward

def calcGradNorm(model):
    ret = 0
    for p in model.parameters():
        if p.grad is not None:
            ret += p.grad.data.norm(2).square()
    ret = (ret ** 0.5)
    ret.detach()
    return ret

def contrastLoss(embeds1, embeds2, nodes, temp):
    embeds1 = F.normalize(embeds1, p=2)
    embeds2 = F.normalize(embeds2, p=2)
    pckEmbeds1 = embeds1[nodes]
    pckEmbeds2 = embeds2[nodes]
    nume = torch.exp(torch.sum(pckEmbeds1 * pckEmbeds2, dim=-1) / temp)
    deno = torch.exp(pckEmbeds1 @ embeds2.T / temp).sum(-1)
    return -torch.log(nume / deno).mean()

def cal_cl_loss(temp, idx, user_view_1,user_view_2,item_view_1=None,item_view_2=None):
    if item_view_1 != None and item_view_2 != None:
        u_idx = torch.unique(torch.Tensor(idx[0]).type(torch.long)).to(device)
        i_idx = torch.unique(torch.Tensor(idx[1]).type(torch.long)).to(device)
        view1 = torch.cat((user_view_1[u_idx],item_view_1[i_idx]),0)
        view2 = torch.cat((user_view_2[u_idx],item_view_2[i_idx]),0)
    else:
        u_idx = torch.unique(torch.Tensor(idx[0]).type(torch.long)).to(device)
        v_idx = torch.unique(torch.Tensor(idx[1]).type(torch.long)).to(device)
        view1 = torch.cat((user_view_1[u_idx],user_view_1[v_idx]),0)
        view2 = torch.cat((user_view_2[u_idx],user_view_2[v_idx]),0)
    return InfoNCE(view1, view2, temp)

def cal_cross_cl_loss(temp, idx, ui_u_view1, ui_u_view2, uu_u_view1, uu_u_view2):
    cross_loss = 0.
    u_idx = torch.unique(torch.Tensor(idx[0]).type(torch.long)).to(device)
    v_idx = torch.unique(torch.Tensor(idx[1]).type(torch.long)).to(device)
    view1 = torch.cat((ui_u_view1[u_idx],ui_u_view1[v_idx]),0)
    view2 = torch.cat((uu_u_view1[u_idx],uu_u_view1[v_idx]),0)
    cross_loss += InfoNCE(view1, view2, temp)
    
    view1 = torch.cat((ui_u_view1[u_idx],ui_u_view1[v_idx]),0)
    view2 = torch.cat((uu_u_view2[u_idx],uu_u_view2[v_idx]),0)
    cross_loss += InfoNCE(view1, view2, temp)
    
    view1 = torch.cat((ui_u_view2[u_idx],ui_u_view2[v_idx]),0)
    view2 = torch.cat((uu_u_view1[u_idx],uu_u_view1[v_idx]),0)
    cross_loss += InfoNCE(view1, view2, temp)
    
    view1 = torch.cat((ui_u_view2[u_idx],ui_u_view2[v_idx]),0)
    view2 = torch.cat((uu_u_view2[u_idx],uu_u_view2[v_idx]),0)
    cross_loss += InfoNCE(view1, view2, temp)
    
    return cross_loss

def InfoNCE(view1, view2, temperature, b_cos = True):
    if b_cos:
        view1, view2 = F.normalize(view1, dim=1), F.normalize(view2, dim=1)
    pos_score = (view1 * view2).sum(dim=-1)
    pos_score = torch.exp(pos_score / temperature)
    ttl_score = torch.matmul(view1, view2.transpose(0, 1))
    ttl_score = torch.exp(ttl_score / temperature).sum(dim=1)
    cl_loss = -torch.log(pos_score / ttl_score+10e-6)
    return torch.mean(cl_loss)




def save_model(model, save_path, optimizer=None):
    os.makedirs(os.path.dirname(save_path), exist_ok=True)
    data2save = {
        'state_dict': model.state_dict(),
        'optimizer': optimizer.state_dict(),
    }
    torch.save(data2save, save_path)


def load_model(model, load_path, optimizer=None):
    data2load = torch.load(load_path, map_location='cpu')
    model.load_state_dict(data2load['state_dict'])
    if optimizer is not None and data2load['optimizer'] is not None:
        optimizer = data2load['optimizer']


def arrange_seed(seed):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False

    
def cosine_similarity(matrix1, matrix2) -> sp.csr_matrix:
    # 计算矩阵的点积
    dot_product = matrix1.dot(matrix2.T)

    # 计算矩阵的范数
    norm_product = np.multiply(np.sqrt(matrix1.power(2).sum(axis=1)), np.sqrt(matrix2.power(2).sum(axis=1)).T)

    # 计算余弦相似度
    similarity = dot_product / norm_product

    return similarity


if __name__ == "__main__":
    pass



if __name__ == "__main__":
    pass