Ранжування пошуку

Ранжування пошуку визначає, які результати з'являться першими коли користувачі шукають. Це ядро пошукових систем, e-commerce платформ та сервісів відкриття контенту. На відміну від класифікації чи регресії, ранжування оптимізує відносний порядок елементів - отримати найкращі результати на верх важливіше за абсолютні оцінки.

Де використовується

Ранжування пошуку забезпечує відкриття скрізь:

• Веб-пошук: Google, Bing, DuckDuckGo
• E-commerce: Amazon, eBay, Shopify пошук продуктів
• Контент платформи: YouTube, Netflix, Spotify
• Job boards: LinkedIn, Indeed пошук вакансій
• Новини: Ранжування статей, персоналізовані фіди
• Ентерпрайз: Пошук документів, управління знаннями

Пайплайн ранжування

Що ви робите на практиці

1. Визначте релевантність - Що робить результат хорошим?
2. Зберіть тренувальні дані - Кліки, рейтинги, явні мітки
3. Побудуйте feature pipeline - Фічі запиту, документа, взаємодії
4. Оберіть loss function - Pointwise, pairwise, або listwise
5. Натренуйте модель ранжування - LambdaMART, neural rankers
6. Оцініть offline - NDCG, MRR на held-out даних
7. A/B тест online - Метрики залученості користувачів
8. Моніторте та ітеруйте - Обробіть position bias, дрейф

Спочатку базова лінія

Почніть з класичного retrieval перед ML ранжуванням:

import numpy as np
from collections import defaultdict
import math

class BM25:
    """BM25 базове ранжування."""

    def __init__(self, k1=1.5, b=0.75):
        self.k1 = k1
        self.b = b
        self.doc_freqs = defaultdict(int)
        self.doc_lengths = {}
        self.avg_doc_length = 0
        self.corpus_size = 0
        self.index = defaultdict(list)

    def fit(self, documents):
        """Індексування документів для BM25 scoring."""
        self.corpus_size = len(documents)
        total_length = 0

        for doc_id, doc in enumerate(documents):
            tokens = self.tokenize(doc)
            self.doc_lengths[doc_id] = len(tokens)
            total_length += len(tokens)

            # Побудова інвертованого індексу
            unique_tokens = set(tokens)
            for token in unique_tokens:
                self.doc_freqs[token] += 1
                self.index[token].append(doc_id)

        self.avg_doc_length = total_length / self.corpus_size
        self.documents = documents

    def tokenize(self, text):
        """Проста токенізація."""
        return text.lower().split()

    def score(self, query, doc_id):
        """Розрахунок BM25 score для документа."""
        query_tokens = self.tokenize(query)
        doc_tokens = self.tokenize(self.documents[doc_id])
        doc_length = self.doc_lengths[doc_id]

        score = 0
        for token in query_tokens:
            if token not in self.doc_freqs:
                continue

            # Частота терміна в документі
            tf = doc_tokens.count(token)

            # Зворотна частота документа
            df = self.doc_freqs[token]
            idf = math.log((self.corpus_size - df + 0.5) / (df + 0.5) + 1)

            # Формула BM25
            numerator = tf * (self.k1 + 1)
            denominator = tf + self.k1 * (1 - self.b + self.b * doc_length / self.avg_doc_length)

            score += idf * numerator / denominator

        return score

    def search(self, query, top_k=10):
        """Пошук top-k документів для запиту."""
        query_tokens = self.tokenize(query)

        # Отримання документів-кандидатів
        candidates = set()
        for token in query_tokens:
            candidates.update(self.index.get(token, []))

        # Оцінка кандидатів
        scores = [(doc_id, self.score(query, doc_id)) for doc_id in candidates]
        scores.sort(key=lambda x: x[1], reverse=True)

        return scores[:top_k]

class TwoStageRanker:
    """Двоетапний retrieval + re-ranking."""

    def __init__(self, retriever, ranker):
        self.retriever = retriever  # Швидкий перший етап (BM25)
        self.ranker = ranker        # ML re-ranker

    def search(self, query, top_k=10, retrieval_k=100):
        """Двоетапний пошук."""
        # Етап 1: Швидкий retrieval
        candidates = self.retriever.search(query, top_k=retrieval_k)

        if not candidates:
            return []

        # Етап 2: ML re-ranking
        doc_ids = [doc_id for doc_id, _ in candidates]
        features = self.extract_features(query, doc_ids)
        scores = self.ranker.predict(features)

        # Комбінування та ранжування
        results = list(zip(doc_ids, scores))
        results.sort(key=lambda x: x[1], reverse=True)

        return results[:top_k]

