Разработка AI-системы на базе Graph Neural Networks (GNN)

Мы разрабатываем AI-системы на базе графовых нейронных сетей (GNN) для задач, где важны связи между объектами. Когда табличные данные теряют контекст, граф его сохраняет. Например, в детекции мошенничества GNN анализирует цепочки транзакций, а не отдельные операции. Наш опыт включает проекты для финтеха, e-commerce и биоинформатики — более 30 внедрений за 5+ лет. Закажите разработку GNN-системы под ключ: мы подберём архитектуру под ваши данные и обеспечим продакшн-готовность. Свяжитесь с нами для консультации и оценки графовой структуры.

Почему GNN превосходят классические ML-модели на графовых данных?

Традиционные модели (GBDT, линейная регрессия) работают с векторами признаков, игнорируя топологию. GNN оперирует не только признаками узлов, но и структурой — какие узлы связаны, какова сила связей, каковы типы рёбер. Благодаря message passing, после K итераций каждый узел «видит» своих соседей на расстоянии K. Это даёт прирост AUC на 5–15% в задачах классификации узлов и предсказания связей по сравнению с MLP или CatBoost.

Практический пример: в задаче детекции мошенничества стандартная градиентная модель смотрит на признаки одной транзакции — сумму, время, геолокацию. GNN добавляет к этому контекст: как данный аккаунт связан с другими, были ли среди соседей ранее заблокированные объекты, насколько плотна вокруг него сеть подозрительных транзакций. Именно этот сетевой контекст позволяет выявлять скоординированные схемы мошенничества, незаметные для точечного анализа. В наших проектах переход с CatBoost на GraphSAGE на финансовых данных давал улучшение recall@5% FPR на 12–18 процентных пунктов.

Теоретическая база и ключевые архитектуры

Основная идея GNN — message passing: каждый узел агрегирует информацию от своих соседей. После K итераций узел «видит» K-hop neighbourhood.

Формула агрегации (GraphSAGE):

h_v^(k) = σ(W · CONCAT(h_v^(k-1), AGG({h_u^(k-1), u ∈ N(v)})))

Ключевые архитектуры:

Реализация GCN с PyTorch Geometric

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch_geometric.nn import GCNConv, SAGEConv, GATConv, global_mean_pool
from torch_geometric.data import Data, DataLoader
import numpy as np
import pandas as pd

class GraphConvNet(nn.Module):
    """
    GCN для классификации/регрессии на графе.
    Подходит для: fraud detection, рекомендаций, молекул.
    """

    def __init__(self, node_features: int,
                  hidden_channels: int = 64,
                  output_dim: int = 1,
                  num_layers: int = 3,
                  dropout: float = 0.3):
        super().__init__()

        self.convs = nn.ModuleList()
        self.bns = nn.ModuleList()

        # Входной слой
        self.convs.append(GCNConv(node_features, hidden_channels))
        self.bns.append(nn.BatchNorm1d(hidden_channels))

        # Скрытые слои
        for _ in range(num_layers - 2):
            self.convs.append(GCNConv(hidden_channels, hidden_channels))
            self.bns.append(nn.BatchNorm1d(hidden_channels))

        # Выходной слой
        self.convs.append(GCNConv(hidden_channels, hidden_channels))
        self.bns.append(nn.BatchNorm1d(hidden_channels))

        self.dropout = dropout
        self.classifier = nn.Linear(hidden_channels, output_dim)

    def forward(self, x: torch.Tensor,
                edge_index: torch.Tensor,
                batch: torch.Tensor = None) -> torch.Tensor:
        """
        x: (N, node_features) — матрица признаков узлов
        edge_index: (2, E) — список рёбер в COO формате
        batch: (N,) — принадлежность узлов к графам (для батчинга)
        """
        for conv, bn in zip(self.convs, self.bns):
            x = conv(x, edge_index)
            x = bn(x)
            x = F.relu(x)
            x = F.dropout(x, p=self.dropout, training=self.training)

        # Graph-level readout (для задач на уровне графа)
        if batch is not None:
            x = global_mean_pool(x, batch)

        return self.classifier(x)


class GraphSAGEEncoder(nn.Module):
    """
    GraphSAGE для inductive learning (работает на новых узлах без переобучения).
    Используется для больших графов: социальные сети, транзакции.
    """

    def __init__(self, in_channels: int, hidden_channels: int, out_channels: int,
                  num_layers: int = 3, aggr: str = 'mean'):
        super().__init__()
        self.convs = nn.ModuleList()

        self.convs.append(SAGEConv(in_channels, hidden_channels, aggr=aggr))
        for _ in range(num_layers - 2):
            self.convs.append(SAGEConv(hidden_channels, hidden_channels, aggr=aggr))
        self.convs.append(SAGEConv(hidden_channels, out_channels, aggr=aggr))

    def forward(self, x, edge_index):
        for i, conv in enumerate(self.convs):
            x = conv(x, edge_index)
            if i < len(self.convs) - 1:
                x = F.relu(x)
                x = F.dropout(x, p=0.2, training=self.training)
        return x

    def encode(self, x, edge_index):
        """L2-нормализованные эмбеддинги для downstream задач"""
        out = self.forward(x, edge_index)
        return F.normalize(out, p=2, dim=-1)


