Content-Based Recommendation System

Реализация рекомендательной системы на основе контента (Content-Based)

Content-Based Filtering рекомендует товары/контент, похожий на то, что пользователь уже любит, на основе характеристик самих объектов. Не требует данных о других пользователях — решает проблему холодного старта для новых пользователей. Особенно эффективен в нишевых доменах с богатыми метаданными.

Многомодальный контентный профиль

import numpy as np
import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.preprocessing import MultiLabelBinarizer, StandardScaler
from sklearn.metrics.pairwise import cosine_similarity
import torch
from sentence_transformers import SentenceTransformer

class ContentBasedRecommender:
    def __init__(self):
        self.text_model = SentenceTransformer('paraphrase-multilingual-mpnet-base-v2')
        self.tfidf = TfidfVectorizer(max_features=5000, ngram_range=(1, 2))
        self.mlb = MultiLabelBinarizer()
        self.scaler = StandardScaler()
        self.item_vectors = None
        self.item_ids = []

    def build_item_profiles(self, items_df: pd.DataFrame) -> np.ndarray:
        """
        items_df: item_id, title, description, category (list), tags (list),
                  price, rating, release_year, ...
        """
        feature_parts = []

        # 1. Семантические эмбеддинги текста (описание + название)
        texts = (
            items_df['title'].fillna('') + '. ' +
            items_df.get('description', pd.Series([''] * len(items_df))).fillna('')
        ).tolist()

        print(f"Encoding {len(texts)} items...")
        text_embeddings = self.text_model.encode(
            texts, batch_size=64, show_progress_bar=True,
            convert_to_numpy=True, normalize_embeddings=True
        )
        feature_parts.append(text_embeddings)

        # 2. TF-IDF признаки из текста
        tfidf_features = self.tfidf.fit_transform(texts).toarray()
        # PCA для сжатия
        from sklearn.decomposition import TruncatedSVD
        svd = TruncatedSVD(n_components=50)
        tfidf_compact = svd.fit_transform(tfidf_features)
        feature_parts.append(tfidf_compact)

        # 3. Категориальные признаки
        if 'categories' in items_df.columns:
            cat_features = self.mlb.fit_transform(
                items_df['categories'].apply(lambda x: x if isinstance(x, list) else [])
            )
            feature_parts.append(cat_features.astype(float))

        # 4. Числовые признаки
        num_cols = ['price', 'rating', 'review_count', 'release_year']
        available_num = [c for c in num_cols if c in items_df.columns]
        if available_num:
            num_features = self.scaler.fit_transform(
                items_df[available_num].fillna(items_df[available_num].median())
            )
            feature_parts.append(num_features)

        # Взвешенная конкатенация
        weights = [2.0, 0.5, 1.0, 0.3][:len(feature_parts)]
        weighted_parts = [p * w for p, w in zip(feature_parts, weights)]
        combined = np.hstack(weighted_parts)

        # L2 нормализация
        norms = np.linalg.norm(combined, axis=1, keepdims=True)
        self.item_vectors = combined / (norms + 1e-10)
        self.item_ids = items_df['item_id'].tolist()

        return self.item_vectors

    def build_user_profile(self, liked_items: list[str],
                            weights: list[float] = None) -> np.ndarray:
        """Профиль пользователя как взвешенное среднее понравившихся товаров"""
        item_indices = [
            self.item_ids.index(item_id)
            for item_id in liked_items
            if item_id in self.item_ids
        ]

        if not item_indices:
            return None

        liked_vectors = self.item_vectors[item_indices]

        if weights is not None and len(weights) == len(item_indices):
            w = np.array(weights[:len(item_indices)]).reshape(-1, 1)
            user_vector = np.average(liked_vectors, axis=0, weights=w.flatten())
        else:
            # Последние взаимодействия важнее
            recency_weights = np.exp(np.linspace(-1, 0, len(item_indices)))
            user_vector = np.average(liked_vectors, axis=0, weights=recency_weights)

        norm = np.linalg.norm(user_vector)
        return user_vector / (norm + 1e-10)

    def recommend(self, user_profile: np.ndarray,
                   exclude_items: list[str] = None,
                   n: int = 10,
                   diversity_penalty: float = 0.1) -> list[tuple]:
        """Рекомендации с штрафом за похожесть между рекомендациями"""
        if user_profile is None:
            return []

        # Базовые скоры
        scores = cosine_similarity(
            user_profile.reshape(1, -1), self.item_vectors
        )[0]

        # Исключение уже просмотренных
        if exclude_items:
            for item_id in exclude_items:
                if item_id in self.item_ids:
                    idx = self.item_ids.index(item_id)
                    scores[idx] = -1

        # MMR (Maximal Marginal Relevance) для разнообразия
        selected_indices = []
        selected_embeddings = []

        while len(selected_indices) < n:
            if not selected_embeddings:
                # Первый — самый релевантный
                best_idx = np.argmax(scores)
            else:
                # Следующие: релевантность минус штраф за схожесть с уже выбранными
                selected_matrix = np.vstack(selected_embeddings)
                similarity_to_selected = cosine_similarity(
                    self.item_vectors, selected_matrix
                ).max(axis=1)

                adjusted_scores = scores - diversity_penalty * similarity_to_selected
                # Маскировка уже выбранных
                for idx in selected_indices:
                    adjusted_scores[idx] = -1
                best_idx = np.argmax(adjusted_scores)

            if scores[best_idx] < 0:
                break

            selected_indices.append(best_idx)
            selected_embeddings.append(self.item_vectors[best_idx])

        return [
            (self.item_ids[idx], float(scores[idx]))
            for idx in selected_indices
        ]

    def item_to_item(self, item_id: str, n: int = 10) -> list[tuple]:
        """Похожие товары для страницы товара"""
        if item_id not in self.item_ids:
            return []

        item_idx = self.item_ids.index(item_id)
        item_vector = self.item_vectors[item_idx]

        scores = cosine_similarity(
            item_vector.reshape(1, -1), self.item_vectors
        )[0]
        scores[item_idx] = -1  # Исключаем сам товар

        top_indices = np.argsort(scores)[-n:][::-1]
        return [(self.item_ids[idx], float(scores[idx])) for idx in top_indices]

