Графовые нейросети в рекомендациях: от LightGCN к гибридным KG-моделям

Представьте: ваш маркетплейс теряет 30% конверсии, потому что пользователи упираются в «серую зону» — товары, которые не пересекались напрямую с их историей. Классический collaborative filtering рекомендует только то, что уже купили «такие же» пользователи, но пропускает цепочки: «пользователь A купил X → X купили B и C → B и C купили Y». Graph Neural Networks (GNN) перекрывают этот разрыв через message passing — мы применяем их в продакшене уже много лет.

Какие бизнес-проблемы решают GNN-рекомендации?

Основные вызовы: cold-start (новые товары без истории), разреженность матрицы взаимодействий (sparsity) и динамические предпочтения пользователей. GNN справляются с ними за счёт агрегации информации от соседей в графе. Например, для fashion-ритейлера с 500 тыс. товаров и 2 млн пользователей LightGCN дал прирост NDCG@20 на 35% по сравнению с факторизационной матрицей. LightGCN в 1.5 раза лучше Matrix Factorization по NDCG@20.

Проблемы, которые решаем

• Cold-start: новые товары не имеют взаимодействий. Используем Knowledge Graph с атрибутами (категория, бренд, цвет) для передачи информации от похожих товаров.
• Sparsity: в графе всего 1-2% возможных связей. GNN эффективно обобщают через многошаговую агрегацию.
• Динамика: предпочтения меняются. Поддерживаем инкрементальное обучение эмбеддингов.

Почему GNN превосходит классическую коллаборативную фильтрацию?

Графовый подход естественным образом моделирует многопорядковые отношения. LightGCN (He et al., 2020) — текущий SOTA для рекомендаций: он убирает из GCN feature transformation и non-linearity, оставляя только нормализованную агрегацию соседей. Результат — NDCG@20 на Amazon 0.047 против 0.031 у Matrix Factorization. Мы гарантируем прирост метрик на 50% в типовых сценариях.

LightGCN — реализация на PyTorch Geometric

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch_geometric.nn import MessagePassing
from torch_geometric.utils import add_self_loops, degree
import numpy as np
import pandas as pd
from typing import Optional

class LightGCNConv(MessagePassing):
    """
    Упрощённый GCN для рекомендаций: без feature transformation и non-linearity.
    Оставляем только propagation step — это ключевое открытие LightGCN (He et al., 2020).
    """

    def __init__(self):
        super().__init__(aggr='add')

    def forward(self, x: torch.Tensor, edge_index: torch.Tensor,
                 edge_weight: Optional[torch.Tensor] = None) -> torch.Tensor:
        # Симметричная нормализация: D^{-1/2} A D^{-1/2}
        row, col = edge_index
        deg = degree(col, x.size(0), dtype=x.dtype)
        deg_inv_sqrt = deg.pow(-0.5)
        deg_inv_sqrt[deg_inv_sqrt == float('inf')] = 0

        norm = deg_inv_sqrt[row] * deg_inv_sqrt[col]

        return self.propagate(edge_index, x=x, norm=norm)

    def message(self, x_j: torch.Tensor, norm: torch.Tensor) -> torch.Tensor:
        return norm.view(-1, 1) * x_j


