bosch-hvac-products-bot/evaluate.py

import json
import os
import time
from typing import Optional
from pathlib import Path
from dotenv import load_dotenv

from src.rag import BoschProductRAG

load_dotenv()


class Evaluator:
    """Evaluate RAG performance: latency and accuracy."""
    
    def __init__(self, catalog_path: str):
        self.catalog_path = catalog_path
        self.rag = BoschProductRAG()
        self.rag.build_index(self.rag.load_products(catalog_path))
        self.results = []
    
    def run_latency_tests(self, num_queries: int = 5) -> dict:
        """Measure end-to-end query latency."""
        test_queries = [
            "What's the most energy-efficient inverter system?",
            "Which heat pump can handle the widest operating range?",
            "What cooling systems use R32 refrigerant?",
            "Can you recommend an IAQ system for home filtration?",
            "What smart thermostat models are available?",
        ]
        
        latencies = []
        for i, query in enumerate(test_queries[:num_queries]):
            result = self.rag.query(query)
            latencies.append(result["latency_ms"])
            print(f"  Query {i+1}: {result['latency_ms']:.1f}ms")
        
        return {
            "mean_latency_ms": sum(latencies) / len(latencies),
            "min_latency_ms": min(latencies),
            "max_latency_ms": max(latencies),
            "num_queries": len(latencies),
            "latencies": latencies,
        }
    
    def run_accuracy_tests(self) -> dict:
        """Manual spot-check: test queries with expected answer patterns."""
        test_cases = [
            {
                "query": "What's the most energy-efficient heat pump system?",
                "expected_keywords": ["IDS", "SEER2", "efficiency"],
                "description": "Efficiency question should mention high SEER2 models"
            },
            {
                "query": "Which system is best for a retrofit installation?",
                "expected_keywords": ["IDS Pro", "ductless", "retrofit"],
                "description": "Retrofit question should mention ductless options"
            },
            {
                "query": "What's the noise level of your heat pumps?",
                "expected_keywords": ["dB", "noise", "quiet"],
                "description": "Noise question should reference dB ratings"
            },
            {
                "query": "Do you offer smart controls for thermostats?",
                "expected_keywords": ["thermostat", "smart", "control"],
                "description": "Smart controls question should mention thermostat models"
            },
            {
                "query": "What air quality systems do you have?",
                "expected_keywords": ["IAQ", "filtration", "air"],
                "description": "Air quality question should mention IAQ system"
            },
        ]
        
        correct = 0
        for i, test_case in enumerate(test_cases):
            result = self.rag.query(test_case["query"])
            answer = result["answer"].lower()
            
            # Check if any expected keyword appears in the answer
            found = any(kw.lower() in answer for kw in test_case["expected_keywords"])
            
            status = "✓" if found else "✗"
            print(f"  {status} Test {i+1}: {test_case['description']}")
            if not found:
                print(f"     Expected keywords: {test_case['expected_keywords']}")
            
            if found:
                correct += 1
        
        accuracy = (correct / len(test_cases)) * 100 if test_cases else 0
        
        return {
            "accuracy_percent": accuracy,
            "passed": correct,
            "total": len(test_cases),
            "test_cases": test_cases,
        }
    
    def generate_report(self, output_path: str = "metrics_report.json") -> None:
        """Run all tests and generate a report."""
        print("\n" + "=" * 60)
        print("RUNNING LATENCY TESTS")
        print("=" * 60)
        latency_results = self.run_latency_tests(num_queries=5)
        
        print("\n" + "=" * 60)
        print("RUNNING ACCURACY TESTS")
        print("=" * 60)
        accuracy_results = self.run_accuracy_tests()
        
        report = {
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "latency": latency_results,
            "accuracy": accuracy_results,
            "summary": {
                "mean_latency_ms": latency_results["mean_latency_ms"],
                "accuracy_percent": accuracy_results["accuracy_percent"],
                "status": "PASS" if accuracy_results["accuracy_percent"] >= 80 else "NEEDS REVIEW"
            }
        }
        
        with open(output_path, 'w') as f:
            json.dump(report, f, indent=2)
        
        print("\n" + "=" * 60)
        print("EVALUATION REPORT")
        print("=" * 60)
        print(f"Mean Latency: {report['summary']['mean_latency_ms']:.1f}ms")
        print(f"Accuracy: {report['summary']['accuracy_percent']:.1f}%")
        print(f"Status: {report['summary']['status']}")
        print(f"Report saved to: {output_path}")
        print("=" * 60 + "\n")


if __name__ == "__main__":
    import sys
    
    catalog_path = os.path.join(
        os.path.dirname(__file__), "src", "products_catalog.json"
    )
    
    evaluator = Evaluator(catalog_path)
    evaluator.generate_report()