cli/evaluation_cli.py

#!/usr/bin/env python3

import argparse
import json
import sys
import os

sys.path.insert(0, '.')

from lib.hybrid_search import HybridSearch
from lib.search_utils import load_movies, PROJECT_ROOT


def calculate_precision_at_k(retrieved_titles: list[str], relevant_titles: list[str], k: int) -> float:
    """Calculate precision@k: percentage of retrieved results that are relevant"""
    # Only consider top k retrieved titles
    retrieved_top_k = retrieved_titles[:k]
    
    # Count how many retrieved titles are in the relevant set
    relevant_set = set(relevant_titles)
    matches = sum(1 for title in retrieved_top_k if title in relevant_set)
    
    # Precision = matches / k
    precision = matches / k if k > 0 else 0.0
    return precision


def calculate_recall_at_k(retrieved_titles: list[str], relevant_titles: list[str], k: int) -> float:
    """Calculate recall@k: percentage of relevant docs that were retrieved"""
    # Only consider top k retrieved titles
    retrieved_top_k = retrieved_titles[:k]
    
    # Count how many relevant titles are in the retrieved set
    retrieved_set = set(retrieved_top_k)
    matches = sum(1 for title in relevant_titles if title in retrieved_set)
    
    # Recall = matches / total_relevant
    total_relevant = len(relevant_titles)
    recall = matches / total_relevant if total_relevant > 0 else 0.0
    return recall


def calculate_f1_score(precision: float, recall: float) -> float:
    """Calculate F1 score: harmonic mean of precision and recall"""
    # F1 = 2 * (precision * recall) / (precision + recall)
    if precision + recall == 0:
        return 0.0
    
    f1 = 2 * (precision * recall) / (precision + recall)
    return f1


def main():
    parser = argparse.ArgumentParser(description="Search Evaluation CLI")
    parser.add_argument(
        "--limit",
        type=int,
        default=5,
        help="Number of results to evaluate (k for precision@k, recall@k)",
    )

    args = parser.parse_args()
    limit = args.limit

    # Load golden dataset
    golden_dataset_path = os.path.join(PROJECT_ROOT, "data", "golden_dataset.json")
    
    with open(golden_dataset_path, "r") as f:
        golden_dataset = json.load(f)
    
    # Extract test cases from the JSON
    test_cases = golden_dataset["test_cases"]
    
    # Load documents and initialize hybrid search
    documents = load_movies()
    hybrid_search = HybridSearch(documents)
    
    # Print header
    print(f"k={limit}\n")
    
    # Evaluate each test case
    for test_case in test_cases:
        query = test_case["query"]
        relevant_titles = test_case["relevant_docs"]
        
        # Perform RRF search
        results = hybrid_search.rrf_search(query, k=60, limit=limit)
        
        # Extract retrieved titles
        retrieved_titles = [result["title"] for result in results]
        
        # Calculate precision@k, recall@k, and F1 score
        precision = calculate_precision_at_k(retrieved_titles, relevant_titles, limit)
        recall = calculate_recall_at_k(retrieved_titles, relevant_titles, limit)
        f1 = calculate_f1_score(precision, recall)
        
        # Print results
        print(f"- Query: {query}")
        print(f"  - Precision@{limit}: {precision:.4f}")
        print(f"  - Recall@{limit}: {recall:.4f}")
        print(f"  - F1 Score: {f1:.4f}")
        print(f"  - Retrieved: {', '.join(retrieved_titles)}")
        print(f"  - Relevant: {', '.join(relevant_titles)}")
        print()


if __name__ == "__main__":
    main()