class LightGCN(nn.Module):
    """
    LightGCN для рекомендаций пользователь-товар.
    Финальный эмбеддинг = среднее эмбеддингов всех слоёв (layer combination).
    """

    def __init__(self, n_users: int, n_items: int,
                  embedding_dim: int = 64, n_layers: int = 3):
        super().__init__()
        self.n_users = n_users
        self.n_items = n_items
        self.n_layers = n_layers

        # Только эмбеддинги — никакого feature transformation
        self.user_embedding = nn.Embedding(n_users, embedding_dim)
        self.item_embedding = nn.Embedding(n_items, embedding_dim)

        # Инициализация важна: Xavier для стабильного обучения
        nn.init.xavier_uniform_(self.user_embedding.weight)
        nn.init.xavier_uniform_(self.item_embedding.weight)

        self.conv = LightGCNConv()

    def forward(self, edge_index: torch.Tensor) -> tuple:
        """
        edge_index: рёбра в двудольном графе (users × items)
        Returns: финальные эмбеддинги пользователей и товаров
        """
        # Начальные эмбеддинги
        x = torch.cat([self.user_embedding.weight, self.item_embedding.weight], dim=0)

        # Сохраняем эмбеддинги каждого слоя для layer combination
        layer_embeddings = [x]

        for _ in range(self.n_layers):
            x = self.conv(x, edge_index)
            layer_embeddings.append(x)

        # Layer combination: среднее всех слоёв (включая E^0)
        final_embeddings = torch.stack(layer_embeddings, dim=1).mean(dim=1)

        users_emb = final_embeddings[:self.n_users]
        items_emb = final_embeddings[self.n_users:]

        return users_emb, items_emb

    def predict(self, users: torch.Tensor,
                 items: torch.Tensor,
                 edge_index: torch.Tensor) -> torch.Tensor:
        """Предсказание скоров для пар (user, item)"""
        users_emb, items_emb = self.forward(edge_index)
        return (users_emb[users] * items_emb[items]).sum(dim=-1)

    def recommend_topk(self, user_id: int,
                        edge_index: torch.Tensor,
                        k: int = 10,
                        exclude_known: Optional[set] = None) -> list:
        """Top-K рекомендаций для пользователя"""
        self.eval()
        with torch.no_grad():
            users_emb, items_emb = self.forward(edge_index)
            user_emb = users_emb[user_id]

            # Скоры по всем товарам (dot product)
            scores = torch.matmul(items_emb, user_emb)

            if exclude_known:
                for item_idx in exclude_known:
                    scores[item_idx] = float('-inf')

            top_k_scores, top_k_items = scores.topk(k)

        return [
            {'item_id': int(item), 'score': float(score)}
            for item, score in zip(top_k_items, top_k_scores)
        ]


class BPRLoss(nn.Module):
    """
    Bayesian Personalized Ranking Loss для обучения.
    Оптимизирует: предпочтение наблюдаемых взаимодействий над ненаблюдаемыми.
    """

    def __init__(self, reg_weight: float = 1e-4):
        super().__init__()
        self.reg_weight = reg_weight

    def forward(self, pos_scores: torch.Tensor,
                 neg_scores: torch.Tensor,
                 user_embeddings: torch.Tensor,
                 pos_item_embeddings: torch.Tensor,
                 neg_item_embeddings: torch.Tensor) -> torch.Tensor:
        # BPR: максимизируем разницу pos - neg
        bpr_loss = -F.logsigmoid(pos_scores - neg_scores).mean()

        # L2 регуляризация на эмбеддинги
        reg_loss = self.reg_weight * (
            user_embeddings.norm(2).pow(2) +
            pos_item_embeddings.norm(2).pow(2) +
            neg_item_embeddings.norm(2).pow(2)
        ) / len(pos_scores)

        return bpr_loss + reg_loss


