大模型应用：从意图识别到个性化推荐：基于语义意图识别的智能推荐系统.121

# 基础语义向量生成与相似度计算
from sentence_transformers import SentenceTransformer
from modelscope import snapshot_download
import torch
import torch.nn.functional as F
import numpy as np

# 配置模型缓存路径（请修改为你的本地路径）
CACHE_DIR = "D:\\modelscope\\hub"

def load_semantic_model():
    """
    加载预训练的语义模型
    返回：初始化后的Sentence-BERT模型
    """
    # 从ModelScope下载模型
    model_dir = snapshot_download(
        model_id="sentence-transformers/all-MiniLM-L6-v2",
        cache_dir=CACHE_DIR,
        revision="master"
    )
    
    # 初始化模型
    model = SentenceTransformer(model_dir)
    print(f"模型加载完成，路径：{model_dir}")
    return model

def get_text_embedding(model, text: str) -> np.ndarray:
    """
    将文本转化为语义向量
    :param model: 已加载的Sentence-BERT模型
    :param text: 输入文本
    :return: 768维的语义向量（numpy数组）
    """
    # 空文本处理
    if not text.strip():
        return np.zeros(768)
    
    # 生成向量（convert_to_tensor=True返回PyTorch张量，便于后续计算）
    embedding = model.encode(
        text,
        convert_to_tensor=True,
        normalize_embeddings=True  # 归一化，便于计算余弦相似度
    )
    
    # 转换为numpy数组，方便查看和保存
    return embedding.cpu().numpy()

def calculate_cosine_similarity(embedding1: np.ndarray, embedding2: np.ndarray) -> float:
    """
    计算两个向量的余弦相似度
    :param embedding1: 第一个向量
    :param embedding2: 第二个向量
    :return: 相似度值（0-1之间）
    """
    # 转换为PyTorch张量
    tensor1 = torch.tensor(embedding1)
    tensor2 = torch.tensor(embedding2)
    
    # 计算余弦相似度
    similarity = F.cosine_similarity(tensor1.unsqueeze(0), tensor2.unsqueeze(0))
    
    return similarity.item()

# 测试基础功能
if __name__ == "__main__":
    # 加载模型
    model = load_semantic_model()
    
    # 测试文本
    text1 = "我想买华为手机"
    text2 = "华为智能手机新品推荐"
    text3 = "苹果笔记本电脑价格"
    
    # 生成向量
    emb1 = get_text_embedding(model, text1)
    emb2 = get_text_embedding(model, text2)
    emb3 = get_text_embedding(model, text3)
    
    # 打印向量形状（all-MiniLM-L6-v2生成768维向量）
    print(f"\n文本1向量形状：{emb1.shape}")
    print(f"文本1向量前5个值：{emb1[:5]}")
    
    # 计算相似度
    sim1_2 = calculate_cosine_similarity(emb1, emb2)
    sim1_3 = calculate_cosine_similarity(emb1, emb3)
    
    print(f"\n文本1与文本2的相似度：{sim1_2:.4f}")  
    print(f"文本1与文本3的相似度：{sim1_3:.4f}")  

# 完整的基于语义意图识别的智能推荐系统
from sentence_transformers import SentenceTransformer
from modelscope import snapshot_download
import torch
import torch.nn.functional as F
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from typing import List, Dict, Tuple

# 全局配置
CACHE_DIR = "D:\\modelscope\\hub"
MODEL_ID = "sentence-transformers/all-MiniLM-L6-v2"
TOP_N_RECOMMEND = 5  # 默认返回Top5推荐结果
plt.rcParams["font.sans-serif"] = ["SimHei"]  # 支持中文显示
plt.rcParams["axes.unicode_minus"] = False    # 支持负号显示

