构建AI智能体：告别“冷启动”：看大模型如何解决推荐系统的世纪难题

import numpy as np
 
def cosine_similarity_basic(vec_a, vec_b):
    """
    计算两个向量的余弦相似度
    
    参数:
    vec_a: 第一个向量
    vec_b: 第二个向量
    
    返回:
    余弦相似度值
    """
    # 转换为numpy数组
    a = np.array(vec_a)
    b = np.array(vec_b)
    
    # 计算点积
    dot_product = np.dot(a, b)
    
    # 计算模长
    norm_a = np.linalg.norm(a)
    norm_b = np.linalg.norm(b)
    
    # 计算余弦相似度
    if norm_a == 0 or norm_b == 0:
        return 0  # 避免除以零
    
    return dot_product / (norm_a * norm_b)
 
# 测试
vector1 = [3, 4]
vector2 = [1, 2]
similarity = cosine_similarity_basic(vector1, vector2)
print(f"向量{vector1}和{vector2}的余弦相似度: {similarity:.3f}")
 
# 用户评分示例
user_a = [5, 3, 0]
user_b = [4, 2, 5]
user_similarity = cosine_similarity_basic(user_a, user_b)
print(f"用户A{user_a}和用户B{user_b}的余弦相似度: {user_similarity:.3f}")

import numpy as np
from scipy.stats import pearsonr
 
# 用户评分数据
userA_ratings = [5, 4, 3, 2, 1]
userB_ratings = [4, 5, 2, 1, 3]
 
# 手动计算皮尔逊相关系数
def manual_pearson(x, y):
    # 计算平均值
    mean_x = np.mean(x)
    mean_y = np.mean(y)
    
    # 计算分子
    numerator = np.sum((np.array(x) - mean_x) * (np.array(y) - mean_y))
    
    # 计算分母
    denominator = np.sqrt(np.sum((np.array(x) - mean_x)**2)) * np.sqrt(np.sum((np.array(y) - mean_y)**2))
    
    return numerator / denominator
 
# 使用scipy计算
scipy_pearson, _ = pearsonr(userA_ratings, userB_ratings)
 
print(f"手动计算皮尔逊系数: {manual_pearson(userA_ratings, userB_ratings):.3f}")
print(f"Scipy计算皮尔逊系数: {scipy_pearson:.3f}")
 
# 计算余弦相似度
from sklearn.metrics.pairwise import cosine_similarity
cosine_sim = cosine_similarity([userA_ratings], [userB_ratings])[0][0]
print(f"余弦相似度: {cosine_sim:.3f}")

import numpy as np
import pandas as pd
 
def adjusted_cosine_similarity(item_i_ratings, item_j_ratings, user_means):
    """
    计算两个物品之间的调整余弦相似度
    
    参数:
    item_i_ratings: 物品i的评分列表
    item_j_ratings: 物品j的评分列表  
    user_means: 相应用户的平均评分列表
    
    返回:
    调整余弦相似度
    """
    # 转换为numpy数组
    ratings_i = np.array(item_i_ratings)
    ratings_j = np.array(item_j_ratings)
    means = np.array(user_means)
    
    # 只考虑两个物品都被评分的用户
    common_ratings = np.where((ratings_i > 0) & (ratings_j > 0))[0]
    
    if len(common_ratings) < 2:
        return 0  # 需要至少2个共同评分用户
    
    ratings_i_common = ratings_i[common_ratings]
    ratings_j_common = ratings_j[common_ratings]
    means_common = means[common_ratings]
    
    # 计算调整后的评分
    adjusted_i = ratings_i_common - means_common
    adjusted_j = ratings_j_common - means_common
    
    # 计算分子和分母
    numerator = np.sum(adjusted_i * adjusted_j)
    denominator = np.sqrt(np.sum(adjusted_i**2)) * np.sqrt(np.sum(adjusted_j**2))
    
    # 避免除以零
    if denominator == 0:
        return 0
    
    return numerator / denominator
 