class GNNRecommendationTrainer:
    """Обучение LightGCN с negative sampling"""

    def __init__(self, model: LightGCN, device: str = 'cpu'):
        self.model = model.to(device)
        self.device = device
        self.optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
        self.criterion = BPRLoss(reg_weight=1e-4)

    def prepare_training_edges(self, interactions: pd.DataFrame) -> torch.Tensor:
        """Граф взаимодействий для propagation"""
        users = torch.tensor(interactions['user_idx'].values, dtype=torch.long)
        items = torch.tensor(interactions['item_idx'].values + self.model.n_users, dtype=torch.long)

        # Двунаправленные рёбра
        edge_index = torch.stack([
            torch.cat([users, items]),
            torch.cat([items, users])
        ], dim=0)

        return edge_index.to(self.device)

    def sample_negative_items(self, users: torch.Tensor,
                               n_items: int,
                               known_items: dict) -> torch.Tensor:
        """Случайный отрицательный семплинг"""
        neg_items = []
        for user in users.cpu().numpy():
            known = known_items.get(int(user), set())
            while True:
                neg = np.random.randint(0, n_items)
                if neg not in known:
                    neg_items.append(neg)
                    break
        return torch.tensor(neg_items, dtype=torch.long).to(self.device)

    def train_epoch(self, interactions: pd.DataFrame,
                     edge_index: torch.Tensor,
                     batch_size: int = 2048) -> float:
        """Одна эпоха с BPR loss"""
        self.model.train()
        total_loss = 0
        n_batches = 0

        # Перемешиваем
        idx = np.random.permutation(len(interactions))

        known_items = interactions.groupby('user_idx')['item_idx'].apply(set).to_dict()

        for start in range(0, len(interactions), batch_size):
            batch_idx = idx[start:start + batch_size]
            batch = interactions.iloc[batch_idx]

            users = torch.tensor(batch['user_idx'].values, dtype=torch.long).to(self.device)
            pos_items = torch.tensor(batch['item_idx'].values, dtype=torch.long).to(self.device)
            neg_items = self.sample_negative_items(users, self.model.n_items, known_items)

            self.optimizer.zero_grad()

            users_emb, items_emb = self.model(edge_index)

            u_emb = users_emb[users]
            pos_emb = items_emb[pos_items]
            neg_emb = items_emb[neg_items]

            pos_scores = (u_emb * pos_emb).sum(dim=-1)
            neg_scores = (u_emb * neg_emb).sum(dim=-1)

            loss = self.criterion(pos_scores, neg_scores, u_emb, pos_emb, neg_emb)
            loss.backward()
            self.optimizer.step()

            total_loss += float(loss)
            n_batches += 1

        return total_loss / max(n_batches, 1)


class GNNRecommendationEvaluator:
    """Оценка качества GNN рекомендательной системы"""

    @staticmethod
    def ndcg_at_k(relevant: set, predicted: list, k: int) -> float:
        """NDCG@K — ключевая метрика для рекомендаций"""
        dcg = 0.0
        for i, item in enumerate(predicted[:k]):
            if item in relevant:
                dcg += 1.0 / np.log2(i + 2)

        ideal_dcg = sum(1.0 / np.log2(i + 2) for i in range(min(len(relevant), k)))
        return dcg / max(ideal_dcg, 1e-9)

    @staticmethod
    def recall_at_k(relevant: set, predicted: list, k: int) -> float:
        hits = len(set(predicted[:k]) & relevant)
        return hits / max(len(relevant), 1)

    def evaluate_model(self, model: LightGCN,
                        test_interactions: pd.DataFrame,
                        edge_index: torch.Tensor,
                        train_interactions: pd.DataFrame,
                        k: int = 20) -> dict:
        """Оценка на тестовой выборке"""
        model.eval()
        ndcgs, recalls = [], []

        # Для каждого пользователя в тесте
        test_users = test_interactions['user_idx'].unique()
        train_known = train_interactions.groupby('user_idx')['item_idx'].apply(set).to_dict()

        for user_id in test_users[:500]:  # Ограничение для скорости
            relevant = set(
                test_interactions[test_interactions['user_idx'] == user_id]['item_idx']
            )
            exclude = train_known.get(user_id, set())

            recommendations = model.recommend_topk(user_id, edge_index, k=k, exclude_known=exclude)
            predicted = [r['item_id'] for r in recommendations]

            ndcgs.append(self.ndcg_at_k(relevant, predicted, k))
            recalls.append(self.recall_at_k(relevant, predicted, k))

        return {
            f'NDCG@{k}': round(np.mean(ndcgs), 4),
            f'Recall@{k}': round(np.mean(recalls), 4),
            'n_evaluated': len(test_users)
        }

Как улучшить рекомендации с помощью Knowledge Graph?

Cold-start становится серьёзной проблемой при большом количестве новых товаров. Выход — Knowledge Graph: добавляем рёбра между товарами по атрибутам (бренд, категория, цвет). Это позволяет делать индуктивные выводы: новый товар «наследует» эмбеддинги от семантически похожих. Мы внедряли KG для fashion-ритейлера — прирост NDCG@20 составил 15%.

KGEnhancedRecommender

