src/guardrails/response_validator.py

"""
Response Validator - Core response quality and safety validation

This module provides comprehensive validation of RAG responses including
quality metrics, safety checks, and content validation.
"""

import logging
import re
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Pattern

logger = logging.getLogger(__name__)


@dataclass
class ValidationResult:
    """Result of response validation with detailed metrics."""

    is_valid: bool
    confidence_score: float
    safety_passed: bool
    quality_score: float
    issues: List[str]
    suggestions: List[str]

    # Detailed quality metrics
    relevance_score: float = 0.0
    completeness_score: float = 0.0
    coherence_score: float = 0.0
    source_fidelity_score: float = 0.0

    # Safety metrics
    contains_pii: bool = False
    inappropriate_content: bool = False
    potential_bias: bool = False
    prompt_injection_detected: bool = False


class ResponseValidator:
    """
    Validates response quality and safety for RAG system.

    Provides comprehensive validation including:
    - Content safety and appropriateness
    - Response quality metrics
    - Source alignment validation
    - Professional tone assessment
    """

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        """
        Initialize ResponseValidator with configuration.

        Args:
            config: Configuration dictionary with validation thresholds
        """
        self.config = config or self._get_default_config()

        # Compile regex patterns for efficiency
        self._pii_patterns = self._compile_pii_patterns()
        self._inappropriate_patterns = self._compile_inappropriate_patterns()
        self._bias_patterns = self._compile_bias_patterns()

        logger.info("ResponseValidator initialized")

    def _get_default_config(self) -> Dict[str, Any]:
        """Get default validation configuration."""
        return {
            "min_relevance_score": 0.7,
            "min_completeness_score": 0.6,
            "min_coherence_score": 0.7,
            "min_source_fidelity_score": 0.8,
            "min_overall_quality": 0.7,
            "max_response_length": 1000,
            "min_response_length": 20,
            "require_citations": True,
            "strict_safety_mode": True,
        }

    def validate_response(self, response: str, sources: List[Dict[str, Any]], query: str) -> ValidationResult:
        """
        Validate response quality and safety.

        Args:
            response: Generated response text
            sources: Source documents used for generation
            query: Original user query

        Returns:
            ValidationResult with detailed validation metrics
        """
        try:
            # Perform safety checks
            safety_result = self.check_safety(response)

            # Calculate quality metrics
            quality_scores = self._calculate_quality_scores(response, sources, query)

            # Check response format and citations
            format_issues = self._validate_format(response, sources)

            # Calculate overall confidence
            confidence = self.calculate_confidence(response, sources, quality_scores)

            # Determine if response passes validation
            is_valid = (
                safety_result["passed"]
                and quality_scores["overall"] >= self.config["min_overall_quality"]
                and len(format_issues) == 0
            )

            # Compile suggestions
            suggestions = []
            if not is_valid:
                suggestions.extend(self._generate_improvement_suggestions(safety_result, quality_scores, format_issues))

            return ValidationResult(
                is_valid=is_valid,
                confidence_score=confidence,
                safety_passed=safety_result["passed"],
                quality_score=quality_scores["overall"],
                issues=safety_result["issues"] + format_issues,
                suggestions=suggestions,
                relevance_score=quality_scores["relevance"],
                completeness_score=quality_scores["completeness"],
                coherence_score=quality_scores["coherence"],
                source_fidelity_score=quality_scores["source_fidelity"],
                contains_pii=safety_result["contains_pii"],
                inappropriate_content=safety_result["inappropriate_content"],
                potential_bias=safety_result["potential_bias"],
                prompt_injection_detected=safety_result["prompt_injection"],
            )

        except Exception as e:
            logger.error(f"Validation error: {e}")
            return ValidationResult(
                is_valid=False,
                confidence_score=0.0,
                safety_passed=False,
                quality_score=0.0,
                issues=[f"Validation error: {str(e)}"],
                suggestions=["Please retry the request"],
            )

    def calculate_confidence(
        self,
        response: str,
        sources: List[Dict[str, Any]],
        quality_scores: Optional[Dict[str, float]] = None,
    ) -> float:
        """
        Calculate overall confidence score for response.

        Args:
            response: Generated response text
            sources: Source documents used
            quality_scores: Pre-calculated quality scores

        Returns:
            Confidence score between 0.0 and 1.0
        """
        if quality_scores is None:
            quality_scores = self._calculate_quality_scores(response, sources, "")

        # Weight different factors
        weights = {
            "source_count": 0.2,
            "avg_source_relevance": 0.3,
            "response_quality": 0.4,
            "citation_presence": 0.1,
        }

        # Source-based confidence
        source_count_score = min(len(sources) / 3.0, 1.0)  # Max at 3 sources

        avg_relevance = sum(source.get("relevance_score", 0.0) for source in sources) / len(sources) if sources else 0.0

        # Citation presence
        has_citations = self._has_proper_citations(response, sources)
        citation_score = 1.0 if has_citations else 0.3

        # Combine scores
        confidence = (
            weights["source_count"] * source_count_score
            + weights["avg_source_relevance"] * avg_relevance
            + weights["response_quality"] * quality_scores["overall"]
            + weights["citation_presence"] * citation_score
        )

        return min(max(confidence, 0.0), 1.0)

    def check_safety(self, content: str) -> Dict[str, Any]:
        """
        Perform comprehensive safety checks on content.

        Args:
            content: Text content to check

        Returns:
            Dictionary with safety check results
        """
        issues = []

        # Check for PII
        contains_pii = self._detect_pii(content)
        if contains_pii:
            issues.append("Content may contain personally identifiable information")

        # Check for inappropriate content
        inappropriate_content = self._detect_inappropriate_content(content)
        if inappropriate_content:
            issues.append("Content contains inappropriate material")

        # Check for potential bias
        potential_bias = self._detect_bias(content)
        if potential_bias:
            issues.append("Content may contain biased language")

        # Check for prompt injection
        prompt_injection = self._detect_prompt_injection(content)
        if prompt_injection:
            issues.append("Potential prompt injection detected")

        # Overall safety assessment
        passed = (
            not contains_pii
            and not inappropriate_content
            and (not potential_bias or not self.config["strict_safety_mode"])
        )

        return {
            "passed": passed,
            "issues": issues,
            "contains_pii": contains_pii,
            "inappropriate_content": inappropriate_content,
            "potential_bias": potential_bias,
            "prompt_injection": prompt_injection,
        }

    def _calculate_quality_scores(self, response: str, sources: List[Dict[str, Any]], query: str) -> Dict[str, float]:
        """Calculate detailed quality metrics."""

        # Relevance: How well does response address the query
        relevance = self._calculate_relevance(response, query)

        # Completeness: Does response adequately address the question
        completeness = self._calculate_completeness(response, query)

        # Coherence: Is the response logically structured and coherent
        coherence = self._calculate_coherence(response)

        # Source fidelity: How well does response align with sources
        source_fidelity = self._calculate_source_fidelity(response, sources)

        # Overall quality (weighted average)
        overall = 0.3 * relevance + 0.25 * completeness + 0.2 * coherence + 0.25 * source_fidelity

        return {
            "relevance": relevance,
            "completeness": completeness,
            "coherence": coherence,
            "source_fidelity": source_fidelity,
            "overall": overall,
        }

    def _calculate_relevance(self, response: str, query: str) -> float:
        """Calculate relevance score between response and query."""
        if not query.strip():
            return 1.0  # No query to compare against

        # Simple keyword overlap for now (can be enhanced with embeddings)
        query_words = set(query.lower().split())
        response_words = set(response.lower().split())

        if not query_words:
            return 1.0

        overlap = len(query_words.intersection(response_words))
        return min(overlap / len(query_words), 1.0)

    def _calculate_completeness(self, response: str, query: str) -> float:
        """Calculate completeness score based on response length and structure."""
        target_length = 200  # Ideal response length

        # Length-based score
        length_score = min(len(response) / target_length, 1.0)

        # Structure score (presence of clear statements)
        has_conclusion = any(
            phrase in response.lower() for phrase in ["according to", "based on", "in summary", "therefore"]
        )
        structure_score = 1.0 if has_conclusion else 0.7

        return (length_score + structure_score) / 2.0

    def _calculate_coherence(self, response: str) -> float:
        """Calculate coherence score based on response structure."""
        sentences = response.split(".")
        if len(sentences) < 2:
            return 0.8  # Short responses are typically coherent

        # Check for repetition
        unique_sentences = len(set(s.strip().lower() for s in sentences if s.strip()))
        repetition_score = unique_sentences / len([s for s in sentences if s.strip()])

        # Check for logical flow indicators
        flow_indicators = [
            "however",
            "therefore",
            "additionally",
            "furthermore",
            "consequently",
        ]
        has_flow = any(indicator in response.lower() for indicator in flow_indicators)
        flow_score = 1.0 if has_flow else 0.8

        return (repetition_score + flow_score) / 2.0

    def _calculate_source_fidelity(self, response: str, sources: List[Dict[str, Any]]) -> float:
        """Calculate how well response aligns with source documents."""
        if not sources:
            return 0.5  # Neutral score if no sources

        # Check for citation presence
        has_citations = self._has_proper_citations(response, sources)
        citation_score = 1.0 if has_citations else 0.3

        # Check for content alignment (simplified)
        source_content = " ".join(source.get("excerpt", "") for source in sources).lower()

        response_lower = response.lower()

        # Look for key terms from sources in response
        source_words = set(source_content.split())
        response_words = set(response_lower.split())

        if source_words:
            alignment = len(source_words.intersection(response_words)) / len(source_words)
        else:
            alignment = 0.5

        return (citation_score + min(alignment * 2, 1.0)) / 2.0

    def _has_proper_citations(self, response: str, sources: List[Dict[str, Any]]) -> bool:
        """Check if response contains proper citations."""
        if not self.config["require_citations"]:
            return True

        # Look for citation patterns
        citation_patterns = [
            r"\[.*?\]",  # [source]
            r"\(.*?\)",  # (source)
            r"according to.*?",  # according to X
            r"based on.*?",  # based on X
        ]

        has_citation_format = any(re.search(pattern, response, re.IGNORECASE) for pattern in citation_patterns)

        # Check if source documents are mentioned
        source_names = [source.get("document", "").lower() for source in sources]

        response_lower = response.lower()
        mentions_sources = any(name in response_lower for name in source_names if name)

        return has_citation_format or mentions_sources

    def _validate_format(self, response: str, sources: List[Dict[str, Any]]) -> List[str]:
        """Validate response format and structure."""
        issues = []

        # Length validation
        if len(response) < self.config["min_response_length"]:
            min_length = self.config["min_response_length"]
            issues.append(f"Response too short (minimum {min_length} characters)")

        if len(response) > self.config["max_response_length"]:
            max_length = self.config["max_response_length"]
            issues.append(f"Response too long (maximum {max_length} characters)")

        # Professional tone check (basic)
        informal_patterns = [
            r"\byo\b",
            r"\bwassup\b",
            r"\bgonna\b",
            r"\bwanna\b",
            r"\bunrealz\b",
            r"\bwtf\b",
            r"\bomg\b",
        ]

        if any(re.search(pattern, response, re.IGNORECASE) for pattern in informal_patterns):
            issues.append("Response contains informal language")

        return issues

    def _generate_improvement_suggestions(
        self,
        safety_result: Dict[str, Any],
        quality_scores: Dict[str, float],
        format_issues: List[str],
    ) -> List[str]:
        """Generate suggestions for improving response quality."""
        suggestions = []

        if not safety_result["passed"]:
            suggestions.append("Review content for safety and appropriateness")

        if quality_scores["relevance"] < self.config["min_relevance_score"]:
            suggestions.append("Ensure response directly addresses the user's question")

        if quality_scores["completeness"] < self.config["min_completeness_score"]:
            suggestions.append("Provide more comprehensive information")

        if quality_scores["source_fidelity"] < self.config["min_source_fidelity_score"]:
            suggestions.append("Include proper citations and source references")

        if format_issues:
            suggestions.append("Review response format and professional tone")

        return suggestions

    def _compile_pii_patterns(self) -> List[Pattern[str]]:
        """Compile regex patterns for PII detection."""
        patterns = [
            r"\b\d{3}-\d{2}-\d{4}\b",  # SSN
            r"\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b",  # Credit card
            r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",  # Email
            r"\b\d{3}[-.]\d{3}[-.]\d{4}\b",  # Phone number
        ]
        return [re.compile(pattern) for pattern in patterns]

    def _compile_inappropriate_patterns(self) -> List[Pattern[str]]:
        """Compile regex patterns for inappropriate content detection."""
        # Basic patterns (expand as needed)
        patterns = [
            r"\b(?:hate|discriminat|harass)\w*\b",
            r"\b(?:offensive|inappropriate|unprofessional)\b",
        ]
        return [re.compile(pattern, re.IGNORECASE) for pattern in patterns]

    def _compile_bias_patterns(self) -> List[Pattern[str]]:
        """Compile regex patterns for bias detection."""
        patterns = [
            r"\b(?:always|never|all|none)\s+(?:men|women|people)\b",
            r"\b(?:typical|usual)\s+(?:man|woman|person)\b",
        ]
        return [re.compile(pattern, re.IGNORECASE) for pattern in patterns]

    def _detect_pii(self, content: str) -> bool:
        """Detect personally identifiable information."""
        return any(pattern.search(content) for pattern in self._pii_patterns)

    def _detect_inappropriate_content(self, content: str) -> bool:
        """Detect inappropriate content."""
        return any(pattern.search(content) for pattern in self._inappropriate_patterns)

    def _detect_bias(self, content: str) -> bool:
        """Detect potential bias in content."""
        return any(pattern.search(content) for pattern in self._bias_patterns)

    def _detect_prompt_injection(self, content: str) -> bool:
        """Detect potential prompt injection attempts."""
        injection_patterns = [
            r"ignore\s+(?:previous|all)\s+instructions",
            r"system\s*:",
            r"assistant\s*:",
            r"user\s*:",
            r"prompt\s*:",
        ]

        return any(re.search(pattern, content, re.IGNORECASE) for pattern in injection_patterns)