bias_prediction_engine.py

"""

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))