AI推荐系统：这次怎么落地的

# 商品表结构（简化版）
class Product:
    id: int
    name: str
    category: str
    tags: List[str]
    price: float
    created_at: datetime
    # ... 其他字段

# 用户行为表结构
class UserBehavior:
    user_id: int
    product_id: int
    action: str  # 'view', 'favorite', 'purchase'
    timestamp: datetime
    # ... 其他字段

import numpy as np
import pandas as pd
from collections import defaultdict

def build_user_item_matrix(behaviors):
    """
    构建用户-商品交互矩阵
    """
    # 初始化数据结构
    user_ids = set()
    item_ids = set()
    interactions = defaultdict(lambda: defaultdict(float))

    # 处理行为数据，给不同行为不同的权重
    action_weights = {
        'view': 1.0,
        'favorite': 2.0,
        'purchase': 5.0
    }

    for behavior in behaviors:
        user_id = behavior['user_id']
        item_id = behavior['product_id']
        action = behavior['action']

        user_ids.add(user_id)
        item_ids.add(item_id)

        # 累加行为权重
        interactions[user_id][item_id] += action_weights.get(action, 0)

    # 转换为矩阵形式
    user_list = sorted(user_ids)
    item_list = sorted(item_ids)
    user_to_idx = {uid: idx for idx, uid in enumerate(user_list)}
    item_to_idx = {iid: idx for idx, iid in enumerate(item_list)}

    matrix = np.zeros((len(user_list), len(item_list)))

    for user_id, items in interactions.items():
        for item_id, weight in items.items():
            matrix[user_to_idx[user_id], item_to_idx[item_id]] = weight

    return matrix, user_to_idx, item_to_idx

from scipy.sparse.linalg import svds

def matrix_factorization(matrix, n_factors=10):
    """
    矩阵分解
    """
    # 填充缺失值为0
    user_means = np.mean(matrix, axis=1)
    matrix_demeaned = matrix - user_means.reshape(-1, 1)

    # SVD分解
    U, sigma, Vt = svds(matrix_demeaned, k=n_factors)

    # 重构预测矩阵
    sigma = np.diag(sigma)
    predicted_ratings = np.dot(np.dot(U, sigma), Vt) + user_means.reshape(-1, 1)

    return predicted_ratings, U, sigma, Vt

from sklearn.metrics.pairwise import cosine_similarity

def calculate_item_similarity(Vt):
    """
    计算商品相似度矩阵
    """
    # Vt是商品特征矩阵的转置，转回来
    item_features = Vt.T

    # 计算余弦相似度
    similarity_matrix = cosine_similarity(item_features)

    return similarity_matrix

def get_similar_items(item_id, item_to_idx, similarity_matrix, top_k=10):
    """
    获取相似商品
    """
    if item_id not in item_to_idx:
        return []

    item_idx = item_to_idx[item_id]
    similarities = similarity_matrix[item_idx]

    # 获取最相似的top_k个商品
    similar_indices = np.argsort(similarities)[::-1][1:top_k+1]  # 排除自己

    idx_to_item = {idx: iid for iid, idx in item_to_idx.items()}
    similar_items = [
        (idx_to_item[idx], similarities[idx])
        for idx in similar_indices
        if similarities[idx] > 0
    ]

    return similar_items

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

def build_content_similarity(products):
    """
    基于商品内容特征构建相似度
    """
    # 提取文本特征
    texts = []
    for product in products:
        # 组合名称、分类、标签
        text = f"{product['name']} {product['category']} {' '.join(product['tags'])}"
        texts.append(text)

    # TF-IDF向量化
    vectorizer = TfidfVectorizer(max_features=1000)
    tfidf_matrix = vectorizer.fit_transform(texts)

    # 计算相似度
    similarity_matrix = cosine_similarity(tfidf_matrix)

    return similarity_matrix, vectorizer

