Two-Tower і FAISS: AI-рекомендаційна система для персоналізації

Новий користувач заходить на маркетплейс — система не знає його вподобань. Він бачить популярні товари, але конверсія падає. Це типовий cold-start. Архітектура Two-Tower + FAISS — перевірений production-ready підхід, який дає CTR 8-15% і додатковий виторг 12-25%. Ми реалізували десятки таких проектів, від стартапів до enterprise.

Розглянемо ключові проблеми: cold-start, масштабування до мільйонів транзакцій, баланс точності та latency. Кожна потребує окремого рішення. Production-ready розробка AI-рекомендаційної системи будується на Two-Tower нейронній мережі та FAISS індексі.

Як Two-Tower вирішує проблему холодного старту?

Холодний старт для нових користувачів — типовий біль. Без історії покупок система видає випадкові товари, конверсія падає. Наш підхід використовує семантичні ембеддинги товарів та демографічні ознаки, щоб дати релевантні рекомендації з першого візиту. Matrix Factorization працює, але тільки для користувачів з історією — для холодного старту потрібні контентні ознаки.

Використовуємо гібридний підхід: content-based фільтрація на основі текстових ембеддингів (BERT, 768-dim) для нових користувачів, а після накопичення 3–5 взаємодій перемикаємось на collaborative filtering. Recall@10 для холодних користувачів досягає 20% проти 5% у чистого MF.

Чому Two-Tower архітектура ефективніша за content-based?

Two-tower дозволяє вчити ембеддинги користувачів і товарів в єдиному просторі, навчаючись на implicit feedback (кліки, перегляди). Це дає CTR 8-15% проти 2-4% у rule-based систем. На великих обсягах (>5M транзакцій) early fusion не масштабується — two-tower з in-batch negatives навчається за 2-3 години на 8 V100.

Порівняння метрик до та після впровадження

У результаті впровадження додатковий виторг зростає на 12–25%, що для великого e-commerce майданчика може скласти мільйони гривень на місяць.

Технічна реалізація: Two-Tower, навчання та інференс

Архітектура Two-Tower

import torch import torch.nn as nn import numpy as np from torch.utils.data import DataLoader, Dataset class UserTower(nn.Module): """Енкодер користувача""" def __init__(self, n_users: int, n_categories: int, embedding_dim: int = 64, hidden_dim: int = 128): super().__init__() self.user_emb = nn.Embedding(n_users + 1, embedding_dim, padding_idx=0) self.category_emb = nn.Embedding(n_categories + 1, 16, padding_idx=0) self.mlp = nn.Sequential( nn.Linear(embedding_dim + 16 * 5 + 10, hidden_dim), nn.ReLU(), nn.Dropout(0.3), nn.Linear(hidden_dim, 64), nn.LayerNorm(64) ) def forward(self, user_id, top_categories, behavior_features): user_vec = self.user_emb(user_id) cat_vecs = self.category_emb(top_categories).view(top_categories.shape[0], -1) combined = torch.cat([user_vec, cat_vecs, behavior_features], dim=1) return self.mlp(combined) class ItemTower(nn.Module): """Енкoder предмета (товару/контенту)""" def __init__(self, n_items: int, n_categories: int, embedding_dim: int = 64, text_dim: int = 128): super().__init__() self.item_emb = nn.Embedding(n_items + 1, embedding_dim, padding_idx=0) self.category_emb = nn.Embedding(n_categories + 1, 16, padding_idx=0) self.mlp = nn.Sequential( nn.Linear(embedding_dim + 16 + text_dim + 5, 128), nn.ReLU(), nn.Dropout(0.3), nn.Linear(128, 64), nn.LayerNorm(64) ) def forward(self, item_id, category_id, text_features, item_features): item_vec = self.item_emb(item_id) cat_vec = self.category_emb(category_id) combined = torch.cat([item_vec, cat_vec, text_features, item_features], dim=1) return self.mlp(combined) class TwoTowerModel(nn.Module): def __init__(self, n_users, n_items, n_categories, text_dim=128): super().__init__() self.user_tower = UserTower(n_users, n_categories) self.item_tower = ItemTower(n_items, n_categories, text_dim=text_dim) self.temperature = nn.Parameter(torch.ones(1) * 0.05) def forward(self, user_inputs, item_inputs): user_emb = self.user_tower(**user_inputs) item_emb = self.item_tower(**item_inputs) # Cosine similarity user_norm = nn.functional.normalize(user_emb, dim=1) item_norm = nn.functional.normalize(item_emb, dim=1) scores = torch.sum(user_norm * item_norm, dim=1) / self.temperature return scores def get_user_embedding(self, user_inputs) -> torch.Tensor: with torch.no_grad(): return nn.functional.normalize(self.user_tower(**user_inputs), dim=1) def get_item_embedding(self, item_inputs) -> torch.Tensor: with torch.no_grad(): return nn.functional.normalize(self.item_tower(**item_inputs), dim=1) 

In-Batch Negative Sampling навчання

