1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

import numpy as np
from typing import Optional, List, Set, Any
 
def reciprocal_rank(relevance: List[int]) -> float:
    """
    Compute Reciprocal Rank for a single query.
    
    Args:
        relevance: Binary relevance in ranked order (1=relevant, 0=not)
    
    Returns:
        RR value in [0, 1], or 0 if no relevant items
    """
    for i, rel in enumerate(relevance):
        if rel == 1:
            return 1.0 / (i + 1)  # i is 0-indexed, rank is 1-indexed
    return 0.0
 
def reciprocal_rank_from_items(
    ranked_items: List[Any],
    relevant_items: Set[Any]
) -> float:
    """
    Compute RR given ranked item IDs and relevant item set.
    
    Args:
        ranked_items: Item IDs in ranked order
        relevant_items: Set of relevant item IDs
    
    Returns:
        RR value
    """
    for i, item in enumerate(ranked_items):
        if item in relevant_items:
            return 1.0 / (i + 1)
    return 0.0
 
# Examples
print("Reciprocal Rank Examples")
print("=" * 50)
 
# Case 1: First result is relevant
rr1 = reciprocal_rank([1, 0, 0, 1, 0])
print(f"[1, 0, 0, 1, 0] → RR = {rr1:.4f} (first relevant at rank 1)")
 
# Case 2: Third result is first relevant
rr2 = reciprocal_rank([0, 0, 1, 0, 1])
print(f"[0, 0, 1, 0, 1] → RR = {rr2:.4f} (first relevant at rank 3)")
 
# Case 3: Relevant at position 10
rr3 = reciprocal_rank([0]*9 + [1])
print(f"[0]*9 + [1]    → RR = {rr3:.4f} (first relevant at rank 10)")
 
# Case 4: No relevant items
rr4 = reciprocal_rank([0, 0, 0, 0, 0])
print(f"[0, 0, 0, 0, 0] → RR = {rr4:.4f} (no relevant items)")
 
# Using item IDs
print("\nWith Item IDs:")
ranked = ['doc_a', 'doc_b', 'doc_c', 'doc_d', 'doc_e']
relevant = {'doc_c', 'doc_e'}
rr5 = reciprocal_rank_from_items(ranked, relevant)
print(f"Ranked: {ranked}")
print(f"Relevant: {relevant}")
print(f"RR = {rr5:.4f} (doc_c found at rank 3)")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

import numpy as np
from typing import List
 
def mean_reciprocal_rank(query_relevances: List[List[int]]) -> dict:
    """
    Compute MRR across multiple queries.
    
    Args:
        query_relevances: List of relevance lists (one per query)
    
    Returns:
        Dict with MRR and diagnostics
    """
    rr_values = []
    ranks = []
    
    for relevance in query_relevances:
        rr = 0.0
        rank = float('inf')
        
        for i, rel in enumerate(relevance):
            if rel == 1:
                rank = i + 1
                rr = 1.0 / rank
                break
        
        rr_values.append(rr)
        ranks.append(rank)
    
    return {
        'MRR': np.mean(rr_values),
        'per_query_RR': rr_values,
        'per_query_rank': ranks,
        'queries_without_relevant': sum(1 for r in ranks if r == float('inf')),
        'num_queries': len(query_relevances)
    }
 
# Example: Evaluate a QA system
queries = [
    # Query 1: Perfect - answer at position 1
    [1, 0, 0, 0, 0],
    # Query 2: Answer at position 3
    [0, 0, 1, 0, 1],
    # Query 3: Answer at position 2
    [0, 1, 1, 0, 0],
    # Query 4: Answer at position 5
    [0, 0, 0, 0, 1],
    # Query 5: No answer found
    [0, 0, 0, 0, 0],
]
 
result = mean_reciprocal_rank(queries)
 
print("MRR Computation Example")
print("=" * 60)
print(f"Number of queries: {result['num_queries']}")
print(f"Queries without relevant: {result['queries_without_relevant']}")
print()
 
print(f"{'Query':<10} {'First Relevant Rank':<22} {'RR':<10}")
print("-" * 45)
for i, (rr, rank) in enumerate(zip(result['per_query_RR'], result['per_query_rank'])):
    rank_str = str(rank) if rank != float('inf') else "Not found"
    print(f"Query {i+1:<4} {rank_str:<22} {rr:.4f}")
 
