Blame for scripts/benchmark_100_cache_queries.py

"""
Sahih Islam - 100-Query RAG Cache Hit & Multi-Attribute Capability Benchmark
Evaluates:
1. Cache Hit Rate (%) across 100 diverse real-world questions across all 20 Islamic topics
2. Dynamic Slicing Accuracy (top_k from 1 to 6)
3. Multilingual Localization (Urdu, Arabic, Hinglish, English)
4. Sanad Dynamic Formatting Toggle (include_sanad = True vs False)
5. Ultra-low Latency Metrics (p50, p95, min, max in ms)
"""

import os
import sys
import time
import asyncio
import statistics
from typing import List, Dict, Any

project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if project_root not in sys.path:
    sys.path.insert(0, project_root)

for _stream in (sys.stdout, sys.stderr):
    try:
        _stream.reconfigure(encoding="utf-8", errors="replace")
    except (AttributeError, OSError):
        pass

from app.infrastructure.redis import redis_manager
from app.rag.cache import get_cached_canonical_documents
from app.rag.top_1000_questions import TOP_1000_QUESTIONS

TOPIC_NAMES = [
    "1. Intention, Sincerity & Tawheed",
    "2. Purification, Wudu & Miswak",
    "3. Ghusl, Menstruation & Tayammum",
    "4. Five Daily Prayers, Times & Rulings",
    "5. Congregation, Jumu'ah & Masjids",
    "6. Sunnah, Nawafil, Tahajjud & Witr",
    "7. Missed Prayers, Travel & Forgetfulness",
    "8. Fasting Ramadan, Suhur & Iftar",
    "9. Voluntary Fasting & Laylatul Qadr",
    "10. Zakat, Wealth & Sadaqah",
    "11. Hajj, Umrah & Dhul Hijjah",
    "12. Nikah, Proposals & Spousal Rights",
    "13. Divorce, Iddah & Child Custody",
    "14. Parents, Children & Relatives",
    "15. Halal & Haram Food, Drink & Habits",
    "16. Finance, Business, Riba & Debts",
    "17. Character, Speech & Sins of Tongue",
    "18. Duas, Adhkar & Repentance",
    "19. Waswas, Ruqyah & Evil Eye",
    "20. Death, Funeral & The Hereafter"
]

LANGUAGES = ["urdu", "english", "arabic", "hinglish"]


def has_arabic(text: str) -> bool:
    """Check if text contains Arabic Unicode characters."""
    return any("\u0600" <= c <= "\u06FF" or "\u0750" <= c <= "\u077F" for c in text)


def has_urdu(text: str) -> bool:
    """Check if text contains Nastaliq/Urdu-specific or Arabic characters."""
    return any("\u0600" <= c <= "\u06FF" for c in text)