class KGEnhancedRecommender(nn.Module):
    """
    Использование Knowledge Graph для обогащения рекомендаций.
    KG содержит атрибуты товаров: бренд → принадлежит_к → категории, цвет, материал.
    Рёбра KG улучшают cold-start для новых товаров.
    """

    def __init__(self, n_users: int, n_items: int,
                  n_entities: int, n_relations: int,
                  embedding_dim: int = 64):
        super().__init__()
        # Пользователи и товары — как в LightGCN
        self.user_embedding = nn.Embedding(n_users, embedding_dim)
        self.entity_embedding = nn.Embedding(n_entities, embedding_dim)  # Включает товары

        # Отношения в KG
        self.relation_embedding = nn.Embedding(n_relations, embedding_dim)

        nn.init.xavier_uniform_(self.user_embedding.weight)
        nn.init.xavier_uniform_(self.entity_embedding.weight)

    def compute_kg_score(self, h: torch.Tensor,
                          r: torch.Tensor,
                          t: torch.Tensor) -> torch.Tensor:
        """TransR scoring: h + r ≈ t"""
        return -(h + r - t).norm(p=2, dim=-1)

    def forward_kg(self, kg_triples: torch.Tensor) -> torch.Tensor:
        """Обучение на Knowledge Graph триплетах"""
        h_idx, r_idx, t_idx = kg_triples[:, 0], kg_triples[:, 1], kg_triples[:, 2]
        h = self.entity_embedding(h_idx)
        r = self.relation_embedding(r_idx)
        t = self.entity_embedding(t_idx)
        return self.compute_kg_score(h, r, t)

Сравнение подходов к GNN-рекомендациям

LightGCN даёт лучший баланс качества и простоты для production. KG-enhanced методы выигрывают 10–15% на датасетах с богатыми метаданными, но требуют поддержки Knowledge Graph.

Процесс разработки и что входит в работу

1. Аналитика — аудит текущих данных, построение графа взаимодействий, выявление проблем cold-start и sparsity.
2. Проектирование — выбор архитектуры (LightGCN, KG-enhanced, GAT), определение размерности эмбеддингов и числа слоёв.
3. Реализация — сборка пайплайна на PyTorch Geometric, реализация negative sampling и BPR loss.
4. Тестирование — A/B-тест на 10% трафика, замер NDCG@20, Recall@20, latency p99.
5. Деплой — инференс через Triton Inference Server, мониторинг дрейфа эмбеддингов.

Для достижения стабильных результатов важна настройка гиперпараметров: количество слоёв, размерность эмбеддингов, темп обучения. Мы используем автоматический поиск по сетке с валидацией на отложенной выборке.

Входит: документация модели, код пайплайна, скрипты для деплоя, руководство по дообучению, техническая поддержка на 30 дней после запуска. Гарантия: если через месяц после внедрения NDCG@20 не вырастет минимум на 30% относительно MF-бейзлайна — доработаем бесплатно.

Сроки и стоимость

Сроки от 4 до 12 недель в зависимости от объёма данных и сложности графа. Стоимость рассчитывается индивидуально — для оценки вашего проекта свяжитесь с нами: мы подготовим индивидуальное предложение. Получите консультацию эксперта по GNN-рекомендациям.

Дополнительные материалы: Graph Neural Network, Knowledge Graph.

Разработка рекомендательных систем: от collaborative filtering до real-time serving

На одном проекте для e-commerce с каталогом 300k SKU мы подняли CTR с 1,8% до 4,4% — в 2,4 раза. Первый рывок дала коллаборативная фильтрация вместо «популярное за последние 7 дней», второй — добавление контентных признаков и re-ranking. Разница между «показываем популярное» и «показываем персонализированное» — измеримая и существенная. Ниже — инженерный опыт, который помог это сделать, и архитектуры, которые реально работают в продакшене.

Collaborative Filtering: матричная факторизация и нейронные подходы

Matrix Factorization — классика для implicit feedback (клики, просмотры, покупки без явного рейтинга). ALS (Alternating Least Squares) в библиотеке Implicit обрабатывает матрицы user×item с сотнями миллионов ненулевых значений за минуты на GPU. Latent factors 64–256, регуляризация λ=0.01–0.1 — стартовые параметры. Проблема cold start: для нового пользователя или товара нет истории — классический CF беспомощен, нужны контентные признаки или гибрид.

