Search Ranking

Search ranking determines which results appear first when users search. It's the core of search engines, e-commerce platforms, and content discovery. Unlike classification or regression, ranking optimizes the relative order of items - getting the best results to the top matters more than absolute scores.

Where It Is Used

Search ranking powers discovery everywhere:

• Web search: Google, Bing, DuckDuckGo
• E-commerce: Amazon, eBay, Shopify product search
• Content platforms: YouTube, Netflix, Spotify
• Job boards: LinkedIn, Indeed job search
• News: Article ranking, personalized feeds
• Enterprise: Document search, knowledge management

Ranking Pipeline

What You Do in Practice

1. Define relevance - What makes a result good?
2. Collect training data - Clicks, ratings, explicit labels
3. Build feature pipeline - Query, document, interaction features
4. Choose loss function - Pointwise, pairwise, or listwise
5. Train ranking model - LambdaMART, neural rankers
6. Evaluate offline - NDCG, MRR on held-out data
7. A/B test online - User engagement metrics
8. Monitor and iterate - Handle position bias, drift

Baseline First

Start with classic retrieval before ML ranking:

import numpy as np
from collections import defaultdict
import math

class BM25:
    """BM25 baseline ranking."""

    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):
        """Index documents for 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)

            # Build inverted index
            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):
        """Simple tokenization."""
        return text.lower().split()

    def score(self, query, doc_id):
        """Calculate BM25 score for a document."""
        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

            # Term frequency in document
            tf = doc_tokens.count(token)

            # Inverse document frequency
            df = self.doc_freqs[token]
            idf = math.log((self.corpus_size - df + 0.5) / (df + 0.5) + 1)

            # BM25 formula
            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):
        """Retrieve top-k documents for query."""
        query_tokens = self.tokenize(query)

        # Get candidate documents
        candidates = set()
        for token in query_tokens:
            candidates.update(self.index.get(token, []))

        # Score candidates
        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:
    """Two-stage retrieval + re-ranking."""

    def __init__(self, retriever, ranker):
        self.retriever = retriever  # Fast first stage (BM25)
        self.ranker = ranker        # ML re-ranker

    def search(self, query, top_k=10, retrieval_k=100):
        """Two-stage search."""
        # Stage 1: Fast retrieval
        candidates = self.retriever.search(query, top_k=retrieval_k)

        if not candidates:
            return []

        # Stage 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)

        # Combine and rank
        results = list(zip(doc_ids, scores))
        results.sort(key=lambda x: x[1], reverse=True)

        return results[:top_k]

    def extract_features(self, query, doc_ids):
        """Extract features for ranking model."""
        features = []
        for doc_id in doc_ids:
            f = {
                'bm25_score': self.retriever.score(query, doc_id),
                'query_length': len(query.split()),
                'doc_length': self.retriever.doc_lengths[doc_id],
                # Add more features
            }
            features.append(f)
        return features

Learning to Rank Features

import numpy as np
from sentence_transformers import SentenceTransformer

class RankingFeatureExtractor:
    """Extract features for learning to rank."""

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

    def extract(self, query, document, query_doc_stats=None):
        """Extract comprehensive ranking features."""
        features = {}

        # Text matching features
        features.update(self.text_match_features(query, document))

        # Semantic similarity
        features.update(self.semantic_features(query, document))

        # Document quality features
        features.update(self.doc_quality_features(document))

        # Query-document interaction
        if query_doc_stats:
            features.update(self.interaction_features(query_doc_stats))

        return features

    def text_match_features(self, query, document):
        """Traditional text matching signals."""
        query_tokens = set(query.lower().split())
        doc_tokens = document.lower().split()
        doc_token_set = set(doc_tokens)

        # Overlap features
        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 semantic features."""
        query_emb = self.text_encoder.encode(query)
        doc_emb = self.text_encoder.encode(document[:512])

        # Cosine similarity
        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):
        """Document quality signals."""
        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):
        """User interaction features."""
        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 Models

import lightgbm as lgb
import numpy as np