class GATNetwork(nn.Module):
    """
    Graph Attention Network: взвешенная агрегация соседей.
    Attention веса показывают «важность» каждого соседа.
    """

    def __init__(self, in_channels: int, hidden_channels: int,
                  out_channels: int, num_heads: int = 8):
        super().__init__()

        self.conv1 = GATConv(in_channels, hidden_channels,
                              heads=num_heads, dropout=0.6)
        self.conv2 = GATConv(hidden_channels * num_heads, out_channels,
                              heads=1, concat=False, dropout=0.6)

    def forward(self, x, edge_index):
        x = F.dropout(x, p=0.6, training=self.training)
        x = F.elu(self.conv1(x, edge_index))
        x = F.dropout(x, p=0.6, training=self.training)
        return self.conv2(x, edge_index)

Построение графа из табличных данных

class GraphBuilder:
    """Конвертация табличных данных в граф для GNN"""

    def build_user_item_graph(self, interactions: pd.DataFrame,
                               user_features: pd.DataFrame,
                               item_features: pd.DataFrame) -> Data:
        """
        Двудольный граф пользователь-товар для рекомендаций.
        interactions: user_id, item_id, rating/count
        """
        # Маппинг ID в индексы узлов
        user_ids = interactions['user_id'].unique()
        item_ids = interactions['item_id'].unique()
        n_users = len(user_ids)

        user_idx = {uid: i for i, uid in enumerate(user_ids)}
        item_idx = {iid: i + n_users for i, iid in enumerate(item_ids)}

        # Рёбра: пользователь → товар
        src = interactions['user_id'].map(user_idx).values
        dst = interactions['item_id'].map(item_idx).values

        # Двунаправленный граф (типично для GNN)
        edge_index = torch.tensor(
            np.vstack([
                np.concatenate([src, dst]),
                np.concatenate([dst, src])
            ]),
            dtype=torch.long
        )

        # Матрица признаков узлов
        # Пользователи: embedding + поведенческие признаки
        user_feat_matrix = user_features.set_index('user_id').reindex(user_ids).fillna(0).values
        # Товары: embedding + характеристики
        item_feat_matrix = item_features.set_index('item_id').reindex(item_ids).fillna(0).values

        # Выравниваем размерности
        max_dim = max(user_feat_matrix.shape[1], item_feat_matrix.shape[1])
        user_feat_padded = np.pad(user_feat_matrix, ((0, 0), (0, max_dim - user_feat_matrix.shape[1])))
        item_feat_padded = np.pad(item_feat_matrix, ((0, 0), (0, max_dim - item_feat_matrix.shape[1])))

        x = torch.tensor(
            np.vstack([user_feat_padded, item_feat_padded]),
            dtype=torch.float
        )

        # Веса рёбер (например, рейтинг)
        edge_attr = torch.tensor(
            np.concatenate([
                interactions['rating'].values,
                interactions['rating'].values  # Зеркальные рёбра
            ]),
            dtype=torch.float
        ).unsqueeze(1)

        return Data(
            x=x,
            edge_index=edge_index,
            edge_attr=edge_attr,
            n_users=n_users
        )

    def build_transaction_graph(self, transactions: pd.DataFrame) -> Data:
        """
        Граф транзакций для fraud detection.
        Узлы: аккаунты, карты, IP-адреса, мерчанты.
        Рёбра: транзакции между ними.
        """
        # Уникальные сущности
        accounts = transactions['account_id'].unique()
        merchants = transactions['merchant_id'].unique()
        n_accounts = len(accounts)

        acc_idx = {a: i for i, a in enumerate(accounts)}
        mer_idx = {m: i + n_accounts for i, m in enumerate(merchants)}

        src = transactions['account_id'].map(acc_idx).values
        dst = transactions['merchant_id'].map(mer_idx).values

        edge_index = torch.tensor([
            np.concatenate([src, dst]),
            np.concatenate([dst, src])
        ], dtype=torch.long)

        # Признаки транзакций как атрибуты рёбер
        edge_attr = torch.tensor(
            transactions[['amount', 'hour_of_day', 'is_international']].values,
            dtype=torch.float
        )
        edge_attr = torch.cat([edge_attr, edge_attr], dim=0)  # Дублируем для зеркальных рёбер

        # Метки: fraud = 1
        if 'is_fraud' in transactions.columns:
            y = torch.tensor(transactions['is_fraud'].values, dtype=torch.long)
        else:
            y = None

        return Data(
            x=torch.zeros(n_accounts + len(merchants), 16),  # Placeholder features
            edge_index=edge_index,
            edge_attr=edge_attr,
            y=y
        )