Learning to Rank фічі

import numpy as np
from sentence_transformers import SentenceTransformer

class RankingFeatureExtractor:
    """Екстракція фіч для learning to rank."""

    def __init__(self):
        self.text_encoder = SentenceTransformer('all-MiniLM-L6-v2')

    def extract(self, query, document, query_doc_stats=None):
        """Екстракція комплексних фіч ранжування."""
        features = {}

        # Фічі текстового matching
        features.update(self.text_match_features(query, document))

        # Семантична подібність
        features.update(self.semantic_features(query, document))

        # Фічі якості документа
        features.update(self.doc_quality_features(document))

        # Взаємодія запит-документ
        if query_doc_stats:
            features.update(self.interaction_features(query_doc_stats))

        return features

    def text_match_features(self, query, document):
        """Традиційні сигнали текстового matching."""
        query_tokens = set(query.lower().split())
        doc_tokens = document.lower().split()
        doc_token_set = set(doc_tokens)

        # Фічі перекриття
        overlap = len(query_tokens & doc_token_set)

        return {
            'query_term_coverage': overlap / len(query_tokens) if query_tokens else 0,
            'exact_match_count': sum(1 for t in query_tokens if t in doc_token_set),
            'query_in_title': int(query.lower() in document.lower()[:100]),
            'doc_length': len(doc_tokens),
            'doc_unique_terms': len(doc_token_set)
        }

    def semantic_features(self, query, document):
        """Embedding-based семантичні фічі."""
        query_emb = self.text_encoder.encode(query)
        doc_emb = self.text_encoder.encode(document[:512])

        # Косинусна подібність
        similarity = np.dot(query_emb, doc_emb) / (
            np.linalg.norm(query_emb) * np.linalg.norm(doc_emb)
        )

        return {
            'semantic_similarity': float(similarity)
        }

    def doc_quality_features(self, document):
        """Сигнали якості документа."""
        words = document.split()

        return {
            'word_count': len(words),
            'avg_word_length': np.mean([len(w) for w in words]) if words else 0,
            'uppercase_ratio': sum(1 for c in document if c.isupper()) / len(document) if document else 0
        }

    def interaction_features(self, stats):
        """Фічі взаємодії користувача."""
        return {
            'click_count': stats.get('clicks', 0),
            'ctr': stats.get('clicks', 0) / max(stats.get('impressions', 1), 1),
            'avg_dwell_time': stats.get('avg_dwell_time', 0),
            'bounce_rate': stats.get('bounces', 0) / max(stats.get('clicks', 1), 1)
        }

Learning to Rank моделі

import lightgbm as lgb
import numpy as np

class LambdaMART:
    """LambdaMART модель ранжування з LightGBM."""

    def __init__(self, n_estimators=100, learning_rate=0.1):
        self.params = {
            'objective': 'lambdarank',
            'metric': 'ndcg',
            'ndcg_eval_at': [1, 3, 5, 10],
            'num_leaves': 31,
            'learning_rate': learning_rate,
            'n_estimators': n_estimators
        }
        self.model = None

    def fit(self, X_train, y_train, groups_train, 
            X_val=None, y_val=None, groups_val=None):
        """
        Тренування LambdaMART моделі.

        Args:
            X_train: Матриця фіч
            y_train: Мітки релевантності (0-4 типово)
            groups_train: Розміри груп запитів
        """
        train_data = lgb.Dataset(
            X_train, label=y_train, group=groups_train
        )

        valid_sets = [train_data]
        if X_val is not None:
            val_data = lgb.Dataset(
                X_val, label=y_val, group=groups_val
            )
            valid_sets.append(val_data)

        self.model = lgb.train(
            self.params,
            train_data,
            valid_sets=valid_sets,
            num_boost_round=self.params['n_estimators']
        )

    def predict(self, X):
        """Прогнозування scores релевантності."""
        return self.model.predict(X)

    def rank(self, query_features):
        """Ранжування документів для одного запиту."""
        scores = self.predict(query_features)
        ranking = np.argsort(-scores)
        return ranking

