Upload bias_prediction_engine.py

bias_prediction_engine.py

@@ -0,0 +1,612 @@
+"""
+Bias Detection and Outcome Prediction Engine using InLegalBERT
+================================================================
+
+This module provides:
+1. Document/Text bias detection (gender, region, caste, etc.)
+2. RAG output bias detection (tone, interpretive bias)
+3. Systemic/Statistical bias analysis
+4. Legal outcome prediction with confidence scores
+
+Model: InLegalBERT (Hugging Face pretrained for Indian legal cases)
+"""
+
+import torch
+import numpy as np
+from transformers import AutoTokenizer, AutoModelForSequenceClassification, AutoModel
+from typing import Dict, List, Any, Optional, Union
+import re
+from collections import Counter
+import json
+from datetime import datetime
+import warnings
+warnings.filterwarnings('ignore')
+
+# ============================================================================
+# MODEL INITIALIZATION
+# ============================================================================
+
+class InLegalBERTEngine:
+    """
+    Main engine for bias detection and outcome prediction using InLegalBERT
+    """
+    
+    def __init__(self, model_name: str = "law-ai/InLegalBERT"):
+        """
+        Initialize the InLegalBERT model and tokenizer
+        
+        Args:
+            model_name: HuggingFace model identifier (use your fine-tuned model path)
+        """
+        print(f"Loading InLegalBERT model: {model_name}")
+        
+        # Load tokenizer and base model for embeddings
+        # TODO: Replace "law-ai/InLegalBERT" with your fine-tuned model path
+        # Example: "your-username/inlegalbert-bias-finetuned"
+        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
+        self.base_model = AutoModel.from_pretrained(model_name)
+        
+        # Set device
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.base_model.to(self.device)
+        self.base_model.eval()
+        
+        # Bias detection keywords (Indian legal context)
+        self.bias_keywords = {
+            'gender': [
+                'woman', 'women', 'girl', 'female', 'lady', 'wife', 'mother',
+                'man', 'men', 'boy', 'male', 'husband', 'father', 'manhood', 'womanhood'
+            ],
+            'caste': [
+                'scheduled caste', 'sc', 'st', 'scheduled tribe', 'obc', 'backward class',
+                'dalit', 'brahmin', 'upper caste', 'lower caste', 'caste', 'jati'
+            ],
+            'religion': [
+                'hindu', 'muslim', 'christian', 'sikh', 'buddhist', 'jain',
+                'religious', 'communal', 'minority', 'majority community'
+            ],
+            'region': [
+                'north', 'south', 'east', 'west', 'rural', 'urban', 'tribal',
+                'metropolitan', 'village', 'city', 'state', 'region'
+            ],
+            'socioeconomic': [
+                'poor', 'rich', 'wealthy', 'poverty', 'income', 'economically',
+                'below poverty line', 'bpl', 'weaker section', 'privileged'
+            ],
+            'age': [
+                'minor', 'juvenile', 'child', 'elderly', 'senior citizen', 'youth',
+                'old', 'young', 'aged'
+            ]
+        }
+        
+        print(f"Model loaded successfully on {self.device}")
+    
+    # ========================================================================
+    # UTILITY FUNCTIONS
+    # ========================================================================
+    
+    def get_embeddings(self, text: str) -> torch.Tensor:
+        """
+        Get BERT embeddings for input text
+        
+        Args:
+            text: Input text string
+            
+        Returns:
+            torch.Tensor: Embedding vector
+        """
+        # Tokenize
+        inputs = self.tokenizer(
+            text,
+            return_tensors="pt",
+            truncation=True,
+            max_length=512,
+            padding=True
+        ).to(self.device)
+        
+        # Get embeddings
+        with torch.no_grad():
+            outputs = self.base_model(**inputs)
+            # Use CLS token embedding (first token)
+            embeddings = outputs.last_hidden_state[:, 0, :]
+        
+        return embeddings
+    
+    def compute_bias_score(self, text: str, bias_type: str) -> float:
+        """
+        Compute bias score for a specific bias type using keyword frequency
+        and contextual analysis
+        
+        Args:
+            text: Input text
+            bias_type: Type of bias (gender, caste, etc.)
+            
+        Returns:
+            float: Bias score between 0 and 1
+        """
+        text_lower = text.lower()
+        keywords = self.bias_keywords.get(bias_type, [])
+        
+        # Count keyword occurrences
+        keyword_count = sum(text_lower.count(keyword) for keyword in keywords)
+        
+        # Normalize by text length (words)
+        word_count = len(text.split())
+        if word_count == 0:
+            return 0.0
+        
+        # Calculate frequency-based score
+        frequency_score = min(keyword_count / word_count * 10, 1.0)
+        
+        # Get contextual score using embeddings (simplified)
+        # In production, use a fine-tuned classifier
+        contextual_score = frequency_score * 0.8  # Simplified
+        
+        return round(contextual_score, 3)
+    
+    # ========================================================================
+    # 1. DOCUMENT/TEXT BIAS DETECTION
+    # ========================================================================
+    
+    def detect_document_bias(self, text: str, threshold: float = 0.15) -> Dict[str, Any]:
+        """
+        Detect various biases in legal documents/FIRs/judgments
+        
+        Args:
+            text: Legal document text
+            threshold: Minimum score to flag a bias (default 0.15)
+            
+        Returns:
+            Dict containing bias flags and detailed scores
+        """
+        bias_scores = {}
+        bias_flags = []
+        
+        # Analyze each bias type
+        for bias_type in self.bias_keywords.keys():
+            score = self.compute_bias_score(text, bias_type)
+            bias_scores[bias_type] = score
+            
+            if score >= threshold:
+                bias_flags.append(bias_type)
+        
+        # Determine severity levels
+        bias_details = []
+        for bias_type, score in bias_scores.items():
+            if score >= threshold:
+                severity = "high" if score >= 0.4 else "medium" if score >= 0.25 else "low"
+                bias_details.append({
+                    "type": bias_type,
+                    "severity": severity,
+                    "score": score,
+                    "description": f"{bias_type.capitalize()} bias detected based on keyword analysis and context"
+                })
+        
+        return {
+            "biasFlags_text": bias_flags,
+            "bias_scores": bias_scores,
+            "bias_details": bias_details,
+            "overall_bias_score": round(np.mean(list(bias_scores.values())), 3),
+            "analysis_timestamp": datetime.now().isoformat()
+        }
+    
+    # ========================================================================
+    # 2. RAG OUTPUT BIAS DETECTION
+    # ========================================================================
+    
+    def detect_rag_output_bias(self, 
+                               rag_summary: str, 
+                               source_documents: List[str]) -> Dict[str, Any]:
+        """
+        Detect bias in AI-generated RAG summaries/reasoning
+        
+        Args:
+            rag_summary: AI-generated summary or reasoning
+            source_documents: Original source documents used for RAG
+            
+        Returns:
+            Dict containing RAG-specific bias flags
+        """
+        bias_flags = []
+        bias_details = []
+        
+        # Get embeddings
+        summary_emb = self.get_embeddings(rag_summary)
+        source_embs = [self.get_embeddings(doc) for doc in source_documents[:5]]  # Limit to 5
+        
+        # 1. TONE BIAS - Check if summary tone differs from sources
+        if source_embs:
+            avg_source_emb = torch.mean(torch.stack(source_embs), dim=0)
+            # Cosine similarity
+            similarity = torch.nn.functional.cosine_similarity(summary_emb, avg_source_emb)
+            
+            if similarity < 0.7:  # Low similarity indicates tone shift
+                bias_flags.append("tone_bias")
+                bias_details.append({
+                    "type": "tone_bias",
+                    "severity": "medium",
+                    "score": round(1 - similarity.item(), 3),
+                    "description": "AI summary tone differs significantly from source documents"
+                })
+        
+        # 2. INTERPRETIVE BIAS - Check for subjective language
+        subjective_words = [
+            'clearly', 'obviously', 'undoubtedly', 'certainly', 'definitely',
+            'surely', 'apparently', 'seemingly', 'arguably', 'presumably'
+        ]
+        summary_lower = rag_summary.lower()
+        subjective_count = sum(summary_lower.count(word) for word in subjective_words)
+        
+        if subjective_count > 2:
+            bias_flags.append("interpretive_bias")
+            bias_details.append({
+                "type": "interpretive_bias",
+                "severity": "medium" if subjective_count > 4 else "low",
+                "score": round(min(subjective_count / 10, 1.0), 3),
+                "description": f"Summary contains {subjective_count} subjective/interpretive terms"
+            })
+        
+        # 3. SELECTIVITY BIAS - Check if summary over-represents certain aspects
+        # Count mentions of different legal aspects
+        aspects = {
+            'procedural': ['procedure', 'process', 'filing', 'hearing', 'appeal'],
+            'substantive': ['law', 'statute', 'provision', 'section', 'act'],
+            'factual': ['fact', 'evidence', 'witness', 'testimony', 'statement']
+        }
+        
+        aspect_counts = {k: sum(summary_lower.count(w) for w in v) for k, v in aspects.items()}
+        max_count = max(aspect_counts.values()) if aspect_counts.values() else 1
+        
+        if max_count > 5 and any(count < max_count * 0.3 for count in aspect_counts.values()):
+            bias_flags.append("selectivity_bias")
+            bias_details.append({
+                "type": "selectivity_bias",
+                "severity": "low",
+                "score": 0.4,
+                "description": "Summary may over-emphasize certain legal aspects"
+            })
+        
+        return {
+            "biasFlags_output": bias_flags,
+            "bias_details": bias_details,
+            "analysis_timestamp": datetime.now().isoformat()
+        }
+    
+    # ========================================================================
+    # 3. SYSTEMIC/STATISTICAL BIAS DETECTION
+    # ========================================================================
+    
+    def detect_systemic_bias(self, 
+                            historical_cases: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """
+        Analyze systemic and statistical biases from historical case data
+        
+        Args:
+            historical_cases: List of case dictionaries with keys:
+                - outcome: str (e.g., "conviction", "acquittal")
+                - gender: str (optional)
+                - region: str (optional)
+                - caste: str (optional)
+                - case_type: str
+                - year: int
+                
+        Returns:
+            Dict containing systemic bias metrics and dashboard data
+        """
+        if not historical_cases:
+            return {"error": "No historical cases provided"}
+        
+        # Initialize analytics
+        outcome_by_gender = {}
+        outcome_by_region = {}
+        outcome_by_caste = {}
+        outcome_by_year = {}
+        
+        # Process cases
+        for case in historical_cases:
+            outcome = case.get('outcome', 'unknown')
+            
+            # Gender analysis
+            if 'gender' in case:
+                gender = case['gender']
+                if gender not in outcome_by_gender:
+                    outcome_by_gender[gender] = []
+                outcome_by_gender[gender].append(outcome)
+            
+            # Region analysis
+            if 'region' in case:
+                region = case['region']
+                if region not in outcome_by_region:
+                    outcome_by_region[region] = []
+                outcome_by_region[region].append(outcome)
+            
+            # Caste analysis
+            if 'caste' in case:
+                caste = case['caste']
+                if caste not in outcome_by_caste:
+                    outcome_by_caste[caste] = []
+                outcome_by_caste[caste].append(outcome)
+            
+            # Temporal analysis
+            if 'year' in case:
+                year = case['year']
+                if year not in outcome_by_year:
+                    outcome_by_year[year] = []
+                outcome_by_year[year].append(outcome)
+        
+        # Calculate disparity metrics
+        def calculate_disparity(outcome_dict: Dict) -> Dict:
+            """Calculate outcome disparities"""
+            disparity_data = {}
+            for category, outcomes in outcome_dict.items():
+                total = len(outcomes)
+                if total > 0:
+                    conviction_rate = outcomes.count('conviction') / total
+                    disparity_data[category] = {
+                        'total_cases': total,
+                        'conviction_rate': round(conviction_rate, 3),
+                        'acquittal_rate': round(outcomes.count('acquittal') / total, 3)
+                    }
+            return disparity_data
+        
+        gender_disparity = calculate_disparity(outcome_by_gender)
+        region_disparity = calculate_disparity(outcome_by_region)
+        caste_disparity = calculate_disparity(outcome_by_caste)
+        
+        # Detect significant disparities
+        bias_flags = []
+        
+        if gender_disparity:
+            rates = [d['conviction_rate'] for d in gender_disparity.values()]
+            if max(rates) - min(rates) > 0.15:
+                bias_flags.append("gender_disparity")
+        
+        if region_disparity:
+            rates = [d['conviction_rate'] for d in region_disparity.values()]
+            if max(rates) - min(rates) > 0.15:
+                bias_flags.append("regional_disparity")
+        
+        if caste_disparity:
+            rates = [d['conviction_rate'] for d in caste_disparity.values()]
+            if max(rates) - min(rates) > 0.15:
+                bias_flags.append("caste_disparity")
+        
+        # Generate dashboard-ready data
+        dashboard_data = {
+            "summary_metrics": {
+                "total_cases_analyzed": len(historical_cases),
+                "overall_conviction_rate": round(
+                    sum(1 for c in historical_cases if c.get('outcome') == 'conviction') / len(historical_cases),
+                    3
+                ),
+                "bias_flags_detected": len(bias_flags)
+            },
+            "gender_analysis": {
+                "disparity_data": gender_disparity,
+                "chart_data": [
+                    {"category": k, "conviction_rate": v['conviction_rate']} 
+                    for k, v in gender_disparity.items()
+                ]
+            },
+            "regional_analysis": {
+                "disparity_data": region_disparity,
+                "chart_data": [
+                    {"category": k, "conviction_rate": v['conviction_rate']} 
+                    for k, v in region_disparity.items()
+                ]
+            },
+            "caste_analysis": {
+                "disparity_data": caste_disparity,
+                "chart_data": [
+                    {"category": k, "conviction_rate": v['conviction_rate']} 
+                    for k, v in caste_disparity.items()
+                ]
+            },
+            "temporal_trends": {
+                "by_year": {
+                    year: {
+                        'total': len(outcomes),
+                        'conviction_rate': round(outcomes.count('conviction') / len(outcomes), 3)
+                    }
+                    for year, outcomes in outcome_by_year.items()
+                }
+            }
+        }
+        
+        return {
+            "systemic_bias_flags": bias_flags,
+            "biasDashboardData": dashboard_data,
+            "analysis_timestamp": datetime.now().isoformat()
+        }
+    
+    # ========================================================================
+    # 4. OUTCOME PREDICTION
+    # ========================================================================
+    
+    def predict_outcome(self, 
+                       case_text: str,
+                       case_metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        """
+        Predict legal case outcome using InLegalBERT embeddings and heuristics
+        
+        Args:
+            case_text: Full case text (FIR, facts, arguments, etc.)
+            case_metadata: Optional metadata (case_type, jurisdiction, etc.)
+            
+        Returns:
+            Dict containing prediction, confidence, and justification
+        """
+        # Get text embeddings
+        embeddings = self.get_embeddings(case_text)
+        
+        # Keyword-based prediction (simplified - in production use fine-tuned classifier)
+        conviction_keywords = [
+            'guilty', 'convicted', 'evidence proves', 'beyond reasonable doubt',
+            'establish', 'proven', 'corroborated', 'substantiated'
+        ]
+        acquittal_keywords = [
+            'not guilty', 'acquitted', 'benefit of doubt', 'insufficient evidence',
+            'failed to prove', 'contradictory', 'unreliable', 'doubt'
+        ]
+        
+        text_lower = case_text.lower()
+        conviction_score = sum(text_lower.count(kw) for kw in conviction_keywords)
+        acquittal_score = sum(text_lower.count(kw) for kw in acquittal_keywords)
+        
+        # Calculate prediction
+        total_score = conviction_score + acquittal_score
+        if total_score == 0:
+            # No strong indicators, use neutral prediction
+            predicted_outcome = "uncertain"
+            confidence_score = 0.5
+            justification = "Insufficient textual indicators for confident prediction"
+        else:
+            conviction_prob = conviction_score / total_score
+            
+            if conviction_prob > 0.6:
+                predicted_outcome = "conviction"
+                confidence_score = round(conviction_prob, 3)
+                justification = f"Text analysis shows {conviction_score} conviction indicators vs {acquittal_score} acquittal indicators"
+            elif conviction_prob < 0.4:
+                predicted_outcome = "acquittal"
+                confidence_score = round(1 - conviction_prob, 3)
+                justification = f"Text analysis shows {acquittal_score} acquittal indicators vs {conviction_score} conviction indicators"
+            else:
+                predicted_outcome = "uncertain"
+                confidence_score = 0.5
+                justification = "Mixed indicators suggest uncertain outcome"
+        
+        # Adjust for metadata if provided
+        if case_metadata:
+            case_type = case_metadata.get('case_type', '').lower()
+            
+            # Example adjustments (customize based on domain knowledge)
+            if 'bail' in case_type:
+                if predicted_outcome == "conviction":
+                    predicted_outcome = "bail_denied"
+                    justification += "; Bail application context considered"
+                elif predicted_outcome == "acquittal":
+                    predicted_outcome = "bail_granted"
+                    justification += "; Bail application context considered"
+        
+        # Confidence level categorization
+        if confidence_score >= 0.75:
+            confidence_level = "high"
+        elif confidence_score >= 0.5:
+            confidence_level = "medium"
+        else:
+            confidence_level = "low"
+        
+        return {
+            "predictedOutcome": predicted_outcome,
+            "confidenceScore": confidence_score,
+            "confidenceLevel": confidence_level,
+            "justification": justification,
+            "embedding_norm": float(torch.norm(embeddings).item()),
+            "analysis_timestamp": datetime.now().isoformat()
+        }
+
+# ============================================================================
+# API INTERFACE FUNCTIONS
+# ============================================================================
+
+# Global model instance (loaded once)
+_model_instance = None
+
+def get_model() -> InLegalBERTEngine:
+    """Get or create model instance (singleton pattern)"""
+    global _model_instance
+    if _model_instance is None:
+        _model_instance = InLegalBERTEngine()
+    return _model_instance
+
+
+def analyze_legal_case(
+    case_text: str,
+    rag_summary: Optional[str] = None,
+    source_documents: Optional[List[str]] = None,
+    historical_cases: Optional[List[Dict]] = None,
+    case_metadata: Optional[Dict] = None
+) -> Dict[str, Any]:
+    """
+    Main API function for comprehensive legal case analysis
+    
+    Args:
+        case_text: Legal document/FIR/judgment text
+        rag_summary: AI-generated summary (for RAG bias detection)
+        source_documents: Source docs used for RAG (for RAG bias detection)
+        historical_cases: Historical case data (for systemic bias analysis)
+        case_metadata: Case metadata for outcome prediction
+        
+    Returns:
+        JSON-serializable dict with all analysis results
+    """
+    model = get_model()
+    
+    results = {
+        "status": "success",
+        "analysis_id": f"analysis_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
+        "timestamp": datetime.now().isoformat()
+    }
+    
+    # 1. Document bias detection
+    if case_text:
+        results["document_bias"] = model.detect_document_bias(case_text)
+    
+    # 2. RAG output bias detection
+    if rag_summary and source_documents:
+        results["rag_bias"] = model.detect_rag_output_bias(rag_summary, source_documents)
+    
+    # 3. Systemic bias analysis
+    if historical_cases:
+        results["systemic_bias"] = model.detect_systemic_bias(historical_cases)
+    
+    # 4. Outcome prediction
+    if case_text:
+        results["outcome_prediction"] = model.predict_outcome(case_text, case_metadata)
+    
+    return results
+
+
+# ============================================================================
+# EXAMPLE USAGE
+# ============================================================================
+
+if __name__ == "__main__":
+    # Example legal case text
+    sample_case = """
+    The accused, a 35-year-old woman from rural Maharashtra, was charged under 
+    Section 302 IPC for alleged murder. The prosecution's case relies heavily on 
+    circumstantial evidence. The witness testimonies are contradictory, and the 
+    forensic evidence is inconclusive. The accused belongs to a scheduled caste 
+    community. The defense argues that there is insufficient evidence to establish 
+    guilt beyond reasonable doubt.
+    """
+    
+    # Example RAG summary
+    sample_rag_summary = """
+    Clearly, the evidence points toward acquittal. The case obviously lacks 
+    substantial proof of guilt.
+    """
+    
+    # Example historical cases
+    sample_historical = [
+        {"outcome": "conviction", "gender": "male", "region": "urban", "year": 2020},
+        {"outcome": "acquittal", "gender": "female", "region": "rural", "year": 2020},
+        {"outcome": "conviction", "gender": "male", "region": "urban", "year": 2021},
+        {"outcome": "conviction", "gender": "female", "region": "urban", "year": 2021},
+    ]
+    
+    # Run comprehensive analysis
+    print("Running comprehensive legal analysis...\n")
+    results = analyze_legal_case(
+        case_text=sample_case,
+        rag_summary=sample_rag_summary,
+        source_documents=[sample_case],
+        historical_cases=sample_historical,
+        case_metadata={"case_type": "criminal", "jurisdiction": "Maharashtra"}
+    )
+    
+    # Print results
+    print(json.dumps(results, indent=2))
+