Обучение и оценка GNN

class GNNTrainer:
    """Pipeline обучения GNN"""

    def __init__(self, model: nn.Module, device: str = 'cuda'):
        self.model = model.to(device)
        self.device = device
        self.optimizer = torch.optim.Adam(model.parameters(), lr=0.01, weight_decay=5e-4)

    def train_epoch(self, data: Data, mask: torch.Tensor = None) -> float:
        """Один эпох для node classification"""
        self.model.train()
        self.optimizer.zero_grad()

        data = data.to(self.device)
        out = self.model(data.x, data.edge_index)

        if mask is not None:
            loss = F.cross_entropy(out[mask], data.y[mask])
        else:
            loss = F.cross_entropy(out, data.y)

        loss.backward()
        self.optimizer.step()
        return float(loss)

    def evaluate(self, data: Data, mask: torch.Tensor) -> dict:
        """Оценка качества предсказаний"""
        self.model.eval()
        with torch.no_grad():
            out = self.model(data.x.to(self.device), data.edge_index.to(self.device))
            pred = out[mask].argmax(dim=-1).cpu()
            true = data.y[mask].cpu()

        from sklearn.metrics import accuracy_score, f1_score, roc_auc_score
        probs = torch.softmax(out[mask], dim=-1)[:, 1].cpu().numpy()

        return {
            'accuracy': accuracy_score(true, pred),
            'f1_macro': f1_score(true, pred, average='macro'),
            'auc': roc_auc_score(true, probs) if len(np.unique(true)) > 1 else 0.5
        }

    def train(self, data: Data,
               n_epochs: int = 200,
               train_mask: torch.Tensor = None,
               val_mask: torch.Tensor = None) -> dict:
        """Полный цикл обучения с early stopping"""
        best_val_auc = 0
        patience, patience_counter = 20, 0
        history = {'train_loss': [], 'val_auc': []}

        for epoch in range(n_epochs):
            loss = self.train_epoch(data, train_mask)
            history['train_loss'].append(loss)

            if val_mask is not None and epoch % 5 == 0:
                metrics = self.evaluate(data, val_mask)
                history['val_auc'].append(metrics['auc'])

                if metrics['auc'] > best_val_auc:
                    best_val_auc = metrics['auc']
                    patience_counter = 0
                    torch.save(self.model.state_dict(), 'best_gnn_model.pt')
                else:
                    patience_counter += 1
                    if patience_counter >= patience:
                        print(f"Early stopping at epoch {epoch}")
                        break

        return {'best_val_auc': best_val_auc, 'history': history}

Масштабирование на большие графы

Стандартный GNN не масштабируется на графы с миллионами узлов — полная матрица смежности не помещается в память. Решения:

• GraphSAGE с mini-batch: семплирование K соседей вместо всех. PyG поддерживает через NeighborLoader с параметром num_neighbors=[25, 10].
• Cluster-GCN: разбиение графа на кластеры, обучение внутри кластеров.
• GraphSAINT: случайное семплирование подграфов с importance sampling.

from torch_geometric.loader import NeighborLoader

def create_scalable_dataloader(data: Data, batch_size: int = 1024) -> NeighborLoader:
    """Mini-batch загрузчик для больших графов"""
    return NeighborLoader(
        data,
        num_neighbors=[25, 10, 5],  # Соседи для 3 hop
        batch_size=batch_size,
        input_nodes=data.train_mask,
        shuffle=True,
        num_workers=4
    )

Область применения и бенчмарки

GNN превосходят традиционные методы только тогда, когда структура графа несёт информацию. Если отношения между объектами случайны — обычный GBDT или MLP покажет сопоставимый результат с меньшей сложностью.

Перед стартом проекта мы проводим аудит: имеет ли ваша доменная область значимую структуру графа, достаточно ли данных для обучения, реалистичен ли целевой AUC. Если GNN не даёт преимущества, честно скажем об этом и предложим более простую модель. Такой подход сохраняет бюджет клиента и повышает долгосрочное доверие к решению.

Что входит в работу

• Анализ графовой структуры и выбор архитектуры (GCN, GraphSAGE, GAT, GIN).
• Построение пайплайна данных: конвертация таблиц в графы, нормализация признаков.
• Обучение и подбор гиперпараметров с валидацией (early stopping, кросс-валидация).
• Развёртывание модели через Triton Inference Server или ONNX Runtime.
• Документация model card, API-спецификация и руководство пользователя.
• Поддержка после внедрения: мониторинг дрейфа данных, дообучение.

Мы гарантируем качество результата — все решения проходят тестирование на ваших данных до фиксации. Оценим ваш проект за 2 – 3 рабочих дня: вышлите описание задачи и примерные объёмы графа (число узлов, рёбер, задача). Свяжитесь с нами, чтобы обсудить задачу.

Разработка рекомендательных систем: от 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.