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-рекомендаційної системи та отримайте консультацію з архітектури вашого проекту.

Розробка рекомендаційних систем: від 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.