ML System Design: Поисковая выдача

ML System Design: Поисковая выдача (Search Ranking)

1. Архитектура: Retrieval → Ranking

Двухэтапная архитектура:

User Query
    ↓
[RETRIEVAL STAGE]  ← Быстро, низкая точность
- ElasticSearch / Solr
- BM25 ranking
- ~1000 документов

    ↓
[RANKING STAGE]    ← Медленнее, высокая точность
- ML модель (Learning to Rank)
- ~100 документов (top k)

    ↓
Final Results (10 документов)

Почему два этапа?

• Retrieval: скорость важнее всего (нельзя проверять миллионы документов)
• Ranking: точность важна (проверяем top-k кандидатов)

2. Feature Engineering

Query Features (зависят только от запроса):

def extract_query_features(query: str):
    """Признаки запроса"""
    features = {}
    
    # Базовые
    features['query_length'] = len(query.split())
    features['query_char_length'] = len(query)
    features['has_special_chars'] = any(c in query for c in '!?')
    
    # Лингвистические
    features['is_question'] = query.endswith('?')
    features['is_brand_query'] = query in popular_brands  # Список популярных брендов
    features['is_navigational'] = 'near me' in query.lower()
    
    # Временные
    features['is_trending'] = query in trending_queries
    features['seasonality'] = get_seasonality_score(query)  # Летний запрос?
    
    # Разнообразие
    features['num_entities'] = extract_named_entities(query)
    
    return features

Document Features:

def extract_document_features(doc: dict):
    """Признаки документа"""
    features = {}
    
    # Популярность
    features['url_clicks'] = doc['total_clicks']
    features['url_impressions'] = doc['total_impressions']
    features['ctr'] = doc['clicks'] / max(doc['impressions'], 1)
    
    # Качество контента
    features['content_length'] = len(doc['text'])
    features['title_length'] = len(doc['title'])
    features['num_images'] = doc['image_count']
    features['num_links'] = doc['link_count']
    features['has_video'] = 1 if doc['video'] else 0
    
    # Свежесть
    from datetime import datetime
    days_old = (datetime.now() - doc['published_date']).days
    features['freshness'] = 1.0 / (1.0 + days_old)  # Exponential decay
    
    # SEO
    features['page_rank'] = doc['pagerank']
    features['domain_authority'] = doc['domain_authority']
    features['has_ssl'] = 1 if doc['url'].startswith('https') else 0
    
    # Mobile-friendliness
    features['is_mobile_friendly'] = doc['mobile_score']
    features['page_speed'] = doc['page_speed_score']
    
    return features

Cross Features (взаимодействие query и document):

def extract_cross_features(query: str, doc: dict, 
                          query_features: dict, doc_features: dict):
    """Признаки взаимодействия запроса и документа"""
    features = {}
    
    # BM25 score (из retrieval этапа)
    features['bm25_score'] = compute_bm25(query, doc['text'])
    
    # TF-IDF косинусное сходство
    from sklearn.feature_extraction.text import TfidfVectorizer
    vectorizer = TfidfVectorizer()
    query_vec = vectorizer.fit_transform([query])
    doc_vec = vectorizer.transform([doc['text']])
    features['tfidf_similarity'] = cosine_similarity(query_vec, doc_vec)[0][0]
    
    # Семантическое сходство (embedding-based)
    query_embedding = get_query_embedding(query)  # BERT/Dense passage retrieval
    doc_embedding = get_document_embedding(doc)
    features['semantic_similarity'] = cosine_similarity(query_embedding, doc_embedding)
    
    # Точное совпадение слов
    query_tokens = set(query.lower().split())
    doc_tokens = set(doc['text'].lower().split())
    features['exact_match_ratio'] = len(query_tokens & doc_tokens) / len(query_tokens)
    
    # Совпадение именованных сущностей
    query_entities = extract_named_entities(query)
    doc_entities = extract_named_entities(doc['text'])
    features['entity_overlap'] = len(query_entities & doc_entities) / len(query_entities)
    
    # Позиция совпадающего слова в заголовке
    title_words = doc['title'].lower().split()
    query_word_in_title_position = [i for i, word in enumerate(title_words) 
                                     if word in query_tokens]
    features['first_word_match_position'] = min(query_word_in_title_position) if query_word_in_title_position else 999
    
    # Query-document freshness match
    features['freshness_match'] = 1.0 if query_features['is_trending'] and doc_features['freshness'] > 0.8 else 0.0
    
    return features

3. Сбор обучающих данных (Ground Truth)

Метод 1: Implicit feedback (Clicks)

# Предполагаем, что кликнутые результаты релевантны
# Минусы: bias (позиция документа влияет на клики)

def create_training_data_from_clicks(search_logs):
    training_pairs = []
    
    for session in search_logs:
        query = session['query']
        clicked_doc_id = session['clicked_document_id']
        impressions = session['serp']  # Документы в выдаче
        
        # Clicked документ = positive
        for doc in impressions:
            if doc['id'] == clicked_doc_id:
                label = 1
            else:
                label = 0
            
            training_pairs.append({
                'query': query,
                'document': doc,
                'label': label
            })
    
    return training_pairs

