Измерение влияния переранжирования

Зачем измерять влияние повторного ранжирования

Создание эталонного набора тестов

golden_test_set = [
    {
        'query': 'How does pgvector HNSW indexing improve search speed?',
        'relevant_doc_ids': ['doc_042', 'doc_107'],
    },
    {
        'query': 'What is the difference between BM25 and dense retrieval?',
        'relevant_doc_ids': ['doc_015'],
    },
    {
        'query': 'How to implement reciprocal rank fusion in Python?',
        'relevant_doc_ids': ['doc_093', 'doc_094'],
    },
    # ... 47 more entries
]

print(f'Test set size: {len(golden_test_set)} queries')
print(f'Avg relevant docs per query: {sum(len(e["relevant_doc_ids"]) for e in golden_test_set) / len(golden_test_set):.1f}')

Метрики поиска: NDCG, MRR, доля попаданий

def compute_retrieval_metrics(results: list[str], relevant_ids: set, k: int = 5):
    results_at_k = results[:k]
    relevant_found = [r for r in results_at_k if r in relevant_ids]

    # Hit rate
    hit = 1 if relevant_found else 0

    # MRR
    rr = 0
    for i, doc_id in enumerate(results_at_k, start=1):
        if doc_id in relevant_ids:
            rr = 1.0 / i
            break

    # NDCG (binary relevance)
    import math
    dcg = sum(
        1.0 / math.log2(i + 1)
        for i, doc_id in enumerate(results_at_k, start=1)
        if doc_id in relevant_ids
    )
    ideal = sum(1.0 / math.log2(i + 1) for i in range(1, min(len(relevant_ids), k) + 1))
    ndcg = dcg / ideal if ideal > 0 else 0

    return {'hit': hit, 'rr': rr, 'ndcg': ndcg}

Базовый вариант: одноэтапный плотный поиск

def evaluate_pipeline(retriever_fn, test_set: list[dict], k: int = 5) -> dict:
    all_metrics = []

    for entry in test_set:
        query = entry['query']
        relevant = set(entry['relevant_doc_ids'])

        results = retriever_fn(query, top_k=k)
        result_ids = [r['id'] for r in results]

        metrics = compute_retrieval_metrics(result_ids, relevant, k)
        all_metrics.append(metrics)

    n = len(all_metrics)
    return {
        f'hit_rate@{k}': sum(m['hit'] for m in all_metrics) / n,
        f'mrr@{k}': sum(m['rr'] for m in all_metrics) / n,
        f'ndcg@{k}': sum(m['ndcg'] for m in all_metrics) / n,
    }

Проведение сравнительного тестирования до и после изменений

import time