# 示例使用
itemA_ratings = [5, 4, 3]  # 三个用户对物品A的评分
itemB_ratings = [3, 2, 4]  # 三个用户对物品B的评分
user_means = [3.33, 2.33, 4.00]  # 三个用户的平均评分
# 计算余弦相似度
from sklearn.metrics.pairwise import cosine_similarity
cosine_sim = cosine_similarity([itemA_ratings], [itemB_ratings])[0][0]
print(f"余弦相似度: {cosine_sim:.3f}")
adj_cosine_sim = adjusted_cosine_similarity(itemA_ratings, itemB_ratings, user_means)
print(f"调整余弦相似度: {adj_cosine_sim:.3f}")

def calculate_pearson_similarity_matrix(ratings_df):
    """
    计算所有用户之间的皮尔逊相关系数矩阵
    
    参数:
    ratings_df: 用户-物品评分DataFrame
    
    返回:
    皮尔逊相似度矩阵DataFrame
    """
    n_users = ratings_df.shape[1]  # 用户数量
    similarity_matrix = np.zeros((n_users, n_users))
    
    # 获取用户列表
    users = ratings_df.columns.tolist()
    
    # 计算每对用户之间的相似度
    for i in range(n_users):
        for j in range(i, n_users):  # 只需计算上三角矩阵
            if i == j:
                similarity_matrix[i, j] = 1.0  # 用户与自己的相似度为1
            else:
                user_i_ratings = ratings_df[users[i]].values
                user_j_ratings = ratings_df[users[j]].values
                similarity = pearson_similarity(user_i_ratings, user_j_ratings)
                similarity_matrix[i, j] = similarity
                similarity_matrix[j, i] = similarity  # 对称矩阵
    
    # 转换为DataFrame
    similarity_df = pd.DataFrame(
        similarity_matrix,
        index=users,
        columns=users
    )
    
    return similarity_df
 
# 创建示例数据
ratings_data = {
    '用户1': [5, 4, 0, 0, 3],
    '用户2': [4, 0, 5, 0, 2],
    '用户3': [0, 3, 4, 5, 0],
    '用户4': [2, 0, 0, 4, 5],
    '用户5': [0, 5, 3, 0, 4]
}
movies = ['电影A', '电影B', '电影C', '电影D', '电影E']
 
ratings_df = pd.DataFrame(ratings_data, index=movies)
 
# 计算皮尔逊相似度矩阵
pearson_sim_df = calculate_pearson_similarity_matrix(ratings_df)
print("皮尔逊相似度矩阵:")
print(pearson_sim_df)

def calculate_adjusted_cosine_matrix(ratings_df):
    """
    计算所有物品之间的调整余弦相似度矩阵
    
    参数:
    ratings_df: 用户-物品评分DataFrame（行:物品, 列:用户）
    
    返回:
    调整余弦相似度矩阵
    """
    n_items = ratings_df.shape[0]  # 物品数量
    items = ratings_df.index.tolist()
    
    # 计算每个用户的平均评分
    user_means = ratings_df.mean(axis=0)
    
    # 初始化相似度矩阵
    similarity_matrix = np.zeros((n_items, n_items))
    
    # 计算每对物品之间的相似度
    for i in range(n_items):
        for j in range(i, n_items):  # 只需计算上三角矩阵
            if i == j:
                similarity_matrix[i, j] = 1.0  # 物品与自身的相似度为1
            else:
                item_i_ratings = ratings_df.iloc[i].values
                item_j_ratings = ratings_df.iloc[j].values
                
                sim = adjusted_cosine_similarity(
                    item_i_ratings, 
                    item_j_ratings, 
                    user_means.values
                )
                similarity_matrix[i, j] = sim
                similarity_matrix[j, i] = sim  # 对称矩阵
    
    # 转换为DataFrame
    similarity_df = pd.DataFrame(
        similarity_matrix,
        index=items,
        columns=items
    )
    
    return similarity_df
 