class RecommendationService:
    def __init__(self):
        self.user_item_matrix = None
        self.predicted_ratings = None
        self.item_similarity = None
        self.content_similarity = None
        self.user_to_idx = {}
        self.item_to_idx = {}
        self.products = {}

    def train(self, behaviors, products):
        """
        训练推荐模型
        """
        # 构建用户-商品矩阵
        self.user_item_matrix, self.user_to_idx, self.item_to_idx = \
            build_user_item_matrix(behaviors)

        # 矩阵分解
        self.predicted_ratings, U, sigma, Vt = \
            matrix_factorization(self.user_item_matrix)

        # 商品相似度（协同过滤）
        self.item_similarity = calculate_item_similarity(Vt)

        # 内容相似度（用于冷启动）
        self.content_similarity, _ = build_content_similarity(products)

        # 保存商品信息
        self.products = {p['id']: p for p in products}

    def recommend_for_user(self, user_id, top_k=10):
        """
        为用户推荐商品
        """
        if user_id not in self.user_to_idx:
            return self._get_popular_items(top_k)

        user_idx = self.user_to_idx[user_id]
        user_ratings = self.predicted_ratings[user_idx]

        # 获取用户未交互的商品
        interacted_items = set(self.user_item_matrix[user_idx].nonzero()[1])
        recommendations = []

        for item_idx, rating in enumerate(user_ratings):
            if item_idx not in interacted_items:
                item_id = list(self.item_to_idx.keys())[list(self.item_to_idx.values()).index(item_idx)]
                recommendations.append((item_id, rating))

        # 按预测评分排序
        recommendations.sort(key=lambda x: x[1], reverse=True)
        return recommendations[:top_k]

    def recommend_similar_items(self, item_id, top_k=10):
        """
        推荐相似商品
        """
        if item_id in self.item_to_idx:
            # 使用协同过滤的相似度
            return get_similar_items(item_id, self.item_to_idx, self.item_similarity, top_k)
        else:
            # 使用内容相似度处理冷启动
            return self._get_content_similar_items(item_id, top_k)

    def _get_popular_items(self, top_k=10):
        """
        获取热门商品
        """
        # 简单实现：基于交互次数
        item_popularity = np.sum(self.user_item_matrix, axis=0)
        popular_indices = np.argsort(item_popularity)[::-1][:top_k]

        idx_to_item = {idx: iid for iid, idx in self.item_to_idx.items()}
        return [(idx_to_item[idx], item_popularity[idx]) for idx in popular_indices]

    def _get_content_similar_items(self, item_id, top_k=10):
        """
        基于内容获取相似商品
        """
        # 这里简化处理，实际需要根据商品ID查找在content_similarity中的位置
        return []

import redis
import json
import hashlib

class CachedRecommendationService:
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis_client = redis.StrictRedis(
            host=redis_host,
            port=redis_port,
            decode_responses=True
        )
        self.base_service = RecommendationService()

    def _get_cache_key(self, method_name, *args):
        """
        生成缓存键
        """
        key_data = f"{method_name}:{args}"
        return hashlib.md5(key_data.encode()).hexdigest()

    def recommend_for_user(self, user_id, top_k=10, cache_ttl=3600):
        """
        带缓存的用户推荐
        """
        cache_key = self._get_cache_key('recommend_for_user', user_id, top_k)

        # 尝试从缓存获取
        cached_result = self.redis_client.get(cache_key)
        if cached_result:
            return json.loads(cached_result)

        # 缓存未命中，调用基础服务
        result = self.base_service.recommend_for_user(user_id, top_k)

        # 写入缓存
        self.redis_client.setex(
            cache_key,
            cache_ttl,
            json.dumps(result)
        )

        return result

    def invalidate_user_cache(self, user_id):
        """
        使特定用户的缓存失效
        """
        # 这里简化处理，实际可能需要更复杂的缓存键管理
        pattern = f"*recommend_for_user:{user_id}*"
        keys = self.redis_client.keys(pattern)
        if keys:
            self.redis_client.delete(*keys)