Обновление профиля в реальном времени

class OnlineUserProfileUpdater:
    """Инкрементальное обновление профиля без перестройки"""

    def __init__(self, recommender: ContentBasedRecommender):
        self.rec = recommender
        self.user_profiles = {}
        self.user_history = {}

    def update_on_interaction(self, user_id: str, item_id: str,
                               interaction_type: str):
        """Обновление профиля после взаимодействия"""
        weights = {
            'view': 1.0, 'click': 1.5, 'add_to_cart': 3.0,
            'purchase': 5.0, 'dislike': -2.0, 'skip': -0.5
        }
        weight = weights.get(interaction_type, 1.0)

        if user_id not in self.user_history:
            self.user_history[user_id] = []

        self.user_history[user_id].append({
            'item_id': item_id,
            'weight': weight,
            'interaction_type': interaction_type
        })

        # Пересчёт профиля (последние 50 взаимодействий)
        history = self.user_history[user_id][-50:]
        liked = [h['item_id'] for h in history if h['weight'] > 0]
        liked_weights = [h['weight'] for h in history if h['weight'] > 0]

        if liked:
            self.user_profiles[user_id] = self.rec.build_user_profile(
                liked, liked_weights
            )

    def get_recommendations(self, user_id: str, n: int = 10) -> list[tuple]:
        profile = self.user_profiles.get(user_id)
        if profile is None:
            return []
        exclude = [h['item_id'] for h in self.user_history.get(user_id, [])]
        return self.rec.recommend(profile, exclude_items=exclude, n=n)

Content-based с sentence-transformers даёт Precision@10 = 0.15-0.30 на новых пользователях (vs 0.01-0.03 у популярных товаров). Строить профиль достаточно из 3-5 взаимодействий. Обновление: SentenceTransformer энкодинг 10K товаров — 5-10 минут на CPU, 30-60 секунд на GPU.