# 创建示例数据
ratings_data = {
    '用户1': [5, 4, 0, 0, 3],
    '用户2': [4, 0, 5, 0, 2], 
    '用户3': [0, 3, 4, 5, 0],
    '用户4': [2, 0, 0, 4, 5],
    '用户5': [0, 5, 3, 0, 4]
}
movies = ['电影A', '电影B', '电影C', '电影D', '电影E']
 
ratings_df = pd.DataFrame(ratings_data, index=movies)
 
# 计算调整余弦相似度矩阵
adj_cosine_sim_df = calculate_adjusted_cosine_matrix(ratings_df)
print("调整余弦相似度矩阵:")
print(adj_cosine_sim_df)

# 创建示例数据：用户对电影的评分
# 行代表用户，列代表电影，数值代表评分（0表示未评分）
ratings_data = {
    '用户1': [5, 4, 3, 0, 0],
    '用户2': [4, 5, 0, 2, 1],
    '用户3': [3, 0, 4, 5, 2],
    '用户4': [0, 3, 5, 4, 3],
    '用户5': [2, 1, 0, 3, 4]
}
movies = ['电影A', '电影B', '电影C', '电影D', '电影E']
 
# 创建评分矩阵
ratings_df = pd.DataFrame(ratings_data, index=movies)
print("用户-电影评分矩阵:")
print(ratings_df)

class TextSimilarityCalculator:
    """基于大模型的文本相似度计算器"""
    
    def __init__(self, model_name='D:/modelscope/hub/models/sentence-transformers/all-MiniLM-L6-v2'):
        # 初始化代码...
    
    def encode_texts(self, texts):
        # 编码文本...
    
    def calculate_similarity(self, text1, text2):
        # 计算两个文本的相似度...
    
    def batch_similarity_matrix(self, texts):
        # 批量计算相似度矩阵...

class HybridRecommender:
    """结合传统协同过滤和大模型的混合推荐系统"""
    
    def __init__(self, ratings_df, text_model_name='D:/modelscope/hub/models/sentence-transformers/all-MiniLM-L6-v2'):
        # 初始化代码...
    
    def _precompute_item_embeddings(self):
        # 预计算物品嵌入...
    
    def calculate_semantic_similarity(self, item_id1, item_id2):
        # 计算语义相似度...
    
    def hybrid_similarity(self, item_id1, item_id2, alpha=0.7):
        # 计算混合相似度...
    
    def _calculate_cf_similarity(self, item_id1, item_id2):
        # 计算协同过滤相似度...
    
    def generate_recommendations(self, target_user_id, top_n=5, alpha=0.7):
        # 生成推荐...
    
    def _predict_rating(self, user_id, item_id, alpha):
        # 预测评分...

用户-电影评分矩阵:
     用户1  用户2  用户3  用户4  用户5
电影A    5    4    3    0    2
电影B    4    5    0    3    1
电影C    3    0    4    5    0
电影D    0    2    5    4    3
电影E    0    1    2    3    4
加载模型: D:/modelscope/hub/models/sentence-transformers/all-MiniLM-L6-v2
电影描述语义相似度矩阵:
[[1.0000004  0.78310543 0.8368864  0.7532436 ]
 [0.78310543 0.9999992  0.79229146 0.6913045 ]
 [0.8368864  0.79229146 0.9999998  0.70960337]
 [0.7532436  0.6913045  0.70960337 0.9999997 ]]
电影相似度: 0.791
加载模型: D:/modelscope/hub/models/sentence-transformers/all-MiniLM-L6-v2
推荐: 电影D, 预测评分: 2.10
推荐: 电影E, 预测评分: 1.95

import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import torch
import pandas as pd
from sklearn.metrics.pairwise import cosine_similarity
 