Метод 2: Explicit feedback (Ratings)

# Асессоры вручную оценивают релевантность (1-5 звёзд)
# Дорого, но качественнее

def create_training_data_from_ratings(assessor_ratings):
    training_pairs = []
    
    for rating in assessor_ratings:
        training_pairs.append({
            'query': rating['query'],
            'document': rating['document'],
            'label': rating['relevance_score']  # 1-5
        })
    
    return training_pairs

Метод 3: Pairwise learning (Pairwise comparisons)

# Асессор выбирает лучший документ из двух
# Более надёжно чем классификация

def create_pairwise_data(assessor_comparisons):
    training_pairs = []
    
    for comparison in assessor_comparisons:
        query = comparison['query']
        winner_doc = comparison['winner']  # Документ, который лучше
        loser_doc = comparison['loser']
        
        training_pairs.append({
            'query': query,
            'doc_better': winner_doc,
            'doc_worse': loser_doc,
            'label': 1  # doc_better > doc_worse
        })
    
    return training_pairs

4. Модели ранжирования (Learning to Rank)

Вариант A: Pointwise (классификация)

# Предсказываем релевантность каждого документа независимо

from xgboost import XGBClassifier
from sklearn.ensemble import GradientBoostingClassifier

model = XGBClassifier(
    max_depth=8,
    learning_rate=0.1,
    n_estimators=100,
    objective='binary:logistic'
)

# X: все признаки (query + document + cross features)
# y: релевантность (0 или 1)
model.fit(X_train, y_train)

# Для каждого документа предсказываем P(relevant)
scores = model.predict_proba(X_test)[:, 1]

Минусы: игнорирует порядок в выдаче

Вариант B: Pairwise (LambdaRank)

# Оптимизируем так, чтобы лучшие документы были выше

from lightgbm import LGBMRanker

model = LGBMRanker(
    objective='rank_xendcg',  # или 'rank_ndcg'
    metric='ndcg',
    num_leaves=31,
    learning_rate=0.05,
    num_rounds=100
)

# X: признаки
# y: релевантность
# group: количество документов на один запрос
model.fit(X_train, y_train, 
         group=group_sizes_train,
         eval_set=[(X_val, y_val)],
         eval_group=[group_sizes_val])

Вариант C: Listwise (XGBoost-NDCG или BERT for ranking)

# Оптимизируем NDCG метрику непосредственно

# XGBoost с NDCG objective
from xgboost import XGBRanker

model = XGBRanker(
    objective='rank:ndcg',
    metric='ndcg',
    max_depth=8,
    learning_rate=0.05
)

model.fit(X_train, y_train, group=group_sizes_train)

Вариант D: Neural Ranking (BERT-based)

# Fine-tune BERT для ranking

from transformers import BertForSequenceClassification, Trainer, TrainingArguments
import torch

model = BertForSequenceClassification.from_pretrained('bert-base-uncased', num_labels=2)

training_args = TrainingArguments(
    output_dir='./results',
    num_train_epochs=3,
    per_device_train_batch_size=32,
    per_device_eval_batch_size=32,
    warmup_steps=500,
    weight_decay=0.01,
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=eval_dataset,
)

trainer.train()

Рекомендация: LightGBM с rank:ndcg для production (скорость + качество)

5. Метрики оценки

Ranking Metrics:

from sklearn.metrics import ndcg_score, dcg_score

# NDCG@10 (Normalized Discounted Cumulative Gain)
# Идеально: кликнутый документ на позиции 1
# Хорошо: кликнутый документ на позиции 3
# Плохо: кликнутый документ на позиции 10

ndcg_10 = ndcg_score(y_true, y_score, k=10)

# MAP@10 (Mean Average Precision)
def compute_map(relevances, k=10):
    """Average Precision учитывает позицию релевантных документов"""
    ap = 0.0
    num_relevant = 0
    
    for i in range(min(k, len(relevances))):
        if relevances[i] > 0:
            num_relevant += 1
            ap += num_relevant / (i + 1)
    
    return ap / max(num_relevant, 1)

# MRR (Mean Reciprocal Rank)
def compute_mrr(relevances):
    """Обратный ранк первого релевантного документа"""
    for i, rel in enumerate(relevances):
        if rel > 0:
            return 1.0 / (i + 1)
    return 0.0

# CTR@k
ctr_10 = clicks_for_top_10 / total_impressions_top_10

Business Metrics:

# Session-level метрики
session_success_rate = sessions_with_clicks / total_sessions
zero_click_rate = sessions_without_clicks / total_sessions
avg_position_of_click = sum(positions) / len(positions)

# Diversity
query_diversity = len(unique_domains_in_top_10) / 10

# User satisfaction
user_satisfaction = sessions_with_multiple_clicks / sessions_with_any_clicks