print("-" * 45)
print(f"MRR = {result['MRR']:.4f}")
 
# Breakdown of MRR calculation
print(f"\nMRR = ({' + '.join([f'{rr:.4f}' for rr in result['per_query_RR']])}) / {result['num_queries']}")
print(f"    = {sum(result['per_query_RR']):.4f} / {result['num_queries']}")
print(f"    = {result['MRR']:.4f}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

import numpy as np
 
def compare_metrics(relevance: list) -> dict:
    """Compare MRR with other metrics for a single query."""
    rel = np.array(relevance)
    n = len(rel)
    
    # MRR (from first relevant)
    mrr = 0.0
    for i, r in enumerate(relevance):
        if r > 0:
            mrr = 1.0 / (i + 1)
            break
    
    # Success@k for various k
    success = {k: 1 if np.sum(rel[:k]) > 0 else 0 for k in [1, 3, 5, 10]}
    
    # MAP (treating any rel > 0 as relevant)
    binary = (rel > 0).astype(float)
    total_rel = np.sum(binary)
    if total_rel > 0:
        cumsum = np.cumsum(binary)
        precisions = cumsum / np.arange(1, n + 1)
        ap = np.sum(precisions * binary) / total_rel
    else:
        ap = 0.0
    
    # NDCG (with binary relevance)
    positions = np.arange(1, n + 1)
    discounts = 1.0 / np.log2(positions + 1)
    gains = binary  # Using binary for fair comparison
    dcg = np.sum(gains * discounts)
    ideal_gains = np.sort(gains)[::-1]
    idcg = np.sum(ideal_gains * discounts)
    ndcg = dcg / idcg if idcg > 0 else 0.0
    
    return {
        'MRR': mrr,
        'Success@1': success[1],
        'Success@3': success[3],
        'Success@5': success[5],
        'AP': ap,
        'NDCG': ndcg
    }
 
# Different scenarios
scenarios = [
    {
        'name': 'Perfect: answer at rank 1',
        'relevance': [1, 0, 1, 0, 0, 0, 0, 0, 0, 0]
    },
    {
        'name': 'Good: answer at rank 2',
        'relevance': [0, 1, 0, 1, 0, 0, 0, 0, 0, 0]
    },
    {
        'name': 'Marginal: answer at rank 5',
        'relevance': [0, 0, 0, 0, 1, 1, 0, 0, 0, 0]
    },
    {
        'name': 'Many relevant, first at rank 3',
        'relevance': [0, 0, 1, 1, 1, 1, 1, 0, 0, 0]
    },
    {
        'name': 'One relevant at rank 10',
        'relevance': [0, 0, 0, 0, 0, 0, 0, 0, 0, 1]
    }
]
 
print("Comparison of Ranking Metrics Across Scenarios")
print("=" * 75)
 
for scenario in scenarios:
    metrics = compare_metrics(scenario['relevance'])
    print(f"\n{scenario['name']}")
    print(f"Relevance: {scenario['relevance']}")
    print(f"{'Metric':<15} {'Value':<10}")
    print("-" * 25)
    for metric, value in metrics.items():
        print(f"{metric:<15} {value:.4f}")
        
print("\n" + "=" * 75)
print("Key Observations:")
print("- MRR only cares about the first relevant item's position")
print("- Success@k is binary and loses position information within k")
print("- AP considers all relevant items and their positions")
print("- NDCG uses log discount (gentler than MRR's 1/rank)")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

import numpy as np
from scipy import stats
 
def mrr_with_bootstrap_ci(
    rr_values: list,
    n_bootstrap: int = 10000,
    confidence_level: float = 0.95
) -> dict:
    """
    Compute MRR with bootstrap confidence interval.
    
    Args:
        rr_values: List of RR values (one per query)
        n_bootstrap: Number of bootstrap samples
        confidence_level: Confidence level for CI
    
    Returns:
        Dict with MRR, CI, and statistics
    """
    rr = np.array(rr_values)
    n = len(rr)
    
    # Point estimate
    mrr = np.mean(rr)
    
    # Bootstrap
    rng = np.random.default_rng(42)
    bootstrap_mrrs = []
    
    for _ in range(n_bootstrap):
        sample_indices = rng.choice(n, size=n, replace=True)
        bootstrap_mrr = np.mean(rr[sample_indices])
        bootstrap_mrrs.append(bootstrap_mrr)
    
    bootstrap_mrrs = np.array(bootstrap_mrrs)
    
    # Confidence interval (percentile method)
    alpha = 1 - confidence_level
    ci_lower = np.percentile(bootstrap_mrrs, 100 * alpha / 2)
    ci_upper = np.percentile(bootstrap_mrrs, 100 * (1 - alpha / 2))
    
    # Standard error
    se = np.std(rr, ddof=1) / np.sqrt(n)
    
    return {
        'MRR': mrr,
        'SE': se,
        'CI_lower': ci_lower,
        'CI_upper': ci_upper,
        'Bootstrap_std': np.std(bootstrap_mrrs),
        'n_queries': n
    }
 
def compare_mrr_systems(rr_a: list, rr_b: list) -> dict:
    """
    Compare two systems using paired bootstrap test.
    
    Tests whether system A has significantly different MRR from system B.
    """
    rr_a = np.array(rr_a)
    rr_b = np.array(rr_b)
    
    assert len(rr_a) == len(rr_b), "Must have same queries"
    n = len(rr_a)
    
    # Observed difference
    observed_diff = np.mean(rr_a) - np.mean(rr_b)
    
    # Paired bootstrap
    rng = np.random.default_rng(42)
    n_bootstrap = 10000
    bootstrap_diffs = []
    
    for _ in range(n_bootstrap):
        indices = rng.choice(n, size=n, replace=True)
        diff = np.mean(rr_a[indices]) - np.mean(rr_b[indices])
        bootstrap_diffs.append(diff)
    
    bootstrap_diffs = np.array(bootstrap_diffs)
    
    # CI for difference
    ci_lower = np.percentile(bootstrap_diffs, 2.5)
    ci_upper = np.percentile(bootstrap_diffs, 97.5)
    
    # Significant if CI doesn't contain 0
    significant = ci_lower > 0 or ci_upper < 0
    
    return {
        'MRR_A': np.mean(rr_a),
        'MRR_B': np.mean(rr_b),
        'Difference': observed_diff,
        'CI_difference': (ci_lower, ci_upper),
        'Significant_at_95': significant
    }
 
# Example: Evaluate QA system performance
np.random.seed(42)
 
# Simulate RR values for 100 queries
n_queries = 100
rr_values = []
for _ in range(n_queries):
    # 60% of queries have answer in top 3
    # 30% have answer in positions 4-10
    # 10% have no answer found
    p = np.random.random()
    if p < 0.6:
        rank = np.random.choice([1, 2, 3], p=[0.5, 0.3, 0.2])
    elif p < 0.9:
        rank = np.random.choice(range(4, 11))
    else:
        rank = float('inf')
    
    rr = 1/rank if rank != float('inf') else 0
    rr_values.append(rr)
 
result = mrr_with_bootstrap_ci(rr_values)
 
print("MRR with Bootstrap Confidence Interval")
print("=" * 50)
print(f"Number of queries: {result['n_queries']}")
print(f"MRR: {result['MRR']:.4f}")
print(f"Standard Error: {result['SE']:.4f}")
print(f"95% CI: [{result['CI_lower']:.4f}, {result['CI_upper']:.4f}]")
print(f"Bootstrap Std: {result['Bootstrap_std']:.4f}")
 
# Compare two systems
print("\nComparing Two QA Systems")
print("=" * 50)
# System A is slightly better
rr_system_a = rr_values
# System B is same with some noise
rr_system_b = [max(0, rr - 0.05 + 0.1 * np.random.random()) for rr in rr_values]
 
comparison = compare_mrr_systems(rr_system_a, rr_system_b)
print(f"System A MRR: {comparison['MRR_A']:.4f}")
print(f"System B MRR: {comparison['MRR_B']:.4f}")
print(f"Difference (A - B): {comparison['Difference']:.4f}")
print(f"95% CI for difference: ({comparison['CI_difference'][0]:.4f}, {comparison['CI_difference'][1]:.4f})")
print(f"Significant at 95%: {comparison['Significant_at_95']}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109

import numpy as np
from typing import Callable, List
 
def mrr_at_k(rr_values: List[float], ranks: List[int], k: int) -> float:
    """
    Compute MRR@k: only count RR if first relevant is within top-k.
    
    Args:
        rr_values: Original RR values
        ranks: Rank of first relevant item for each query
        k: Cutoff position
    
    Returns:
        MRR@k value
    """
    truncated_rr = [
        rr if (r <= k and r != float('inf')) else 0.0
        for rr, r in zip(rr_values, ranks)
    ]
    return np.mean(truncated_rr)
 
def generalized_rr(rank: int, discount_fn: Callable[[int], float]) -> float:
    """
    Compute generalized reciprocal rank with custom discount.
    
    Args:
        rank: Rank of first relevant item (inf if none found)
        discount_fn: Function mapping rank to score
    
    Returns:
        Generalized RR value
    """
    if rank == float('inf'):
        return 0.0
    return discount_fn(rank)
 
def expected_reciprocal_rank(relevance: List[float], max_rel: float = 4) -> float:
    """
    Compute ERR (Expected Reciprocal Rank).
    
    Models cascade where user stops when satisfied.
    
    Args:
        relevance: Graded relevance scores in ranked order
        max_rel: Maximum relevance grade (for normalization)
    
    Returns:
        ERR value
    """
    # Convert relevance to satisfaction probability
    # R_i = (2^rel_i - 1) / 2^max_rel
    R = [(2**rel - 1) / (2**max_rel) for rel in relevance]
    
    err = 0.0
    cascade_prob = 1.0  # Probability of reaching this position
    
    for i, r_i in enumerate(R):
        position = i + 1
        # Contribution: (1/rank) * P(reach position) * P(satisfied here)
        err += (1 / position) * cascade_prob * r_i
        # Update cascade probability
        cascade_prob *= (1 - r_i)
    
    return err
 
# Examples
print("MRR Variants Demonstration")
print("=" * 60)
 
# Sample data: ranks of first relevant item for 10 queries
ranks = [1, 3, 2, 15, 5, 1, 8, float('inf'), 2, 6]
rr_values = [1/r if r != float('inf') else 0.0 for r in ranks]
 
print("Query ranks of first relevant:", ranks)
print("RR values:", [f"{rr:.3f}" for rr in rr_values])
print()
 
# Standard MRR
mrr_all = np.mean(rr_values)
print(f"Standard MRR:  {mrr_all:.4f}")
 
# MRR@k for various k
for k in [3, 5, 10]:
    mrr_k = mrr_at_k(rr_values, ranks, k)
    print(f"MRR@{k}:        {mrr_k:.4f}")
 
# Generalized RR with different discounts
print("\nGeneralized RR with Different Discounts (for rank=3):")
print("-" * 50)
rank = 3
 
discounts = {
    'Standard (1/r)': lambda r: 1/r,
    'Log (1/log2(r+1))': lambda r: 1/np.log2(r+1),
    'Sqrt (1/sqrt(r))': lambda r: 1/np.sqrt(r),
    'Exponential (e^(-(r-1)/3))': lambda r: np.exp(-(r-1)/3)
}
 
for name, fn in discounts.items():
    grr = generalized_rr(rank, fn)
    print(f"{name:<30}: {grr:.4f}")
 
# ERR example
print("\nERR (Expected Reciprocal Rank):")
print("-" * 50)
graded_relevance = [3, 1, 4, 0, 2]  # Graded on 0-4 scale
print(f"Graded relevance: {graded_relevance}")
err = expected_reciprocal_rank(graded_relevance, max_rel=4)
print(f"ERR: {err:.4f}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167

import numpy as np
from typing import List, Set, Any, Optional, Dict
from dataclasses import dataclass
 
@dataclass
class RRResult:
    """Container for Reciprocal Rank result."""
    rr: float
    rank_of_first: Optional[int]  # None if no relevant found
    found: bool
 
@dataclass
class MRRResult:
    """Container for MRR computation result."""
    mrr: float
    num_queries: int
    queries_with_result: int
    per_query_rr: List[float]
    per_query_rank: List[Optional[int]]
    ci_lower: float
    ci_upper: float
 
class MRREvaluator:
    """
    Production-grade MRR evaluator with comprehensive features.
    """
    
    def __init__(self, cutoff_k: Optional[int] = None):
        """
        Initialize MRR evaluator.
        
        Args:
            cutoff_k: Maximum rank to consider (None = unlimited)
        """
        self.cutoff_k = cutoff_k
    
    def reciprocal_rank(
        self,
        ranked_items: List[Any],
        relevant_items: Set[Any]
    ) -> RRResult:
        """
        Compute RR for a single query.
        
        Args:
            ranked_items: Items in ranked order
            relevant_items: Set of relevant items
        
        Returns:
            RRResult with RR value and diagnostics
        """
        max_pos = len(ranked_items)
        if self.cutoff_k is not None:
            max_pos = min(max_pos, self.cutoff_k)
        
        for i in range(max_pos):
            if ranked_items[i] in relevant_items:
                rank = i + 1
                return RRResult(
                    rr=1.0 / rank,
                    rank_of_first=rank,
                    found=True
                )
        
        return RRResult(rr=0.0, rank_of_first=None, found=False)
    
    def mean_reciprocal_rank(
        self,
        queries: List[Dict[str, Any]],
        n_bootstrap: int = 10000
    ) -> MRRResult:
        """
        Compute MRR across queries with bootstrap CI.
        
        Args:
            queries: List of dicts with 'ranked_items' and 'relevant_items'
            n_bootstrap: Bootstrap samples for CI
        
        Returns:
            MRRResult with MRR, CI, and per-query values
        """
        rr_results = [
            self.reciprocal_rank(q['ranked_items'], q['relevant_items'])
            for q in queries
        ]
        
        rr_values = np.array([r.rr for r in rr_results])
        ranks = [r.rank_of_first for r in rr_results]
        
        n = len(rr_values)
        mrr = np.mean(rr_values)
        queries_found = sum(1 for r in rr_results if r.found)
        
        # Bootstrap CI
        rng = np.random.default_rng(42)
        bootstrap_mrrs = []
        for _ in range(n_bootstrap):
            indices = rng.choice(n, size=n, replace=True)
            bootstrap_mrrs.append(np.mean(rr_values[indices]))
        
        ci_lower = np.percentile(bootstrap_mrrs, 2.5)
        ci_upper = np.percentile(bootstrap_mrrs, 97.5)
        
        return MRRResult(
            mrr=float(mrr),
            num_queries=n,
            queries_with_result=queries_found,
            per_query_rr=rr_values.tolist(),
            per_query_rank=ranks,
            ci_lower=float(ci_lower),
            ci_upper=float(ci_upper)
        )
 
# Demo: Evaluate a QA system
def demo():
    evaluator = MRREvaluator(cutoff_k=10)  # Only consider top 10
    
    # Simulated QA results
    queries = [
        {
            'question': 'What is the capital of France?',
            'ranked_items': ['Paris', 'Lyon', 'Marseille', 'Nice', 'Bordeaux'],
            'relevant_items': {'Paris'}
        },
        {
            'question': 'Who wrote Romeo and Juliet?',
            'ranked_items': ['Marlowe', 'Shakespeare', 'Jonson', 'Bacon', 'Oxford'],
            'relevant_items': {'Shakespeare'}
        },
        {
            'question': 'What year did WW2 end?',
            'ranked_items': ['1944', '1946', '1943', '1945', '1947'],
            'relevant_items': {'1945'}
        },
        {
            'question': 'Capital of Liechtenstein?',  # Harder question
            'ranked_items': ['Bern', 'Vienna', 'Zurich', 'Munich', 'Vaduz'],
            'relevant_items': {'Vaduz'}
        },
        {
            'question': 'Obscure trivia?',
            'ranked_items': ['wrong1', 'wrong2', 'wrong3', 'wrong4', 'wrong5'],
            'relevant_items': {'correct_answer'}  # Not in list
        }
    ]
    
    result = evaluator.mean_reciprocal_rank(queries)
    
    print("QA System Evaluation with MRR")
    print("=" * 60)
    print(f"{'Query':<25} {'Rank':<10} {'RR':<10}")
    print("-" * 45)
    
    for i, (q, rr, rank) in enumerate(zip(
        queries, result.per_query_rr, result.per_query_rank
    )):
        q_short = q['question'][:22] + '...' if len(q['question']) > 25 else q['question']
        rank_str = str(rank) if rank else 'Not found'
        print(f"{q_short:<25} {rank_str:<10} {rr:.4f}")
    
    print("-" * 45)
    print(f"MRR: {result.mrr:.4f}")
    print(f"95% CI: [{result.ci_lower:.4f}, {result.ci_upper:.4f}]")
    print(f"Queries with answer found: {result.queries_with_result}/{result.num_queries}")
 
if __name__ == "__main__":
    demo()