# 创建示例数据：用户对电影的评分
# 行代表用户，列代表电影，数值代表评分（0表示未评分）
ratings_data = {
    '用户1': [5, 4, 3, 0, 0],
    '用户2': [4, 5, 0, 2, 1],
    '用户3': [3, 0, 4, 5, 2],
    '用户4': [0, 3, 5, 4, 3],
    '用户5': [2, 1, 0, 3, 4]
}
movies = ['电影A', '电影B', '电影C', '电影D', '电影E']
 
# 创建评分矩阵
ratings_df = pd.DataFrame(ratings_data, index=movies)
print("用户-电影评分矩阵:")
print(ratings_df)
class TextSimilarityCalculator:
    """基于大模型的文本相似度计算器"""
    
    def __init__(self, model_name='D:/modelscope/hub/models/sentence-transformers/all-MiniLM-L6-v2'):
        """
        初始化文本相似度计算器
        
        参数:
        model_name: 预训练模型名称
            - 'all-MiniLM-L6-v2': 轻量级通用模型
            - 'all-mpnet-base-v2': 高质量通用模型  
            - 'multi-qa-mpnet-base-dot-v1': 问答优化模型
        """
        self.model = SentenceTransformer(model_name)
        print(f"加载模型: {model_name}")
    
    def encode_texts(self, texts):
        """
        将文本列表编码为向量表示
        
        参数:
        texts: 文本列表
        
        返回:
        文本向量矩阵
        """
        return self.model.encode(texts, convert_to_tensor=True)
    
    def calculate_similarity(self, text1, text2):
        """
        计算两个文本之间的相似度
        
        参数:
        text1: 第一个文本
        text2: 第二个文本
        
        返回:
        相似度分数 (0-1)
        """
        # 编码文本
        embeddings = self.encode_texts([text1, text2])
        
        # 计算余弦相似度
        similarity = cosine_similarity(
            embeddings[0].cpu().numpy().reshape(1, -1),
            embeddings[1].cpu().numpy().reshape(1, -1)
        )[0][0]
        
        return similarity
    
    def batch_similarity_matrix(self, texts):
        """
        计算文本集合的相似度矩阵
        
        参数:
        texts: 文本列表
        
        返回:
        相似度矩阵
        """
        embeddings = self.encode_texts(texts)
        similarity_matrix = cosine_similarity(embeddings.cpu().numpy())
        return similarity_matrix
 
# 使用示例
text_calculator = TextSimilarityCalculator()
 
# 电影描述示例
movie_descriptions = [
    "一部讲述黑客与反乌托邦系统斗争的赛博朋克惊悚片",
    "一部设定在未来世界、科技先进的科幻冒险片",
    "一部讲述两人在纽约坠入爱河的浪漫喜剧",
    "一部以二战为背景的历史剧"
]
 
# 计算相似度矩阵
similarity_matrix = text_calculator.batch_similarity_matrix(movie_descriptions)
print("电影描述语义相似度矩阵:")
print(similarity_matrix)
 