async def main():
    print("=" * 95)
    print("SAHIH ISLAM - 100-QUERY COMPREHENSIVE RAG CACHE BENCHMARK")
    print("Evaluating: Top-K Slicing (1-6) | Sanad Toggle (True/False) | Multilingual (UR, AR, HING, EN)")
    print("=" * 95)

    await redis_manager.init()

    # Generate 100 distinct queries (5 from each of the 20 categories)
    test_queries = []
    for cat_idx in range(20):
        base_idx = cat_idx * 50
        offsets = [0, 10, 20, 30, 40]
        for offset in offsets:
            q_idx = base_idx + offset
            test_queries.append({
                "topic": TOPIC_NAMES[cat_idx],
                "index": q_idx,
                "query": TOP_1000_QUESTIONS[q_idx]
            })

    assert len(test_queries) == 100, f"Expected 100 queries, got {len(test_queries)}"

    results = []
    latencies = []
    hits = 0
    slicing_correct = 0
    sanad_correct = 0
    language_correct = 0

    print(f"\n{'ID':<4} | {'TOPIC':<30} | {'TOP-K':<5} | {'SANAD':<5} | {'LANG':<8} | {'LATENCY':<8} | {'STATUS'}")
    print("-" * 95)

    for i, item in enumerate(test_queries):
        top_k = (i % 6) + 1  # 1 to 6
        include_sanad = (i % 2 == 0)  # True, False
        lang = LANGUAGES[i % 4]  # urdu, english, arabic, hinglish

        t0 = time.perf_counter()
        docs = await get_cached_canonical_documents(
            query=item["query"],
            top_k=top_k,
            language=lang,
            include_sanad=include_sanad
        )
        latency_ms = (time.perf_counter() - t0) * 1000
        latencies.append(latency_ms)

        if docs and len(docs) > 0:
            hits += 1
            # 1. Slicing check
            if len(docs) == min(top_k, 6):
                slicing_correct += 1

            # 2. Sanad check on primary document
            sample_text = docs[0].get("display_text", "")
            sanad_field = docs[0].get("sanad", "")
            if include_sanad:
                # If sanad is requested and available, it should format with [SANAD]:
                if not sanad_field or "[SANAD]:" in sample_text:
                    sanad_correct += 1
            else:
                # If sanad is NOT requested, [SANAD]: must NOT appear in display_text
                if "[SANAD]:" not in sample_text:
                    sanad_correct += 1

            # 3. Multilingual check
            if lang == "urdu":
                if has_urdu(sample_text) or docs[0].get("text_urdu"):
                    language_correct += 1
            elif lang == "arabic":
                if has_arabic(sample_text) or docs[0].get("text_arabic"):
                    language_correct += 1
            elif lang in ("hinglish", "english"):
                # English or transliterated latin script
                if len(sample_text) > 0:
                    language_correct += 1

            status = f"✅ HIT ({len(docs)} docs, Ref: {docs[0].get('reference', 'N/A')})"
        else:
            status = "❌ MISS"

        topic_short = item["topic"].split(". ")[1][:28]
        sanad_str = "True" if include_sanad else "False"
        print(f"#{i+1:03d} | {topic_short:<30} | k={top_k:<3} | {sanad_str:<5} | {lang:<8} | {latency_ms:6.2f}ms | {status}")

        results.append({
            "id": i + 1,
            "query": item["query"],
            "topic": item["topic"],
            "top_k": top_k,
            "include_sanad": include_sanad,
            "language": lang,
            "hit": bool(docs),
            "doc_count": len(docs) if docs else 0,
            "latency_ms": latency_ms
        })

    # Summary Statistics
    hit_rate = (hits / len(test_queries)) * 100.0
    slicing_acc = (slicing_correct / hits) * 100.0 if hits else 0
    sanad_acc = (sanad_correct / hits) * 100.0 if hits else 0
    lang_acc = (language_correct / hits) * 100.0 if hits else 0

    p50 = statistics.median(latencies)
    mean_lat = statistics.mean(latencies)
    sorted_lats = sorted(latencies)
    p95 = sorted_lats[int(len(sorted_lats) * 0.95)]
    min_lat = min(latencies)
    max_lat = max(latencies)

    print("\n" + "=" * 95)
    print("BENCHMARK RESULTS & METRICS SUMMARY")
    print("=" * 95)
    print(f"Total Questions Evaluated       : {len(test_queries)}")
    print(f"Total Cache Hits                : {hits} / {len(test_queries)} ({hit_rate:.1f}%)")
    print(f"Dynamic Slicing Accuracy        : {slicing_correct} / {hits} ({slicing_acc:.1f}%) [k=1..6]")
    print(f"Sanad Dynamic Toggle Accuracy   : {sanad_correct} / {hits} ({sanad_acc:.1f}%)")
    print(f"Multilingual Fidelity Accuracy  : {language_correct} / {hits} ({lang_acc:.1f}%) [UR, AR, HING, EN]")
    print("-" * 95)
    print("LATENCY DISTRIBUTION (Direct Upstash Redis over TLS):")
    print(f"   • Mean Latency               : {mean_lat:.2f} ms")
    print(f"   • Median (P50) Latency       : {p50:.2f} ms")
    print(f"   • 95th Percentile (P95)      : {p95:.2f} ms")
    print(f"   • Fastest Response (Min)     : {min_lat:.2f} ms")
    print(f"   • Peak Response (Max)        : {max_lat:.2f} ms")
    print("=" * 95)

    await redis_manager.close()


if __name__ == "__main__":
    asyncio.run(main())