Разработка 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 рабочих дня: вышлите описание задачи и примерные объёмы графа (число узлов, рёбер, задача). Свяжитесь с нами, чтобы обсудить задачу.