# 计算特定文本对相似度
text1 = "一部关于人工智能和机器人的电影"
text2 = "一部探索人类与机器之间关系的电影"
similarity = text_calculator.calculate_similarity(text1, text2)
print(f"\n电影相似度: {similarity:.3f}")
print("=" * 50)
print("混合推荐系统")
print("=" * 50)
class HybridRecommender:
    """结合传统协同过滤和大模型的混合推荐系统"""
    
    def __init__(self, ratings_df, text_model_name='D:/modelscope/hub/models/sentence-transformers/all-MiniLM-L6-v2'):
        """
        初始化混合推荐系统
        
        参数:
        ratings_df: 用户-物品评分DataFrame
        text_model_name: 文本编码模型名称
        """
        self.ratings_df = ratings_df
        self.text_calculator = TextSimilarityCalculator(text_model_name)
        
        # 预计算物品文本特征
        self.item_descriptions = {}  # 假设有物品描述数据
        self.item_embeddings = self._precompute_item_embeddings()
        
    def _precompute_item_embeddings(self):
        """预计算所有物品的文本嵌入"""
        item_ids = self.ratings_df.index.tolist()
        descriptions = [self.item_descriptions.get(item_id, "") for item_id in item_ids]
        return self.text_calculator.encode_texts(descriptions)
    
    def calculate_semantic_similarity(self, item_id1, item_id2):
        """计算两个物品的语义相似度"""
        idx1 = self.ratings_df.index.get_loc(item_id1)
        idx2 = self.ratings_df.index.get_loc(item_id2)
        
        emb1 = self.item_embeddings[idx1].cpu().numpy()
        emb2 = self.item_embeddings[idx2].cpu().numpy()
        
        return cosine_similarity([emb1], [emb2])[0][0]
    
    def hybrid_similarity(self, item_id1, item_id2, alpha=0.7):
        """
        计算混合相似度（协同过滤 + 语义相似度）
        
        参数:
        item_id1: 物品1ID
        item_id2: 物品2ID  
        alpha: 协同过滤权重 (1-alpha为语义相似度权重)
        """
        # 计算传统协同过滤相似度（基于评分）
        cf_similarity = self._calculate_cf_similarity(item_id1, item_id2)
        
        # 计算语义相似度
        semantic_similarity = self.calculate_semantic_similarity(item_id1, item_id2)
        
        # 混合相似度
        hybrid_sim = alpha * cf_similarity + (1 - alpha) * semantic_similarity
        return hybrid_sim
    
    def _calculate_cf_similarity(self, item_id1, item_id2):
        """计算基于评分的协同过滤相似度"""
        ratings1 = self.ratings_df.loc[item_id1].values
        ratings2 = self.ratings_df.loc[item_id2].values
        
        # 只考虑共同评分的用户
        common_ratings = np.where((ratings1 > 0) & (ratings2 > 0))[0]
        
        if len(common_ratings) < 2:
            return 0
        
        # 使用皮尔逊相关系数
        from scipy.stats import pearsonr
        corr, _ = pearsonr(ratings1[common_ratings], ratings2[common_ratings])
        return 0 if np.isnan(corr) else corr
    
    def generate_recommendations(self, target_user_id, top_n=5, alpha=0.7):
        """生成混合推荐"""
        # 获取目标用户未评分的物品
        user_ratings = self.ratings_df[target_user_id]
        unrated_items = user_ratings[user_ratings == 0].index
        
        predictions = []
        for item_id in unrated_items:
            # 计算预测评分
            predicted_rating = self._predict_rating(target_user_id, item_id, alpha)
            predictions.append((item_id, predicted_rating))
        
        # 返回topN推荐
        predictions.sort(key=lambda x: x[1], reverse=True)
        return predictions[:top_n]
    
    def _predict_rating(self, user_id, item_id, alpha):
        """预测用户对物品的评分"""
        # 基于相似物品的评分进行预测
        user_mean = self.ratings_df[user_id].mean()
        
        numerator = 0
        denominator = 0
        
        # 找到用户已评分的物品
        rated_items = self.ratings_df[user_id][self.ratings_df[user_id] > 0].index
        
        for rated_item in rated_items:
            similarity = self.hybrid_similarity(item_id, rated_item, alpha)
            rating = self.ratings_df.loc[rated_item, user_id]
            item_mean = self.ratings_df.loc[rated_item].mean()
            
            numerator += similarity * (rating - item_mean)
            denominator += abs(similarity)
        
        if denominator == 0:
            return user_mean
        
        predicted_rating = user_mean + (numerator / denominator)
        return max(1, min(5, predicted_rating))  # 确保评分在1-5范围内
 
# 使用示例
# 假设已有ratings_df和物品描述
hybrid_rec = HybridRecommender(ratings_df)
recommendations = hybrid_rec.generate_recommendations('用户1')
for item_id, rating in recommendations:
    print(f"推荐: {item_id}, 预测评分: {rating:.2f}")