File size: 2,940 Bytes
36b2bff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | #!/usr/bin/env python3
"""Calibrate the relevance threshold for the reranker.
Tests retrieval at various threshold values and reports precision/recall
to find the optimal F1 threshold. Used for Gate 2.
Usage:
python scripts/evaluation/run_threshold_calibration.py
"""
import json
import logging
import sys
from pathlib import Path
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger(__name__)
def main():
eval_path = Path(__file__).parent.parent.parent / "tests" / "evaluation" / "eval_queries.json"
queries = json.loads(eval_path.read_text())
relevant_queries = [q for q in queries if q.get("expected_category") is not None]
logger.info(f"Loaded {len(relevant_queries)} relevant queries for calibration")
import asyncio
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
from app.core.config import get_settings
from app.services.rag import RAGService
settings = get_settings()
rag = RAGService(settings)
asyncio.run(rag.initialize())
# Test thresholds from 0.20 to 0.60
thresholds = [0.20, 0.25, 0.30, 0.35, 0.40, 0.45, 0.50, 0.55, 0.60]
logger.info("\nThreshold | Precision | Recall | F1")
logger.info("-" * 45)
best_f1 = 0
best_threshold = 0.40
for threshold in thresholds:
true_positives = 0
false_positives = 0
false_negatives = 0
for q in relevant_queries:
results = rag.retrieve(q["query"], n_results=10)
if not results:
false_negatives += 1
continue
# Filter by threshold
from app.services.rag_safety import filter_by_relevance
filtered = filter_by_relevance(results, threshold=threshold)
if not filtered:
false_negatives += 1
else:
top_cat = filtered[0].get("metadata", {}).get("category", "")
if top_cat == q["expected_category"]:
true_positives += 1
else:
false_positives += 1
precision = true_positives / max(true_positives + false_positives, 1)
recall = true_positives / max(true_positives + false_negatives, 1)
f1 = 2 * precision * recall / max(precision + recall, 0.001)
logger.info(f" {threshold:.2f} | {precision:.3f} | {recall:.3f} | {f1:.3f}")
if f1 > best_f1:
best_f1 = f1
best_threshold = threshold
logger.info(f"\nOptimal threshold: {best_threshold} (F1={best_f1:.3f})")
logger.info(f"Update config: rag_similarity_threshold = {best_threshold}")
output = {"thresholds_tested": thresholds, "best_threshold": best_threshold, "best_f1": best_f1}
output_path = Path(__file__).parent / "threshold_calibration_results.json"
output_path.write_text(json.dumps(output, indent=2))
if __name__ == "__main__":
main()
|