Замовте розробку AI-системи на базі графових нейронних мереж (GNN)

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 ) 

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} 

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 не дає переваги, чесно скажемо про це і запропонуємо простішу модель. Такий підхід зберігає бюджет клієнта та підвищує довгострокову довіру до рішення.

Що входить в роботу (покроковий план)

1. Аналіз графової структури та вибір архітектури (GCN, GraphSAGE, GAT, GIN).
2. Побудова пайплайну даних: конвертація таблиць у графи, нормалізація ознак.
3. Навчання та підбір гіперпараметрів з валідацією (early stopping, крос-валідація).
4. Розгортання моделі через Triton Inference Server або ONNX Runtime.
5. Документація model card, API-специфікація та керівництво користувача.
6. Підтримка після впровадження: моніторинг дрейфу даних, донавчання.

Ми гарантуємо якість результату — всі рішення проходять тестування на ваших даних до фіксації. Оцінимо ваш проєкт за 2 – 3 робочих дні: надішліть опис задачі та приблизні обсяги графа (число вузлів, ребер, задача). Зв'яжіться з нами, щоб обговорити задачу.