class SemanticRecommender:
    """基于语义意图识别的推荐系统类"""
    
    def __init__(self):
        """初始化推荐系统：加载模型、初始化商品库"""
        self.model = self._load_model()
        self.product_db = self._init_product_database()
        self.product_embeddings = self._precompute_product_embeddings()
    
    def _load_model(self) -> SentenceTransformer:
        """加载语义模型（内部方法）"""
        print("正在下载/加载语义模型...")
        model_dir = snapshot_download(
            model_id=MODEL_ID,
            cache_dir=CACHE_DIR,
            revision="master"
        )
        model = SentenceTransformer(model_dir)
        # 将模型移动到GPU（如果可用）
        device = "cuda" if torch.cuda.is_available() else "cpu"
        model = model.to(device)
        print(f"模型加载完成：{MODEL_ID}（设备：{device}）")
        return model
    
    def _init_product_database(self) -> pd.DataFrame:
        """初始化商品数据库（模拟电商商品数据）"""
        products = [
            {"product_id": 1001, "name": "华为MateBook 14", "description": "14英寸轻薄笔记本电脑，酷睿i5，16G内存，512G固态，续航长，便携办公，商务笔记本"},
            {"product_id": 1002, "name": "苹果iPhone 15", "description": "iPhone 15 256G 黑色，新款智能手机，A16芯片，超清摄像头，长续航，苹果手机"},
            {"product_id": 1003, "name": "小米平板6", "description": "11.2英寸平板电脑，骁龙8+，8G+256G，学习办公娱乐，平板设备"},
            {"product_id": 1004, "name": "戴尔XPS 13", "description": "13.4英寸超极本，酷睿i7，32G内存，1TB固态，轻薄便携，高端办公，商务笔记本"},
            {"product_id": 1005, "name": "华为Pura 70 Pro", "description": "华为新款智能手机，麒麟芯片，超光变摄像头，512G存储，鸿蒙系统，安卓手机"},
            {"product_id": 1006, "name": "联想拯救者Y9000P", "description": "16英寸游戏笔记本电脑，酷睿i9，RTX4060，32G内存，1TB固态，高性能，电竞游戏本，游戏电脑"},
            {"product_id": 1007, "name": "iPad Pro 2024", "description": "12.9英寸iPad Pro，M4芯片，256G，支持Apple Pencil，专业创作，平板电脑"},
            {"product_id": 1008, "name": "三星Galaxy S24", "description": "三星旗舰智能手机，骁龙8 Gen3，2K屏幕，5000万像素，12G+256G，安卓手机"},
            {"product_id": 1009, "name": "华硕天选5", "description": "15.6英寸游戏本，锐龙7，RTX4070，16G+1TB，高刷新率屏幕，电竞游戏本，游戏电脑"},
            {"product_id": 1010, "name": "微软Surface Pro 9", "description": "二合一平板电脑，酷睿i7，16G+512G，触控屏，便携办公，平板笔记本"}
        ]
        df = pd.DataFrame(products)
        # 合并名称和描述，生成用于语义匹配的文本
        df["match_text"] = df["name"] + "：" + df["description"]
        print(f"商品库初始化完成，共{len(df)}个商品")
        return df
    
    def _precompute_product_embeddings(self) -> Dict[int, torch.Tensor]:
        """预计算所有商品的语义向量（提升推荐速度）"""
        print("正在预计算商品语义向量...")
        device = "cuda" if torch.cuda.is_available() else "cpu"
        embeddings = {}
        for _, row in self.product_db.iterrows():
            product_id = row["product_id"]
            match_text = row["match_text"]
            # 生成归一化的向量，并移动到相同设备
            embedding = self.model.encode(
                match_text,
                convert_to_tensor=True,
                normalize_embeddings=True,
                device=device
            )
            embeddings[product_id] = embedding
        print(f"商品向量预计算完成（设备：{device}）")
        return embeddings
    
    def get_user_embedding(self, user_query: str) -> torch.Tensor:
        """生成用户查询文本的语义向量"""
        if not user_query.strip():
            raise ValueError("用户查询文本不能为空")

        device = "cuda" if torch.cuda.is_available() else "cpu"
        embedding = self.model.encode(
            user_query,
            convert_to_tensor=True,
            normalize_embeddings=True,
            device=device
        )
        return embedding
    
    def calculate_product_similarity(self, user_embedding: torch.Tensor) -> pd.DataFrame:
        """计算用户查询与所有商品的相似度"""
        # 准备相似度计算结果
        similarity_results = []
        
        for product_id, prod_embedding in self.product_embeddings.items():
            # 计算余弦相似度
            similarity = F.cosine_similarity(user_embedding.unsqueeze(0), prod_embedding.unsqueeze(0)).item()
            # 获取商品信息
            product_info = self.product_db[self.product_db["product_id"] == product_id].iloc[0]
            similarity_results.append({
                "product_id": product_id,
                "name": product_info["name"],
                "similarity": similarity,
                "description": product_info["description"]
            })
        
        # 转换为DataFrame并按相似度降序排序
        results_df = pd.DataFrame(similarity_results)
        results_df = results_df.sort_values(by="similarity", ascending=False)
        return results_df
    
    def recommend(self, user_query: str, top_n: int = TOP_N_RECOMMEND) -> Tuple[pd.DataFrame, pd.DataFrame]:
        """
        核心推荐函数
        :param user_query: 用户输入的查询文本
        :param top_n: 返回的推荐商品数量
        :return: Top-N推荐结果、所有商品的相似度结果
        """
        # 生成用户查询向量
        user_embedding = self.get_user_embedding(user_query)

        # 计算所有商品相似度
        all_results = self.calculate_product_similarity(user_embedding)

        # 调试：打印所有商品的相似度
        print(f"\n【调试】所有商品相似度排序：")
        for idx, row in all_results.iterrows():
            print(f"  {row['name']}: {row['similarity']:.4f}")

        # 获取Top-N结果
        top_results = all_results.head(top_n).copy()

        # 格式化相似度值（保留4位小数）
        top_results["similarity"] = top_results["similarity"].apply(lambda x: f"{x:.4f}")
        all_results["similarity"] = all_results["similarity"].apply(lambda x: f"{x:.4f}")

        return top_results, all_results
    
    def visualize_recommendation(self, user_query: str, all_results: pd.DataFrame):
        """
        可视化推荐结果
        :param user_query: 用户查询文本
        :param all_results: 所有商品的相似度结果
        """
        # 准备可视化数据
        plot_data = all_results.head(10).copy()  # 取前10个商品可视化
        plot_data["similarity_num"] = plot_data["similarity"].astype(float)
        
        # 创建画布
        fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(16, 6))
        
        # 子图1：相似度条形图
        sns.barplot(
            x="similarity_num",
            y="name",
            data=plot_data,
            ax=ax1,
            palette="viridis"
        )
        ax1.set_title(f"用户查询：{user_query}\n商品相似度排名（Top10）", fontsize=14, pad=20)
        ax1.set_xlabel("语义相似度", fontsize=12)
        ax1.set_ylabel("商品名称", fontsize=12)
        ax1.set_xlim(0, 1.0)
        
        # 在条形图上添加数值标签
        for i, v in enumerate(plot_data["similarity_num"]):
            ax1.text(v + 0.01, i, f"{v:.3f}", va="center", fontsize=10)
        
        # 子图2：相似度分布直方图
        all_similarities = all_results["similarity"].astype(float)
        sns.histplot(
            all_similarities,
            bins=8,
            ax=ax2,
            kde=True,
            color="#2ecc71"
        )
        ax2.set_title("所有商品相似度分布", fontsize=14, pad=20)
        ax2.set_xlabel("相似度值", fontsize=12)
        ax2.set_ylabel("商品数量", fontsize=12)
        ax2.axvline(
            all_similarities.mean(),
            color="red",
            linestyle="--",
            label=f"平均值：{all_similarities.mean():.3f}"
        )
        ax2.legend()
        
        # 调整布局并保存图片
        plt.tight_layout()
        plt.savefig(f"121.{user_query} recommendation_result_{hash(user_query)}.png", dpi=300, bbox_inches="tight")
        print(f"\n可视化图表已保存为：recommendation_result_{hash(user_query)}.png")
        plt.show()