Метрики оцінки

import numpy as np

def ndcg_at_k(relevance_scores, k=10):
    """Розрахунок NDCG@k для одного запиту."""
    relevance = np.array(relevance_scores[:k])

    # DCG
    positions = np.arange(1, len(relevance) + 1)
    dcg = np.sum(relevance / np.log2(positions + 1))

    # Ідеальний DCG
    ideal_relevance = np.sort(relevance_scores)[::-1][:k]
    ideal_positions = np.arange(1, len(ideal_relevance) + 1)
    idcg = np.sum(ideal_relevance / np.log2(ideal_positions + 1))

    return dcg / idcg if idcg > 0 else 0

def mrr(rankings):
    """Розрахунок Mean Reciprocal Rank."""
    reciprocal_ranks = []

    for ranking in rankings:
        # Пошук першого релевантного результату
        for i, is_relevant in enumerate(ranking):
            if is_relevant:
                reciprocal_ranks.append(1 / (i + 1))
                break
        else:
            reciprocal_ranks.append(0)

    return np.mean(reciprocal_ranks)

Обробка Position Bias

class PositionBiasCorrector:
    """Обробка position bias в даних кліків."""

    def __init__(self, num_positions=10):
        self.num_positions = num_positions
        self.position_propensities = None

    def estimate_propensities(self, click_logs):
        """Оцінка position bias з рандомізованих даних."""
        position_clicks = [0] * self.num_positions
        position_impressions = [0] * self.num_positions

        for log in click_logs:
            position = log['position']
            if position < self.num_positions:
                position_impressions[position] += 1
                if log['clicked']:
                    position_clicks[position] += 1

        # Propensity = P(examine | position)
        self.position_propensities = [
            clicks / max(imps, 1) 
            for clicks, imps in zip(position_clicks, position_impressions)
        ]

        # Нормалізація так щоб позиція 0 мала propensity 1
        max_prop = max(self.position_propensities)
        self.position_propensities = [
            p / max_prop for p in self.position_propensities
        ]

    def debias_clicks(self, clicks, positions):
        """Застосування inverse propensity weighting."""
        debiased = []

        for click, pos in zip(clicks, positions):
            if pos < self.num_positions:
                propensity = self.position_propensities[pos]
                debiased.append(click / max(propensity, 0.01))
            else:
                debiased.append(click)

        return debiased

Типові помилки та чек-лист

• [ ] Position bias: Тренування на biased даних кліків
• [ ] Feature leakage: Використання target info у фічах
• [ ] Metric mismatch: Оптимізація неправильної цілі
• [ ] Cold start: Нові елементи/запити без сигналів
• [ ] Popularity bias: Завжди ранжування популярних елементів
• [ ] Latency issues: Модель занадто повільна для real-time
• [ ] Freshness: Застарілі фічі або модель

Чек-лист для продакшену

Перед деплоєм: - [ ] Offline оцінка на held-out даних - [ ] A/B тест фреймворк готовий - [ ] Бюджет латентності встановлено (< 100ms типово) - [ ] Feature computation pipeline протестовано - [ ] Fallback стратегія ранжування

Після деплою: - [ ] Моніторинг NDCG, CTR, engagement - [ ] Відстеження метрик position bias - [ ] Регулярне перенавчання моделі - [ ] Виявлення дрейфу фіч - [ ] Interleaving експерименти

Інструменти та стек

Приклад міні-кейсу

Сценарій: Ранжування пошуку вакансій для кар'єрної платформи.

Фічі: - Query-job text match (BM25, семантичний) - Свіжість вакансії (днів з публікації) - Match зарплати (якщо вказано) - Відстань локації - Історичний CTR для типу вакансії

Результати: - 18% покращення application rate - 25% збільшення NDCG@10 - Зменшення position bias на 40% з debiasing - Латентність: 45ms p99

Пов'язані сторінки

• Індекс: ML Field Guide
• Learning to Rank
• Розуміння запитів
• Семантичний пошук

Додаткові матеріали

• Introduction to Information Retrieval
• Learning to Rank for Information Retrieval
• TensorFlow Ranking

Спробуйте наживо

Дослідіть пов'язану модель на практиці: відкрийте Retrieval Evaluation and Reranking Lab і змінюйте параметри під час роботи. Усе лишається у вашому браузері.