Create confidence_scoring_v2.py

confidence_scoring_v2.py

@@ -0,0 +1,677 @@
+"""
+Confidence Scoring Module v2 for Khmer Legal Bridge
+====================================================
+
+This module provides scientifically-validated confidence scoring using:
+1. COMETKiwi (PRIMARY) - Reference-free neural QE, correlates with human judgments
+2. Back-Translation Verification (SECONDARY) - Catches semantic drift
+3. Legal Terminology Coverage (DOMAIN) - Domain-specific quality signal
+
+Based on findings from:
+- Fomicheva et al. (2020) - Token probabilities don't correlate with quality
+- Rei et al. (2022) - COMETKiwi for reference-free QE
+- WMT 2022/2023 - COMETKiwi winning system
+
+Author: Khmer Legal Bridge Project
+License: MIT
+"""
+
+import logging
+import re
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Tuple, Callable
+from difflib import SequenceMatcher
+
+logger = logging.getLogger(__name__)
+
+# ============================================================================
+# Data Classes
+# ============================================================================
+
+@dataclass
+class ConfidenceResult:
+    """Result of confidence scoring with full transparency."""
+    overall_score: float          # 0.0 to 1.0
+    quality_level: str            # "excellent", "good", "acceptable", "low", "very_low"
+    human_review_recommended: bool
+    components: Dict[str, float]  # Individual signal scores
+    explanations: List[str]       # Human-readable explanations
+    word_level_scores: Optional[List[Tuple[str, float]]] = None
+
+
+# ============================================================================
+# COMETKiwi Scorer (PRIMARY)
+# ============================================================================
+
+class COMETKiwiScorer:
+    """
+    Quality Estimation using COMETKiwi (reference-free).
+    
+    This is the PRIMARY confidence method - scientifically validated
+    to correlate with human judgments even for low-resource languages.
+    
+    Supports Khmer (khm_Khmr) natively.
+    """
+    
+    def __init__(self, model_name: str = "Unbabel/wmt22-cometkiwi-da"):
+        """
+        Initialize COMETKiwi model.
+        
+        Args:
+            model_name: HuggingFace model identifier
+                - "Unbabel/wmt22-cometkiwi-da" (560M params, recommended)
+                - "Unbabel/wmt23-cometkiwi-da-xl" (3.5B params, requires GPU)
+        """
+        self.model = None
+        self.model_name = model_name
+        self._loaded = False
+    
+    def _load_model(self):
+        """Lazy load the model to save memory on startup."""
+        if self._loaded:
+            return
+        
+        try:
+            from comet import download_model, load_from_checkpoint
+            
+            logger.info(f"Loading COMETKiwi model: {self.model_name}")
+            model_path = download_model(self.model_name)
+            self.model = load_from_checkpoint(model_path)
+            self._loaded = True
+            logger.info("COMETKiwi model loaded successfully")
+            
+        except ImportError:
+            logger.warning(
+                "unbabel-comet not installed. "
+                "Install with: pip install unbabel-comet"
+            )
+            raise
+        except Exception as e:
+            logger.error(f"Failed to load COMETKiwi: {e}")
+            raise
+    
+    def score(
+        self,
+        source_text: str,
+        translation: str,
+        batch_size: int = 1,
+        gpus: int = 0  # CPU by default
+    ) -> float:
+        """
+        Score a single translation.
+        
+        Args:
+            source_text: Original text
+            translation: Translated text
+            batch_size: Batch size for inference
+            gpus: Number of GPUs (0 for CPU)
+        
+        Returns:
+            Quality score between 0.0 and 1.0
+        """
+        self._load_model()
+        
+        data = [{
+            "src": source_text,
+            "mt": translation
+        }]
+        
+        output = self.model.predict(data, batch_size=batch_size, gpus=gpus)
+        
+        # COMETKiwi returns scores in output.scores (already 0-1 range)
+        return float(output.scores[0])
+    
+    def score_batch(
+        self,
+        pairs: List[Dict[str, str]],
+        batch_size: int = 8,
+        gpus: int = 0
+    ) -> List[float]:
+        """
+        Score multiple translations efficiently.
+        
+        Args:
+            pairs: List of {"src": ..., "mt": ...} dicts
+            batch_size: Batch size for inference
+            gpus: Number of GPUs
+        
+        Returns:
+            List of quality scores
+        """
+        self._load_model()
+        
+        if not pairs:
+            return []
+        
+        output = self.model.predict(pairs, batch_size=batch_size, gpus=gpus)
+        return [float(s) for s in output.scores]
+
+
+# ============================================================================
+# Back-Translation Verifier (SECONDARY)
+# ============================================================================
+
+class BackTranslationVerifier:
+    """
+    Verify translation quality via round-trip translation.
+    
+    Process: source -> translation -> back_translation
+    Then compare source with back_translation using semantic similarity.
+    """
+    
+    def __init__(self, translator_func: Callable):
+        """
+        Args:
+            translator_func: Function that takes (text, src_lang, tgt_lang)
+                           and returns translated text
+        """
+        self.translate = translator_func
+    
+    def verify(
+        self,
+        source_text: str,
+        translation: str,
+        src_lang: str,
+        tgt_lang: str
+    ) -> Dict:
+        """
+        Perform back-translation verification.
+        
+        Args:
+            source_text: Original text
+            translation: Forward translation
+            src_lang: Source language code (e.g., "eng_Latn")
+            tgt_lang: Target language code (e.g., "khm_Khmr")
+        
+        Returns:
+            Dict with similarity score and quality flags
+        """
+        try:
+            # Back-translate: tgt_lang -> src_lang
+            back_translation = self.translate(translation, tgt_lang, src_lang)
+            
+            # Compute similarity
+            similarity = self._compute_similarity(source_text, back_translation)
+            
+            # Determine quality flag
+            if similarity >= 0.7:
+                quality_flag = "good"
+            elif similarity >= 0.5:
+                quality_flag = "acceptable"
+            elif similarity >= 0.3:
+                quality_flag = "concerning"
+            else:
+                quality_flag = "poor"
+            
+            return {
+                "similarity": similarity,
+                "back_translation": back_translation,
+                "quality_flag": quality_flag
+            }
+            
+        except Exception as e:
+            logger.warning(f"Back-translation failed: {e}")
+            return {
+                "similarity": 0.5,  # Neutral fallback
+                "back_translation": None,
+                "quality_flag": "unknown"
+            }
+    
+    def _compute_similarity(self, text1: str, text2: str) -> float:
+        """
+        Compute semantic similarity between two texts.
+        Uses character-level similarity as a lightweight proxy.
+        """
+        # Normalize texts
+        t1 = text1.lower().strip()
+        t2 = text2.lower().strip()
+        
+        # Use SequenceMatcher for character-level similarity
+        # This is a lightweight alternative to embedding-based similarity
+        ratio = SequenceMatcher(None, t1, t2).ratio()
+        
+        return ratio
+
+
+# ============================================================================
+# Legal Terminology Checker (DOMAIN)
+# ============================================================================
+
+class LegalTerminologyChecker:
+    """
+    Check coverage and accuracy of legal terminology.
+    
+    Verifies that domain-specific terms are translated correctly
+    according to the legal glossary.
+    """
+    
+    def __init__(self, glossary: Dict[str, str]):
+        """
+        Args:
+            glossary: Dict mapping source terms to target terms
+        """
+        self.glossary = glossary
+        # Pre-compile patterns for efficiency
+        self._source_patterns = {
+            term: re.compile(r'\b' + re.escape(term) + r'\b', re.IGNORECASE)
+            for term in glossary.keys()
+        }
+    
+    def check(
+        self,
+        source_text: str,
+        translation: str,
+        direction: str = "en2km"
+    ) -> Dict:
+        """
+        Check terminology coverage and accuracy.
+        
+        Args:
+            source_text: Original text
+            translation: Translated text
+            direction: "en2km" or "km2en"
+        
+        Returns:
+            Dict with coverage score and term details
+        """
+        found_terms = []
+        correct_terms = []
+        missing_terms = []
+        
+        # For en2km: source terms are English (glossary keys)
+        # For km2en: we'd need reverse glossary
+        
+        for term, expected_translation in self.glossary.items():
+            pattern = self._source_patterns.get(term)
+            if pattern and pattern.search(source_text):
+                found_terms.append(term)
+                
+                # Check if expected translation appears in output
+                if expected_translation.lower() in translation.lower():
+                    correct_terms.append(term)
+                else:
+                    missing_terms.append({
+                        "term": term,
+                        "expected": expected_translation
+                    })
+        
+        # Calculate coverage score
+        if found_terms:
+            coverage = len(correct_terms) / len(found_terms)
+        else:
+            coverage = 1.0  # No terms to check = full coverage
+        
+        return {
+            "coverage": coverage,
+            "found_terms": len(found_terms),
+            "correct_terms": len(correct_terms),
+            "missing_terms": missing_terms
+        }
+
+
+# ============================================================================
+# Lightweight Scorer (FALLBACK)
+# ============================================================================
+
+class LightweightScorer:
+    """
+    Fast heuristic-based scoring when COMETKiwi is too slow.
+    
+    Uses:
+    - Length ratio (source vs translation)
+    - Character coverage
+    - Basic sanity checks
+    
+    ~50ms per sentence vs ~3-5s for COMETKiwi
+    """
+    
+    # Expected length ratios (empirically determined)
+    LENGTH_RATIOS = {
+        "en2km": (0.8, 2.5),   # Khmer is often longer due to syllabic script
+        "km2en": (0.4, 1.2)    # English is often shorter
+    }
+    
+    def score(
+        self,
+        source_text: str,
+        translation: str,
+        direction: str = "en2km"
+    ) -> Dict:
+        """
+        Compute lightweight confidence score.
+        
+        Returns:
+            Dict with score and explanation
+        """
+        scores = []
+        explanations = []
+        
+        # 1. Length ratio check
+        src_len = len(source_text)
+        tgt_len = len(translation)
+        
+        if src_len > 0:
+            ratio = tgt_len / src_len
+            min_ratio, max_ratio = self.LENGTH_RATIOS.get(direction, (0.5, 2.0))
+            
+            if min_ratio <= ratio <= max_ratio:
+                length_score = 1.0
+            else:
+                # Penalize based on how far outside range
+                if ratio < min_ratio:
+                    length_score = max(0.3, ratio / min_ratio)
+                    explanations.append(f"Translation may be too short (ratio: {ratio:.2f})")
+                else:
+                    length_score = max(0.3, max_ratio / ratio)
+                    explanations.append(f"Translation may be too long (ratio: {ratio:.2f})")
+            
+            scores.append(length_score)
+        
+        # 2. Empty/trivial check
+        if not translation.strip():
+            return {
+                "score": 0.0,
+                "explanations": ["Translation is empty"]
+            }
+        
+        # 3. Repetition check
+        words = translation.split()
+        if len(words) > 3:
+            unique_ratio = len(set(words)) / len(words)
+            if unique_ratio < 0.3:
+                scores.append(0.3)
+                explanations.append("High word repetition detected")
+            else:
+                scores.append(min(1.0, unique_ratio + 0.3))
+        
+        # 4. Script check for Khmer output
+        if direction == "en2km":
+            khmer_chars = sum(1 for c in translation if '\u1780' <= c <= '\u17FF')
+            khmer_ratio = khmer_chars / len(translation) if translation else 0
+            
+            if khmer_ratio < 0.3:
+                scores.append(0.5)
+                explanations.append("Low Khmer script ratio")
+            else:
+                scores.append(1.0)
+        
+        # Average all scores
+        final_score = sum(scores) / len(scores) if scores else 0.5
+        
+        return {
+            "score": final_score,
+            "explanations": explanations
+        }
+
+
+# ============================================================================
+# Transparency Scorer (ENSEMBLE)
+# ============================================================================
+
+class TransparencyScorer:
+    """
+    Ensemble confidence scorer combining multiple signals.
+    
+    Weights based on empirical findings from QE literature:
+    - COMETKiwi: 60% (primary, correlates with human judgments)
+    - Back-translation: 25% (catches semantic drift)
+    - Terminology: 15% (domain-specific quality)
+    """
+    
+    # Thresholds for quality levels
+    THRESHOLDS = {
+        "excellent": 0.85,
+        "good": 0.70,
+        "acceptable": 0.55,
+        "low": 0.40
+        # Below 0.40 = "very_low"
+    }
+    
+    # Weights for ensemble (sum to 1.0)
+    WEIGHTS = {
+        "comet_kiwi": 0.60,
+        "back_translation": 0.25,
+        "terminology": 0.15
+    }
+    
+    # Legal context requires higher threshold for human review
+    LEGAL_REVIEW_THRESHOLD = 0.75
+    
+    def __init__(
+        self,
+        translator_func: Optional[Callable] = None,
+        glossary: Optional[Dict[str, str]] = None,
+        use_comet: bool = True,
+        use_back_translation: bool = True,
+        use_terminology: bool = True
+    ):
+        """
+        Initialize scorer with configurable components.
+        
+        Args:
+            translator_func: Translation function for back-translation
+            glossary: Legal glossary for terminology checking
+            use_comet: Whether to use COMETKiwi
+            use_back_translation: Whether to use back-translation verification
+            use_terminology: Whether to check legal terminology
+        """
+        self.comet_scorer = COMETKiwiScorer() if use_comet else None
+        
+        self.bt_verifier = (
+            BackTranslationVerifier(translator_func) 
+            if use_back_translation and translator_func 
+            else None
+        )
+        
+        self.term_checker = (
+            LegalTerminologyChecker(glossary) 
+            if use_terminology and glossary 
+            else None
+        )
+        
+        self.lightweight_scorer = LightweightScorer()
+    
+    def score(
+        self,
+        source_text: str,
+        translation: str,
+        direction: str = "en2km",
+        detailed: bool = True
+    ) -> ConfidenceResult:
+        """
+        Compute ensemble confidence score.
+        
+        Args:
+            source_text: Original text
+            translation: Translated text
+            direction: "en2km" or "km2en"
+            detailed: Whether to compute all signals (slower but more accurate)
+        
+        Returns:
+            ConfidenceResult with overall score and breakdown
+        """
+        components = {}
+        explanations = []
+        
+        # 1. COMETKiwi (PRIMARY)
+        if self.comet_scorer and detailed:
+            try:
+                comet_score = self.comet_scorer.score(source_text, translation)
+                components["comet_kiwi"] = comet_score
+                
+                if comet_score < 0.5:
+                    explanations.append(
+                        f"Neural QE indicates low quality ({comet_score:.2f})"
+                    )
+                elif comet_score > 0.8:
+                    explanations.append(
+                        f"Neural QE indicates high quality ({comet_score:.2f})"
+                    )
+            except Exception as e:
+                logger.warning(f"COMETKiwi scoring failed: {e}")
+                # Fall back to lightweight
+                lw_result = self.lightweight_scorer.score(
+                    source_text, translation, direction
+                )
+                components["comet_kiwi"] = lw_result["score"]
+                explanations.extend(lw_result["explanations"])
+        else:
+            # Use lightweight scorer as fallback
+            lw_result = self.lightweight_scorer.score(
+                source_text, translation, direction
+            )
+            components["comet_kiwi"] = lw_result["score"]
+            explanations.extend(lw_result["explanations"])
+        
+        # 2. Back-translation (SECONDARY)
+        if self.bt_verifier and detailed:
+            try:
+                src_lang = "eng_Latn" if direction == "en2km" else "khm_Khmr"
+                tgt_lang = "khm_Khmr" if direction == "en2km" else "eng_Latn"
+                
+                bt_result = self.bt_verifier.verify(
+                    source_text, translation, src_lang, tgt_lang
+                )
+                components["back_translation"] = bt_result["similarity"]
+                
+                if bt_result["quality_flag"] in ["concerning", "poor"]:
+                    explanations.append(
+                        f"Back-translation diverges ({bt_result['quality_flag']}): "
+                        f"similarity={bt_result['similarity']:.2f}"
+                    )
+            except Exception as e:
+                logger.warning(f"Back-translation verification failed: {e}")
+                components["back_translation"] = 0.5
+        else:
+            components["back_translation"] = 0.5  # Neutral
+        
+        # 3. Terminology (DOMAIN)
+        if self.term_checker:
+            try:
+                term_result = self.term_checker.check(
+                    source_text, translation, direction
+                )
+                components["terminology"] = term_result["coverage"]
+                
+                if term_result["missing_terms"]:
+                    missing_list = [t["term"] for t in term_result["missing_terms"][:3]]
+                    explanations.append(
+                        f"Missing legal terms: {', '.join(missing_list)}"
+                    )
+            except Exception as e:
+                logger.warning(f"Terminology check failed: {e}")
+                components["terminology"] = 0.5
+        else:
+            components["terminology"] = 0.5  # Neutral
+        
+        # Compute weighted average
+        overall_score = sum(
+            components.get(k, 0.5) * v 
+            for k, v in self.WEIGHTS.items()
+        )
+        
+        # Determine quality level
+        quality_level = "very_low"
+        for level, threshold in sorted(
+            self.THRESHOLDS.items(), 
+            key=lambda x: x[1], 
+            reverse=True
+        ):
+            if overall_score >= threshold:
+                quality_level = level
+                break
+        
+        # Legal context: recommend human review below threshold
+        human_review = overall_score < self.LEGAL_REVIEW_THRESHOLD
+        
+        if human_review and not any("review" in e.lower() for e in explanations):
+            explanations.append(
+                "Human review recommended for legal accuracy"
+            )
+        
+        return ConfidenceResult(
+            overall_score=round(overall_score, 3),
+            quality_level=quality_level,
+            human_review_recommended=human_review,
+            components=components,
+            explanations=explanations
+        )
+    
+    def score_fast(
+        self,
+        source_text: str,
+        translation: str,
+        direction: str = "en2km"
+    ) -> ConfidenceResult:
+        """
+        Fast scoring without COMETKiwi (uses lightweight scorer only).
+        ~50ms vs ~3-5s for full scoring.
+        """
+        return self.score(
+            source_text, translation, direction, detailed=False
+        )
+
+
+# ============================================================================
+# Utility Functions
+# ============================================================================
+
+def create_default_scorer(
+    translator_func: Optional[Callable] = None,
+    glossary: Optional[Dict[str, str]] = None
+) -> TransparencyScorer:
+    """
+    Create a scorer with sensible defaults for Khmer Legal Bridge.
+    
+    Args:
+        translator_func: Translation function for back-translation
+        glossary: Legal glossary (optional)
+    
+    Returns:
+        Configured TransparencyScorer
+    """
+    return TransparencyScorer(
+        translator_func=translator_func,
+        glossary=glossary,
+        use_comet=True,
+        use_back_translation=True,
+        use_terminology=bool(glossary)
+    )
+
+
+# Default legal glossary (subset for demo)
+DEFAULT_LEGAL_GLOSSARY = {
+    # Criminal procedure
+    "arrest": "ការចាប់ខ្លួន",
+    "detention": "ការឃុំខ្លួន",
+    "bail": "ការដាក់ធានា",
+    "prosecutor": "រដ្ឋអាជ្ញា",
+    "defendant": "ជនជាប់ចោទ",
+    "verdict": "សាលក្រម",
+    "sentence": "ទោស",
+    "appeal": "បណ្ដឹងឧទ្ធរណ៍",
+    
+    # Juvenile justice
+    "minor": "អនីតិជន",
+    "juvenile": "អនីតិជន",
+    "guardian": "អាណាព្យាបាល",
+    "rehabilitation": "ការអប់រំកែប្រែ",
+    "diversion": "ការបង្វែរក្រៅប្រព័ន្ធតុលាការ",
+    
+    # Human rights
+    "asylum": "សិទ្ធិជ្រកកោន",
+    "refugee": "ជនភៀសខ្លួន",
+    "persecution": "ការធ្វើទារុណកម្ម",
+    "torture": "ទារុណកម្ម",
+    "due process": "នីតិវិធីត្រឹមត្រូវ",
+    
+    # General legal
+    "court": "តុលាការ",
+    "judge": "ចៅក្រម",
+    "lawyer": "មេធាវី",
+    "evidence": "ភស្តុតាង",
+    "witness": "សាក្សី",
+    "testimony": "សក្ខីកម្ម",
+    "rights": "សិទ្ធិ",
+    "law": "ច្បាប់",
+}