6. A/B Тестирование

import numpy as np
from scipy import stats

def run_ab_test(control_metrics, variant_metrics, metric_name='ndcg_10'):
    """Проверяем, значимо ли улучшение"""
    
    control_values = control_metrics[metric_name]
    variant_values = variant_metrics[metric_name]
    
    # T-тест
    t_stat, p_value = stats.ttest_ind(control_values, variant_values)
    
    mean_diff = variant_values.mean() - control_values.mean()
    percent_change = (mean_diff / control_values.mean()) * 100
    
    print(f"Metric: {metric_name}")
    print(f"Control: {control_values.mean():.4f}")
    print(f"Variant: {variant_values.mean():.4f}")
    print(f"Change: {percent_change:.2f}%")
    print(f"P-value: {p_value:.4f}")
    print(f"Statistically Significant: {p_value < 0.05}")
    
    return p_value < 0.05


# Пример: тестируем новую модель ранжирования

# Направляем 10% трафика на новую модель (99% получают старую)
# Сравниваем метрики

results = {
    'control': {
        'ndcg_10': control_ndcg_10_values,
        'ctr': control_ctr_values,
        'sessions_with_clicks': control_session_success,
    },
    'variant': {
        'ndcg_10': variant_ndcg_10_values,
        'ctr': variant_ctr_values,
        'sessions_with_clicks': variant_session_success,
    }
}

# Рекомендация по внедрению
for metric in ['ndcg_10', 'ctr', 'sessions_with_clicks']:
    is_significant = run_ab_test(results['control'], results['variant'], metric)
    if is_significant:
        print(f"✓ {metric} улучшилось значимо")
    else:
        print(f"✗ {metric} не улучшилось")

7. Production Pipeline

import redis
import json
from datetime import datetime

class RankingService:
    def __init__(self, model_path, feature_processor):
        self.model = load_model(model_path)
        self.feature_processor = feature_processor
        self.redis_client = redis.Redis()
    
    def rank_documents(self, query: str, candidates: list) -> list:
        """Ранжирование документов"""
        
        # 1. Кешируем результаты известных запросов
        cache_key = f"rank:{query}"
        cached = self.redis_client.get(cache_key)
        if cached:
            return json.loads(cached)
        
        # 2. Проектируем признаки для всех кандидатов
        query_features = extract_query_features(query)
        
        X_candidates = []
        for doc in candidates:
            doc_features = extract_document_features(doc)
            cross_features = extract_cross_features(query, doc, query_features, doc_features)
            
            all_features = {**query_features, **doc_features, **cross_features}
            X_candidates.append(list(all_features.values()))
        
        # 3. Предсказываем релевантность
        scores = self.model.predict(np.array(X_candidates))
        
        # 4. Сортируем по scores
        ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
        
        # 5. Кешируем результат на 1 час
        self.redis_client.setex(cache_key, 3600, json.dumps(ranked))
        
        return ranked[:10]  # Top 10
    
    def log_feedback(self, query: str, clicked_doc_id: str, position: int):
        """Логируем клики для переобучения"""
        log_entry = {
            'query': query,
            'clicked_doc_id': clicked_doc_id,
            'position': position,
            'timestamp': datetime.now().isoformat()
        }
        self.redis_client.rpush('ranking_feedback', json.dumps(log_entry))

8. Continuous Improvement

# Каждый день:
# 1. Берём feedback из логов (clicks)
# 2. Переобучаем модель на новых данных
# 3. Оцениваем на валидационной выборке
# 4. Если метрики улучшились → A/B тест
# 5. Если A/B тест успешен → deploy новой версии

def daily_model_improvement():
    # Получаем feedback
    clicks_today = fetch_feedback_logs(days=1)
    
    # Преобразуем в training data
    train_data = create_training_data_from_clicks(clicks_today)
    
    # Переобучаем
    new_model = train_ranking_model(train_data)
    
    # Оцениваем
    metrics_new = evaluate_model(new_model, validation_set)
    metrics_old = evaluate_model(current_model, validation_set)
    
    # Решаем
    if metrics_new['ndcg_10'] > metrics_old['ndcg_10'] * 1.01:  # 1% улучшение
        create_ab_test(new_model)
    else:
        log_warning("New model not better")

Итоговая архитектура

1. Retrieval: BM25 / Dense retrieval (~1000 documents)
2. Ranking: LightGBM/XGBoost with learning to rank (~100 → 10 documents)
3. Features: Query + Document + Cross features (200+ признаков)
4. Training data: Implicit (clicks) + Explicit (ratings) feedback
5. Metrics: NDCG@10, MAP@10, MRR + Business KPIs
6. A/B test: 10% трафика на новую модель перед полным деплойментом
7. Continuous learning: Daily retraining на новых кликах

Key Success Factors:

• Качественные training data (асессоры)
• Правильные метрики (NDCG>CTR)
• Достаточный трафик для A/B тестирования
• Мониторинг качества в production
• Быстрая итерация (новая версия каждую неделю)