class LambdaMART:
    """LambdaMART ranking model using 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):
        """
        Train LambdaMART model.

        Args:
            X_train: Feature matrix
            y_train: Relevance labels (0-4 typically)
            groups_train: Query group sizes
        """
        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):
        """Predict relevance scores."""
        return self.model.predict(X)

    def rank(self, query_features):
        """Rank documents for a single query."""
        scores = self.predict(query_features)
        ranking = np.argsort(-scores)
        return ranking

class NeuralRanker:
    """Neural network for ranking."""

    def __init__(self, input_dim, hidden_dims=[128, 64]):
        import torch
        import torch.nn as nn

        layers = []
        prev_dim = input_dim

        for dim in hidden_dims:
            layers.extend([
                nn.Linear(prev_dim, dim),
                nn.ReLU(),
                nn.Dropout(0.2)
            ])
            prev_dim = dim

        layers.append(nn.Linear(prev_dim, 1))

        self.model = nn.Sequential(*layers)
        self.optimizer = torch.optim.Adam(self.model.parameters(), lr=0.001)

    def pairwise_loss(self, pos_scores, neg_scores, margin=1.0):
        """Pairwise ranking loss."""
        import torch
        return torch.mean(torch.clamp(margin - pos_scores + neg_scores, min=0))

    def train_step(self, pos_features, neg_features):
        """Single training step with pairwise loss."""
        import torch

        self.optimizer.zero_grad()

        pos_scores = self.model(torch.FloatTensor(pos_features))
        neg_scores = self.model(torch.FloatTensor(neg_features))

        loss = self.pairwise_loss(pos_scores, neg_scores)
        loss.backward()
        self.optimizer.step()

        return loss.item()

Evaluation Metrics

import numpy as np

def ndcg_at_k(relevance_scores, k=10):
    """Calculate NDCG@k for a single query."""
    relevance = np.array(relevance_scores[:k])

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

    # Ideal 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):
    """Calculate Mean Reciprocal Rank."""
    reciprocal_ranks = []

    for ranking in rankings:
        # Find first relevant result
        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)

def evaluate_ranking(model, test_queries, test_docs, relevance_labels):
    """Comprehensive ranking evaluation."""
    ndcg_scores = []
    mrr_scores = []

    for query, docs, labels in zip(test_queries, test_docs, relevance_labels):
        # Get model ranking
        scores = model.predict(query, docs)
        ranking = np.argsort(-scores)

        # Reorder labels by model ranking
        ranked_labels = [labels[i] for i in ranking]

        ndcg_scores.append(ndcg_at_k(ranked_labels, k=10))
        mrr_scores.append(1 / (ranking.tolist().index(labels.index(max(labels))) + 1) if max(labels) > 0 else 0)

    return {
        'ndcg@10': np.mean(ndcg_scores),
        'mrr': np.mean(mrr_scores)
    }

Position Bias Handling

class PositionBiasCorrector:
    """Handle position bias in click data."""

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

    def estimate_propensities(self, click_logs):
        """Estimate position bias from randomized data."""
        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)
        ]

        # Normalize so position 0 has 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):
        """Apply 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

Failure Modes and Debugging Checklist

• [ ] Position bias: Training on biased click data
• [ ] Feature leakage: Using target info in features
• [ ] Metric mismatch: Optimizing wrong objective
• [ ] Cold start: New items/queries with no signals
• [ ] Popularity bias: Always ranking popular items
• [ ] Latency issues: Model too slow for real-time
• [ ] Freshness: Stale features or model

Production Checklist

Before deployment: - [ ] Offline evaluation on held-out data - [ ] A/B test framework ready - [ ] Latency budget established (< 100ms typical) - [ ] Feature computation pipeline tested - [ ] Fallback ranking strategy

After deployment: - [ ] Monitor NDCG, CTR, engagement - [ ] Track position bias metrics - [ ] Regular model retraining - [ ] Feature drift detection - [ ] Interleaving experiments

Tools and Stack

Example Mini-Case

Scenario: Job search ranking for career platform.

Features: - Query-job text match (BM25, semantic) - Job freshness (days since posted) - Salary match (if specified) - Location distance - Historical CTR for job type

Results: - 18% improvement in application rate - 25% increase in NDCG@10 - Reduced position bias by 40% with debiasing - Latency: 45ms p99

Related Pages

• Index: ML Field Guide
• Learning to Rank
• Query Understanding
• Semantic Search

Further Reading

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

Try it live

Everything above runs in your browser — open Retrieval Evaluation and Reranking Lab and change the parameters while it is running. Nothing is installed, nothing is uploaded, the whole model lives in one tab.