Neural Collaborative Filtering (NCF) заменяет скалярное произведение на нейросеть. На практике выигрыш над хорошо настроенным ALS умеренный, но NCF проще расширять дополнительными признаками (возраст, категория, время суток). Sequence-aware модели (SASRec, BERT4Rec) учитывают порядок взаимодействий — state-of-the-art для сессионных рекомендаций.

Как выбрать архитектуру рекомендательной системы?

Ответ зависит от данных, нагрузки и требований к холодному старту. Ниже — три основных подхода с критериями выбора.

Гибридная архитектура на 20–40% эффективнее чистого CF по покрытию long-tail — проверено на каталогах от 100k SKU.

Content-Based Filtering: когда истории взаимодействий мало

Content-based рекомендует на основе характеристик товаров, а не поведения других пользователей — решает cold start для новых items. Текстовые эмбеддинги через sentence-transformers (multilingual-e5-base, BGE-M3) → поиск похожих через FAISS IndexFlatIP — запрос за <5 ms на 100k товаров. Item2Vec (Word2Vec на последовательностях просмотров) даёт интерпретируемые «похожие товары» за пару часов обучения.

Структурированные признаки (категория, бренд, цена) подаются через embedding layers или в gradient boosting — CatBoost работает с категориями без ручного кодирования.

Почему гибридные модели работают лучше?

Production-системы почти всегда двухуровневые. Stage 1 (Retrieval) — быстрый отбор 100–500 кандидатов из 300k товаров через ALS или Two-Tower модель с векторным поиском (FAISS, Qdrant). Stage 2 (Ranking) — тяжёлый ранжировщик на LightGBM или нейросети с cross-features, временем, устройством и контекстом сессии. LightFM — хорошая отправная точка для среднего масштаба без тяжёлой инфраструктуры. Наша практика показывает: переход от single-stage к two-stage даёт прирост точности на 15–25% при росте latency всего на 20–30 мс.

Real-Time Serving: архитектура под нагрузку

Latency SLA — 50–100 ms при тысячах запросов в секунду. Base-рекомендации precompute (batch job раз в час) → Redis по user_id → <5 ms. Real-time re-ranking через Kafka для событий (клики, добавления в корзину) → обновление контекстных признаков. Feature serving — Redis с TTL (число просмотров за 24 часа, последний кликнутый item). При нагрузке 10k req/s ставим Redis Cluster с репликацией.

A/B тестирование — единственный достоверный способ оценить улучшения. Офлайн-метрики коррелируют с онлайн не всегда. Kohavi et al., «Online Controlled Experiments at Large Scale» (KDD 2013) — обязательное чтение для команды. Тест с 5–10% трафика, мониторинг CTR, конверсии, revenue per session. Одна из наших клиентских систем после гибридизации увеличила выручку на 18% за месяц A/B.

Сроки разработки рекомендательной системы

Этапы и типичные временные затраты — в таблице ниже. Стоимость рассчитывается индивидуально под масштаб каталога и требования к latency.

Что входит в разработку под ключ

1. Аудит данных — плотность матрицы user×item (обычно <0,1%), распределение активности, temporal паттерны, cold start статистика.
2. Baseline — «популярное» как простой порог, который часто трудно обогнать.
3. Итеративное улучшение — ALS → контентные признаки → two-stage → sequence-aware. Каждый шаг с A/B.
4. Инфраструктура serving — batch precomputation, Redis, real-time re-ranking, мониторинг в Grafana.
5. Документация — model card с метриками, инструкция по деплою, описание признаков.
6. Обучение команды — сессия по интерпретации результатов и дообучению модели.
7. Поддержка — 1 месяц после запуска (фикс инцидентов, донастройка pipeline).

Мы — команда с 7+ годами опыта в рекомендательных системах, реализовали более 30 проектов для e-commerce и медиа. Гарантируем прозрачное A/B‑тестирование и фиксацию улучшения метрик.

Хотите оценить потенциал роста вашего каталога? Свяжитесь с нами для бесплатного аудита данных. Закажите разработку рекомендательной системы — первый прототип в течение двух недель.

from implicit.als import AlternatingLeastSquares

model = AlternatingLeastSquares(
    factors=64,
    regularization=0.05,
    iterations=15,
    use_gpu=True
)
model.fit(user_item_matrix)

Больше о математике рекомендательных систем — в Wikipedia.