eval_pipeline.py

"""
eval_pipeline.py
Evaluation script for the RAG pipeline.
Tests retrieval relevance and answer quality on a small QA set.
Run: python eval_pipeline.py
"""

import json
import time
import logging
from dataclasses import dataclass, field
from typing import List

from rag_pipeline import RAGPipeline

logging.basicConfig(level=logging.WARNING)  # Suppress info during eval


# Test Dataset


SAMPLE_DOC = "data/sample.txt"

# Manually crafted QA pairs based on sample.txt
TEST_CASES = [
    {
        "question": "What is the main topic of this document?",
        "expected_keywords": ["machine learning", "artificial intelligence", "data"],
    },
    {
        "question": "What methods or techniques are described?",
        "expected_keywords": ["model", "training", "algorithm", "classification"],
    },
    {
        "question": "What are the key findings or conclusions?",
        "expected_keywords": ["result", "performance", "accuracy", "conclusion"],
    },
]



# Metrics


@dataclass
class EvalResult:
    question: str
    answer: str
    latency_s: float
    top_similarity: float
    keyword_hit: bool
    sources_returned: int


def keyword_match(answer: str, keywords: List[str]) -> bool:
    """Check if any expected keyword appears in the answer (case-insensitive)."""
    answer_lower = answer.lower()
    return any(kw.lower() in answer_lower for kw in keywords)



# Run Evaluation


def run_eval():
    print("\n" + "="*60)
    print("  RAG PIPELINE EVALUATION")
    print("="*60)

    rag = RAGPipeline()

    print(f"\n📄 Ingesting: {SAMPLE_DOC}")
    stats = rag.ingest_document(SAMPLE_DOC)
    print(f"   Chunks: {stats['chunks']} | Dim: {stats['embedding_dim']} | Time: {stats['ingestion_time_s']}s")

    results: List[EvalResult] = []

    print(f"\n🔍 Running {len(TEST_CASES)} test queries...\n")

    for i, tc in enumerate(TEST_CASES):
        question = tc["question"]
        expected = tc["expected_keywords"]

        output = rag.query(question)

        top_score = output["sources"][0]["score"] if output["sources"] else 0.0
        hit = keyword_match(output["answer"], expected)

        result = EvalResult(
            question=question,
            answer=output["answer"],
            latency_s=output["latency_s"],
            top_similarity=top_score,
            keyword_hit=hit,
            sources_returned=len(output["sources"]),
        )
        results.append(result)

        status = "✅" if hit else "⚠️"
        print(f"Q{i+1}: {question}")
        print(f"   Answer   : {output['answer'][:120]}...")
        print(f"   Latency  : {result.latency_s}s")
        print(f"   Top Sim  : {round(top_score * 100, 1)}%")
        print(f"   Keyword  : {status} {'HIT' if hit else 'MISS'}")
        print()

    # ── Summary ──
    avg_latency = round(sum(r.latency_s for r in results) / len(results), 2)
    avg_similarity = round(sum(r.top_similarity for r in results) / len(results) * 100, 1)
    keyword_hits = sum(1 for r in results if r.keyword_hit)

    print("="*60)
    print("  SUMMARY")
    print("="*60)
    print(f"  Total queries      : {len(results)}")
    print(f"  Keyword hit rate   : {keyword_hits}/{len(results)} ({round(keyword_hits/len(results)*100)}%)")
    print(f"  Avg latency        : {avg_latency}s")
    print(f"  Avg top similarity : {avg_similarity}%")
    print("="*60 + "\n")

    # Save results
    report = {
        "summary": {
            "total_queries": len(results),
            "keyword_hit_rate": f"{keyword_hits}/{len(results)}",
            "avg_latency_s": avg_latency,
            "avg_top_similarity_pct": avg_similarity,
        },
        "results": [
            {
                "question": r.question,
                "answer": r.answer,
                "latency_s": r.latency_s,
                "top_similarity_pct": round(r.top_similarity * 100, 1),
                "keyword_hit": r.keyword_hit,
            }
            for r in results
        ]
    }

    with open("eval_report.json", "w") as f:
        json.dump(report, f, indent=2)
    print("📊 Report saved to eval_report.json")


if __name__ == "__main__":
    run_eval()