test_inadequate_responses.py

#!/usr/bin/env python3
"""
Test script for identifying inadequate responses in the GemmaSOS system
This script helps identify where the system generates responses that need improvement
"""

import sys
import os
import json
import logging
from typing import Dict, List, Any

# Add the current directory to the path
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from crisis_detector import CrisisDetector
from response_generator import CrisisResponseGenerator
from safety_system import SafetySystem

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class InadequateResponseTester:
    """
    Test suite for identifying inadequate responses in the crisis intervention system
    """
    
    def __init__(self):
        """Initialize the tester"""
        self.test_results = []
        self.crisis_detector = None
        self.response_generator = None
        self.safety_system = None
        
        # Load inadequate response examples
        self.inadequate_examples = self._load_inadequate_examples()
    
    def _load_inadequate_examples(self) -> List[Dict[str, Any]]:
        """Load examples of inadequate responses"""
        try:
            with open("inadequate_response_examples.json", "r") as f:
                data = json.load(f)
                return data["inadequate_response_examples"]
        except FileNotFoundError:
            logger.error("inadequate_response_examples.json not found")
            return []
        except Exception as e:
            logger.error(f"Error loading inadequate examples: {e}")
            return []
    
    def setup_components(self):
        """Initialize all system components"""
        try:
            logger.info("Setting up test components...")
            self.crisis_detector = CrisisDetector()
            self.response_generator = CrisisResponseGenerator()
            self.safety_system = SafetySystem()
            logger.info("Components initialized successfully")
            return True
        except Exception as e:
            logger.error(f"Error setting up components: {e}")
            return False
    
    def test_response_quality(self):
        """Test the quality of responses generated by the system"""
        logger.info("Testing response quality...")
        
        results = []
        
        for example in self.inadequate_examples:
            try:
                # Detect crisis
                crisis_result = self.crisis_detector.detect_crisis_from_text(example["user_input"])
                
                # Generate response
                if crisis_result["crisis_detected"]:
                    primary_category = crisis_result["categories"][0]["category"] if crisis_result["categories"] else "general"
                    response = self.response_generator.generate_response(
                        crisis_type=primary_category,
                        user_message=example["user_input"],
                        confidence=crisis_result["confidence"],
                        immediate_risk=crisis_result.get("immediate_risk", False)
                    )
                    generated_response = response["response"]
                else:
                    generated_response = "No crisis detected - general response would be generated"
                
                # Evaluate response quality
                quality_score = self._evaluate_response_quality(
                    generated_response, 
                    example["user_input"], 
                    example["crisis_type"],
                    example["severity"]
                )
                
                test_result = {
                    "example_id": example["id"],
                    "name": example["name"],
                    "user_input": example["user_input"],
                    "generated_response": generated_response,
                    "inadequate_example": example["inadequate_response"],
                    "why_inadequate": example["why_inadequate"],
                    "better_response": example["better_response"],
                    "quality_score": quality_score,
                    "crisis_detected": crisis_result["crisis_detected"],
                    "crisis_confidence": crisis_result["confidence"]
                }
                
                results.append(test_result)
                
                # Log results
                status = "GOOD" if quality_score >= 0.7 else "NEEDS_IMPROVEMENT"
                logger.info(f"Response Test {example['id']}: {status} - {example['name']} (Score: {quality_score:.2f})")
                
            except Exception as e:
                logger.error(f"Error testing example {example['id']}: {e}")
                results.append({
                    "example_id": example["id"],
                    "name": example["name"],
                    "error": str(e),
                    "quality_score": 0.0
                })
        
        return results
    
    def _evaluate_response_quality(self, response: str, user_input: str, crisis_type: str, severity: str) -> float:
        """
        Evaluate the quality of a generated response
        
        Args:
            response: The generated response
            user_input: The user's input
            crisis_type: Type of crisis
            severity: Severity level
            
        Returns:
            Quality score between 0.0 and 1.0
        """
        score = 0.0
        max_score = 10.0
        
        response_lower = response.lower()
        
        # Check for empathy indicators (2 points)
        empathy_indicators = ["i'm", "i care", "i understand", "i'm here", "i can hear", "i'm concerned"]
        if any(indicator in response_lower for indicator in empathy_indicators):
            score += 2.0
        
        # Check for validation (2 points)
        validation_indicators = ["valid", "understand", "hear", "acknowledge", "feel"]
        if any(indicator in response_lower for indicator in validation_indicators):
            score += 2.0
        
        # Check for appropriate resources (2 points)
        resource_indicators = ["988", "911", "crisis", "hotline", "emergency", "help"]
        if any(indicator in response_lower for indicator in resource_indicators):
            score += 2.0
        
        # Check for appropriate length (1 point)
        word_count = len(response.split())
        if 10 <= word_count <= 150:  # Appropriate length
            score += 1.0
        
        # Check for non-judgmental tone (1 point)
        judgmental_indicators = ["should", "must", "need to", "have to", "wrong", "bad"]
        if not any(indicator in response_lower for indicator in judgmental_indicators):
            score += 1.0
        
        # Check for immediate safety concerns if high severity (2 points)
        if severity in ["immediate", "high"]:
            safety_indicators = ["immediate", "emergency", "911", "988", "safety", "danger"]
            if any(indicator in response_lower for indicator in safety_indicators):
                score += 2.0
        else:
            score += 2.0  # Full points for non-immediate situations
        
        return min(score / max_score, 1.0)
    
    def identify_improvement_areas(self, test_results: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Identify areas where the system needs improvement"""
        improvement_areas = {
            "low_quality_responses": [],
            "common_issues": {},
            "recommendations": []
        }
        
        # Find low quality responses
        for result in test_results:
            if result.get("quality_score", 0.0) < 0.7:
                improvement_areas["low_quality_responses"].append({
                    "example_id": result["example_id"],
                    "name": result["name"],
                    "quality_score": result["quality_score"],
                    "issues": self._identify_specific_issues(result)
                })
        
        # Analyze common issues
        issue_counts = {}
        for result in improvement_areas["low_quality_responses"]:
            for issue in result["issues"]:
                issue_counts[issue] = issue_counts.get(issue, 0) + 1
        
        improvement_areas["common_issues"] = issue_counts
        
        # Generate recommendations
        if "lacks_empathy" in issue_counts:
            improvement_areas["recommendations"].append("Improve empathy in response templates")
        
        if "missing_resources" in issue_counts:
            improvement_areas["recommendations"].append("Ensure all crisis responses include relevant resources")
        
        if "inappropriate_length" in issue_counts:
            improvement_areas["recommendations"].append("Adjust response length guidelines")
        
        if "judgmental_tone" in issue_counts:
            improvement_areas["recommendations"].append("Review response templates for judgmental language")
        
        return improvement_areas
    
    def _identify_specific_issues(self, result: Dict[str, Any]) -> List[str]:
        """Identify specific issues with a response"""
        issues = []
        response = result.get("generated_response", "").lower()
        
        # Check for empathy
        empathy_indicators = ["i'm", "i care", "i understand", "i'm here", "i can hear"]
        if not any(indicator in response for indicator in empathy_indicators):
            issues.append("lacks_empathy")
        
        # Check for resources
        resource_indicators = ["988", "911", "crisis", "hotline", "emergency"]
        if not any(indicator in response for indicator in resource_indicators):
            issues.append("missing_resources")
        
        # Check for length
        word_count = len(result.get("generated_response", "").split())
        if word_count < 10 or word_count > 150:
            issues.append("inappropriate_length")
        
        # Check for judgmental tone
        judgmental_indicators = ["should", "must", "need to", "have to", "wrong", "bad"]
        if any(indicator in response for indicator in judgmental_indicators):
            issues.append("judgmental_tone")
        
        return issues
    
    def run_quality_tests(self):
        """Run all quality tests"""
        logger.info("Starting response quality testing...")
        
        if not self.setup_components():
            logger.error("Failed to setup components. Aborting tests.")
            return False
        
        # Test response quality
        test_results = self.test_response_quality()
        
        # Identify improvement areas
        improvement_areas = self.identify_improvement_areas(test_results)
        
        # Print summary
        logger.info("\n" + "="*60)
        logger.info("RESPONSE QUALITY TEST RESULTS")
        logger.info("="*60)
        
        total_tests = len(test_results)
        good_responses = sum(1 for result in test_results if result.get("quality_score", 0.0) >= 0.7)
        needs_improvement = total_tests - good_responses
        
        logger.info(f"Total tests: {total_tests}")
        logger.info(f"Good responses: {good_responses}")
        logger.info(f"Need improvement: {needs_improvement}")
        
        if improvement_areas["low_quality_responses"]:
            logger.info("\nLow quality responses identified:")
            for response in improvement_areas["low_quality_responses"]:
                logger.info(f"  - {response['name']} (Score: {response['quality_score']:.2f})")
        
        if improvement_areas["common_issues"]:
            logger.info("\nCommon issues:")
            for issue, count in improvement_areas["common_issues"].items():
                logger.info(f"  - {issue}: {count} occurrences")
        
        if improvement_areas["recommendations"]:
            logger.info("\nRecommendations:")
            for recommendation in improvement_areas["recommendations"]:
                logger.info(f"  - {recommendation}")
        
        # Save detailed results
        self.save_quality_results(test_results, improvement_areas)
        
        return needs_improvement == 0
    
    def save_quality_results(self, test_results: List[Dict[str, Any]], improvement_areas: Dict[str, Any]):
        """Save quality test results to file"""
        try:
            results = {
                "test_results": test_results,
                "improvement_areas": improvement_areas,
                "summary": {
                    "total_tests": len(test_results),
                    "good_responses": sum(1 for result in test_results if result.get("quality_score", 0.0) >= 0.7),
                    "needs_improvement": sum(1 for result in test_results if result.get("quality_score", 0.0) < 0.7)
                }
            }
            
            with open("response_quality_results.json", "w") as f:
                json.dump(results, f, indent=2)
            logger.info("Quality test results saved to response_quality_results.json")
        except Exception as e:
            logger.error(f"Error saving quality results: {e}")

def main():
    """Main test function"""
    tester = InadequateResponseTester()
    success = tester.run_quality_tests()
    
    if success:
        print("\n✅ All responses meet quality standards!")
        sys.exit(0)
    else:
        print("\n⚠️ Some responses need improvement. Check the results for details.")
        sys.exit(1)

if __name__ == "__main__":
    main()