class RecommendationTrainer: def __init__(self, model: TwoTowerModel, lr: float = 1e-3): self.model = model self.optimizer = torch.optim.Adam(model.parameters(), lr=lr) self.scheduler = torch.optim.lr_scheduler.CosineAnnealingLR( self.optimizer, T_max=10 ) def train_epoch(self, dataloader: DataLoader) -> float: self.model.train() total_loss = 0 for batch in dataloader: user_inputs, pos_item_inputs = batch['user'], batch['positive_item'] # User embeddings: [batch_size, dim] user_embs = nn.functional.normalize( self.model.user_tower(**user_inputs), dim=1 ) # Item embeddings: [batch_size, dim] item_embs = nn.functional.normalize( self.model.item_tower(**pos_item_inputs), dim=1 ) # In-batch negatives: матриця схожостей [batch_size x batch_size] scores = torch.matmul(user_embs, item_embs.T) / self.model.temperature # Діагональ = позитивні приклади labels = torch.arange(len(user_embs)).to(scores.device) loss = nn.functional.cross_entropy(scores, labels) self.optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0) self.optimizer.step() total_loss += loss.item() self.scheduler.step() return total_loss / len(dataloader) 

ANN-індекс для реального часу

import faiss class RecommendationIndex: """Індекс для швидкого пошуку top-K рекомендацій""" def __init__(self, dim: int = 64, n_items: int = 100000): # IVF-PQ індекс: компроміс швидкість/точність quantizer = faiss.IndexFlatIP(dim) n_lists = min(int(np.sqrt(n_items)), 4096) self.index = faiss.IndexIVFPQ(quantizer, dim, n_lists, 8, 8) self.index.nprobe = 32 # Число clusters для пошуку self.item_ids = [] def build(self, item_embeddings: np.ndarray, item_ids: list): """Побудова індексу з ембеддингів""" embeddings_norm = item_embeddings / np.linalg.norm( item_embeddings, axis=1, keepdims=True ) self.index.train(embeddings_norm.astype(np.float32)) self.index.add(embeddings_norm.astype(np.float32)) self.item_ids = item_ids def recommend(self, user_embedding: np.ndarray, k: int = 50, exclude_ids: list = None) -> list[tuple]: """Top-K рекомендацій для користувача""" user_norm = user_embedding / np.linalg.norm(user_embedding) scores, indices = self.index.search( user_norm.reshape(1, -1).astype(np.float32), k + (len(exclude_ids) if exclude_ids else 0) + 10 ) results = [] for score, idx in zip(scores[0], indices[0]): if idx < 0: continue item_id = self.item_ids[idx] if exclude_ids and item_id in exclude_ids: continue results.append((item_id, float(score))) if len(results) >= k: break return results 

Re-ranking та бізнес-правила

class RecommendationReranker: """Застосування бізнес-логіки до кандидатів від retrieval моделі""" def rerank(self, candidates: list[tuple], user_context: dict, business_rules: dict) -> list[tuple]: """ candidates: [(item_id, base_score)] business_rules: {'boost_new_items': 1.2, 'boost_on_sale': 1.15, 'max_same_category': 3} """ scored_candidates = [] category_count = {} for item_id, base_score in candidates: item = self._get_item_metadata(item_id) if item is None: continue # Застосування бусту score = base_score if business_rules.get('boost_new_items') and item.get('is_new'): score *= business_rules['boost_new_items'] if business_rules.get('boost_on_sale') and item.get('on_sale'): score *= business_rules['boost_on_sale'] # Ліміт категорій cat = item.get('category') max_per_cat = business_rules.get('max_same_category', 5) if category_count.get(cat, 0) >= max_per_cat: continue category_count[cat] = category_count.get(cat, 0) + 1 scored_candidates.append((item_id, score)) return sorted(scored_candidates, key=lambda x: x[1], reverse=True) 

Типові метрики після впровадження рекомендаційної системи: CTR рекомендацій 8-15% (проти 2-4% популярних товарів), конверсія з рекомендації 3-7%, додатковий виторг 12-25% від загального обороту. Час холодного старту для нових користувачів без історії: 3-5 взаємодій для отримання персоналізованих рекомендацій.

Процес роботи, склад та вартість

1. Аналітика: аудит даних, профілювання користувачів, визначення бізнес-цілей.
2. Проектування: вибір архітектури (content-based, collaborative, two-tower), пайплайни навчання.
3. Розробка: реалізація моделей (PyTorch), навчання на історичних даних, налаштування ANN-індексу (FAISS).
4. Інтеграція: REST API, gRPC, контейнеризація (Docker), розгортання в Kubernetes.
5. A/B тестування: метрики (Recall, NDCG, частка виторгу), моніторинг дрейфу даних.
6. Підтримка: 12 місяців гарантії, оновлення моделей, доопрацювання бізнес-правил.

Що входить в роботу

• Архітектурний документ (вибір підходів, обґрунтування)
• Вихідний код моделей та пайплайнів (PyTorch, FAISS)
• Docker-образи та Kubernetes manifests
• REST API / gRPC для real-time інференсу
• Інтеграція з вашою CRM/сайтом
• Документація та навчання команди
• A/B тестування та налаштування моніторингу (Prometheus + Grafana)
• 12 місяців підтримки та оновлень

Терміни: від 30 до 60 робочих днів залежно від обсягу даних та складності інтеграції. Вартість розраховується індивідуально — ми оцінимо ваш проект за 1 день після знайомства із завданням.

Десятки реалізованих проектів — від стартапів до enterprise. Додатковий виторг після впровадження обчислюється мільйонами гривень на місяць. Замовте розробку AI-рекомендаційної системи та отримайте консультацію з архітектури вашого проекту.