# 系统测试与使用示例
def main():
    """推荐系统使用示例"""
    # 初始化推荐系统（首次运行会下载模型，后续运行直接加载缓存）
    recommender = SemanticRecommender()
    
    # 测试多个用户查询场景
    test_queries = [
        "我想买轻薄的笔记本电脑，方便办公携带",
        "推荐一款性能好的智能手机",
        "适合玩游戏的笔记本电脑",
        "平板电脑 学习办公 二合一"
    ]
    
    # 遍历测试查询
    for query in test_queries:
        print("\n" + "="*80)
        print(f"用户查询：{query}")
        print("="*80)
        
        # 获取推荐结果
        top_recommendations, all_results = recommender.recommend(query, top_n=5)
        
        # 打印Top-N推荐结果
        print("\n【Top5推荐商品】")
        print("-" * 60)
        for idx, (_, row) in enumerate(top_recommendations.iterrows(), 1):
            print(f"{idx}. 商品ID：{row['product_id']}")
            print(f"   名称：{row['name']}")
            print(f"   相似度：{row['similarity']}")
            print(f"   描述：{row['description']}")
            print("-" * 60)
        
        # 可视化推荐结果
        recommender.visualize_recommendation(query, all_results)

if __name__ == "__main__":
    main()

def calculate_batch_similarity(self, user_embedding: torch.Tensor) -> pd.DataFrame:
    """批量计算相似度（优化版本）"""
    # 将所有商品向量堆叠成矩阵
    product_embeddings_tensor = torch.stack(list(self.product_embeddings.values()))
    
    # 批量计算相似度（一次计算所有商品）
    similarities = F.cosine_similarity(
        user_embedding.unsqueeze(0).repeat(len(product_embeddings_tensor), 1),
        product_embeddings_tensor
    )
    
    # 构建结果
    product_ids = list(self.product_embeddings.keys())
    similarity_results = []
    
    for i, product_id in enumerate(product_ids):
        product_info = self.product_db[self.product_db["product_id"] == product_id].iloc[0]
        similarity_results.append({
            "product_id": product_id,
            "name": product_info["name"],
            "similarity": similarities[i].item(),
            "description": product_info["description"]
        })
    
    results_df = pd.DataFrame(similarity_results)
    results_df = results_df.sort_values(by="similarity", ascending=False)
    return results_df

import pickle

def save_product_embeddings(self, filepath: str = "product_embeddings.pkl"):
    """保存商品向量到本地"""
    with open(filepath, "wb") as f:
        pickle.dump(self.product_embeddings, f)
    print(f"商品向量已保存到：{filepath}")

def load_product_embeddings(self, filepath: str = "product_embeddings.pkl"):
    """从本地加载商品向量"""
    import os
    if os.path.exists(filepath):
        with open(filepath, "rb") as f:
            self.product_embeddings = pickle.load(f)
        print(f"从本地加载商品向量：{filepath}")
    else:
        self.product_embeddings = self._precompute_product_embeddings()
        self.save_product_embeddings(filepath)

from fastapi import FastAPI
import uvicorn

app = FastAPI(title="语义推荐系统API")
recommender = SemanticRecommender()

@app.get("/recommend")
def recommend_api(query: str, top_n: int = 5):
    """推荐API接口"""
    top_results, _ = recommender.recommend(query, top_n=top_n)
    return {
        "query": query,
        "top_recommendations": top_results.to_dict("records")
    }

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)