""" 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())