مقاييس الاسترجاع: معدل الإصابة وMRR وNDCG

def hit_rate_at_k(golden_dataset, retriever, k=5):
    hits = 0
    for item in golden_dataset:
        retrieved = retriever.retrieve(item['question'], top_k=k)
        retrieved_ids = [r['id'] for r in retrieved[:k]]
        relevant_ids = set(item['relevant_chunk_ids'])
        if any(rid in relevant_ids for rid in retrieved_ids):
            hits += 1
    return hits / len(golden_dataset)

for k in [1, 3, 5, 10]:
    hr = hit_rate_at_k(golden_dataset, retriever, k=k)
    print(f'Hit rate@{k}: {hr:.1%}')

def mean_reciprocal_rank(golden_dataset, retriever, top_k=10):
    reciprocal_ranks = []
    for item in golden_dataset:
        retrieved = retriever.retrieve(item['question'], top_k=top_k)
        retrieved_ids = [r['id'] for r in retrieved]
        relevant_ids = set(item['relevant_chunk_ids'])

        rr = 0.0
        for rank, rid in enumerate(retrieved_ids, start=1):
            if rid in relevant_ids:
                rr = 1.0 / rank
                break  # only the first relevant result counts
        reciprocal_ranks.append(rr)

    mrr = sum(reciprocal_ranks) / len(reciprocal_ranks)
    print(f'MRR@{top_k}: {mrr:.3f}')
    return mrr

# MRR interpretation table
mrr_interpretations = {
    1.0:  'Perfect — relevant chunk always at rank 1',
    0.5:  'Good — typically at rank 2',
    0.33: 'Acceptable — typically at rank 3',
    0.25: 'Weak — typically at rank 4',
    0.1:  'Poor — relevant chunk rarely near the top'
}

for score, description in mrr_interpretations.items():
    print(f'MRR {score:.2f}: {description}')

def precision_at_k(golden_dataset, retriever, k=5):
    precisions = []
    for item in golden_dataset:
        retrieved = retriever.retrieve(item['question'], top_k=k)
        retrieved_ids = [r['id'] for r in retrieved[:k]]
        relevant_ids = set(item['relevant_chunk_ids'])

        relevant_retrieved = sum(
            1 for rid in retrieved_ids if rid in relevant_ids
        )
        precision = relevant_retrieved / k
        precisions.append(precision)

    mean_precision = sum(precisions) / len(precisions)
    print(f'Precision@{k}: {mean_precision:.3f}')
    return mean_precision

import math

def dcg_at_k(relevances, k):
    '''relevances[i] = 1 if chunk at rank i+1 is relevant, else 0'''
    dcg = 0.0
    for i, rel in enumerate(relevances[:k]):
        # rank is i+1, discount is log2(rank + 1)
        dcg += rel / math.log2(i + 2)
    return dcg

def ndcg_at_k(retrieved_ids, relevant_ids, k):
    relevances = [1 if rid in relevant_ids else 0
                  for rid in retrieved_ids[:k]]
    actual_dcg = dcg_at_k(relevances, k)
    ideal_dcg = dcg_at_k([1] * min(len(relevant_ids), k), k)
    return actual_dcg / ideal_dcg if ideal_dcg > 0 else 0.0

def mean_ndcg_at_k(golden_dataset, retriever, k=5):
    ndcg_scores = []
    for item in golden_dataset:
        retrieved = retriever.retrieve(item['question'], top_k=k)
        retrieved_ids = [r['id'] for r in retrieved[:k]]
        relevant_ids = set(item['relevant_chunk_ids'])
        score = ndcg_at_k(retrieved_ids, relevant_ids, k)
        ndcg_scores.append(score)
    mean = sum(ndcg_scores) / len(ndcg_scores)
    print(f'NDCG@{k}: {mean:.4f}')
    return mean

# Compute all retrieval metrics together
hit_rate = hit_rate_at_k(golden_dataset, retriever, k=5)
mrr = mean_reciprocal_rank(golden_dataset, retriever, top_k=5)
ndcg = mean_ndcg_at_k(golden_dataset, retriever, k=5)

from ragas import evaluate
from ragas.metrics import (
    context_precision,
    context_recall,
    faithfulness,
    answer_relevancy
)
from datasets import Dataset

eval_data = Dataset.from_list([
    {
        'question': item['question'],
        'answer': item['generated_answer'],
        'contexts': item['retrieved_texts'],
        'ground_truth': item['expected_answer']
    }
    for item in golden_dataset_with_answers
])

results = evaluate(
    eval_data,
    metrics=[context_precision, context_recall, faithfulness, answer_relevancy]
)
print(results)

from collections import defaultdict

def evaluate_by_category(golden_dataset, retriever):
    by_category = defaultdict(list)
    for item in golden_dataset:
        category = item.get('category', 'unknown')
        retrieved = retriever.retrieve(item['question'], top_k=5)
        retrieved_ids = {r['id'] for r in retrieved}
        hit = bool(retrieved_ids & set(item['relevant_chunk_ids']))
        by_category[category].append(hit)

    print('Hit rate by category:')
    for cat, hits in sorted(by_category.items()):
        hr = sum(hits) / len(hits)
        print(f'  {cat}: {hr:.1%} ({sum(hits)}/{len(hits)})')

def analyze_failures(golden_dataset, retriever, top_k=5):
    failures = []
    for item in golden_dataset:
        retrieved = retriever.retrieve(item['question'], top_k=top_k)
        retrieved_ids = {r['id'] for r in retrieved}
        relevant_ids = set(item['relevant_chunk_ids'])
        if not (retrieved_ids & relevant_ids):
            failures.append({
                'question': item['question'],
                'expected_chunks': list(relevant_ids),
                'retrieved_chunks': [r['id'] for r in retrieved],
                'top_score': retrieved[0]['score'] if retrieved else None
            })
    print(f'Failures: {len(failures)}/{len(golden_dataset)}')
    return failures

def recall_at_k(golden_dataset, retriever, k=5):
    recalls = []
    for item in golden_dataset:
        retrieved = retriever.retrieve(item['question'], top_k=k)
        retrieved_ids = set(r['id'] for r in retrieved[:k])
        relevant_ids = set(item['relevant_chunk_ids'])
        if not relevant_ids:
            continue  # skip items with no annotated relevant chunks
        retrieved_relevant = retrieved_ids & relevant_ids
        recall = len(retrieved_relevant) / len(relevant_ids)
        recalls.append(recall)
    mean_recall = sum(recalls) / len(recalls)
    print(f'Recall@{k}: {mean_recall:.3f}')
    return mean_recall