GNN-рекомендації: LightGCN та KG-моделі

Уявіть: ваш маркетплейс втрачає 30% конверсії, тому що користувачі натикаються на «сіру зону» — товари, які не перетиналися безпосередньо з їхньою історією. Класичний collaborative filtering рекомендує лише те, що вже купили «такі ж» користувачі, але пропускає ланцюжки: «користувач A купив X → X купили B і C → B і C купили Y». Graph Neural Networks (GNN) перекривають цей розрив через message passing — ми застосовуємо їх у продакшені вже багато років. Маємо 5+ років досвіду впровадження GNN-рішень, реалізовано понад 10 проектів у рекомендаційних системах. Типова економія клієнтів від впровадження складає 30% витрат на маркетинг.

Які бізнес-проблеми вирішують GNN-рекомендації?

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

Проблеми, які вирішуємо

• 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.