def evaluate_with_latency(retriever_fn, test_set, k=5):
    metrics_list = []
    latencies = []

    for entry in test_set:
        t0 = time.perf_counter()
        results = retriever_fn(entry['query'], top_k=k)
        latencies.append((time.perf_counter() - t0) * 1000)

        result_ids = [r['id'] for r in results]
        metrics_list.append(compute_retrieval_metrics(
            result_ids, set(entry['relevant_doc_ids']), k
        ))

    n = len(metrics_list)
    return {
        f'hit_rate@{k}': sum(m['hit'] for m in metrics_list) / n,
        f'ndcg@{k}': sum(m['ndcg'] for m in metrics_list) / n,
        'p50_latency_ms': sorted(latencies)[n // 2],
        'p99_latency_ms': sorted(latencies)[int(n * 0.99)],
    }

baseline = evaluate_with_latency(dense_retrieval_fn, golden_test_set)
two_stage = evaluate_with_latency(two_stage_fn, golden_test_set)
print('Baseline:', baseline)
print('Two-stage:', two_stage)

Интерпретация улучшений NDCG

# Interpreting benchmark results

example_results = {
    'baseline': {'hit_rate@5': 0.78, 'ndcg@5': 0.64, 'p99_latency_ms': 35},
    'two_stage': {'hit_rate@5': 0.89, 'ndcg@5': 0.77, 'p99_latency_ms': 287},
}

delta_ndcg = example_results['two_stage']['ndcg@5'] - example_results['baseline']['ndcg@5']
delta_latency = example_results['two_stage']['p99_latency_ms'] - example_results['baseline']['p99_latency_ms']

print(f'NDCG improvement: +{delta_ndcg:.2f} (+{delta_ndcg/example_results["baseline"]["ndcg@5"]*100:.0f}%)')
print(f'Latency increase: +{delta_latency}ms')
# NDCG improvement: +0.13 (+20%) — clearly worth the 252ms latency cost

Измерение качества ответов по всему процессу

from openai import OpenAI

client = OpenAI()

JUDGE_PROMPT = '''
Rate the following answer on a scale of 1-5 for correctness and faithfulness to the context.

Question: {question}
Context: {context}
Answer: {answer}
Ground truth: {ground_truth}

Return a JSON with fields: {"correctness": int, "faithfulness": int, "explanation": str}
'''

def judge_answer(question, context, answer, ground_truth):
    prompt = JUDGE_PROMPT.format(
        question=question, context=context,
        answer=answer, ground_truth=ground_truth,
    )
    response = client.chat.completions.create(
        model='gpt-4o',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'},
    )
    import json
    return json.loads(response.choices[0].message.content)

Стратифицированный анализ по типу запроса

def stratified_eval(retriever_fn, test_set, k=5):
    groups = {'factual': [], 'procedural': [], 'conceptual': [], 'technical': []}

    for entry in test_set:
        q = entry['query'].lower()
        if any(w in q for w in ['how to', 'how do', 'steps to']):
            groups['procedural'].append(entry)
        elif any(w in q for w in ['why', 'explain', 'what is the reason']):
            groups['conceptual'].append(entry)
        elif any(c.isupper() for c in q.split()) or 'error' in q:
            groups['technical'].append(entry)
        else:
            groups['factual'].append(entry)

    for group_name, group_entries in groups.items():
        if group_entries:
            metrics = evaluate_pipeline(retriever_fn, group_entries, k)
            print(f'{group_name} ({len(group_entries)} queries): ndcg@{k}={metrics[f"ndcg@{k}"]:.3f}')

Регрессионное тестирование с интеграцией в CI

# pytest integration for retrieval quality gates
import pytest

MIN_NDCG_5 = 0.70
MIN_HIT_RATE_5 = 0.85

def test_retrieval_quality_meets_threshold():
    metrics = evaluate_pipeline(production_retriever_fn, golden_test_set, k=5)
    assert metrics['ndcg@5'] >= MIN_NDCG_5, (
        f'NDCG@5 {metrics["ndcg@5"]:.3f} below threshold {MIN_NDCG_5}'
    )
    assert metrics['hit_rate@5'] >= MIN_HIT_RATE_5, (
        f'Hit rate {metrics["hit_rate@5"]:.3f} below threshold {MIN_HIT_RATE_5}'
    )

# Run with: pytest tests/test_retrieval.py -v

Визуализация метрик поиска

def print_comparison_table(results: dict[str, dict]):
    headers = ['Pipeline', 'NDCG@5', 'MRR@5', 'Hit@5', 'P99 ms']
    print('|'.join(f'{h:20}' for h in headers))
    print('-' * (len(headers) * 21))
    for pipeline_name, metrics in results.items():
        row = [
            pipeline_name,
            f'{metrics.get("ndcg@5", 0):.3f}',
            f'{metrics.get("mrr@5", 0):.3f}',
            f'{metrics.get("hit_rate@5", 0):.3f}',
            f'{metrics.get("p99_latency_ms", 0):.0f}',
        ]
        print('|'.join(f'{v:20}' for v in row))

results = {
    'Dense only': {'ndcg@5': 0.64, 'mrr@5': 0.68, 'hit_rate@5': 0.78, 'p99_latency_ms': 35},
    'Hybrid RRF': {'ndcg@5': 0.71, 'mrr@5': 0.74, 'hit_rate@5': 0.84, 'p99_latency_ms': 55},
    'Hybrid + Rerank': {'ndcg@5': 0.77, 'mrr@5': 0.81, 'hit_rate@5': 0.89, 'p99_latency_ms': 287},
}
print_comparison_table(results)

Действия по результатам сравнительного тестирования

Быстрая проверка

Итоги урока