sofia_meta_cognition.py

#!/usr/bin/env python3
"""
SOFIA Meta-Cognition System
Provides self-awareness, error detection, and decision analysis capabilities
"""

import torch
import torch.nn as nn
import numpy as np
import json
import logging
from typing import Dict, List, Tuple, Optional, Any, Union
from datetime import datetime, timedelta
from collections import defaultdict, deque
import statistics
import re

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ConfidenceEstimator(nn.Module):
    """
    Estimates confidence scores for SOFIA's predictions
    """

    def __init__(self, embedding_dim: int = 768):
        super().__init__()
        self.confidence_head = nn.Sequential(
            nn.Linear(embedding_dim * 2, 256),
            nn.ReLU(),
            nn.Dropout(0.1),
            nn.Linear(256, 128),
            nn.ReLU(),
            nn.Linear(128, 1),
            nn.Sigmoid()  # Output between 0 and 1
        )

    def forward(self, embedding1: torch.Tensor, embedding2: torch.Tensor) -> torch.Tensor:
        """Estimate confidence for similarity prediction"""
        combined = torch.cat([embedding1, embedding2], dim=1)
        confidence = self.confidence_head(combined)
        return confidence.squeeze()

class ErrorDetector:
    """
    Detects and analyzes errors in SOFIA's predictions
    """

    def __init__(self, error_threshold: float = 0.3):
        self.error_threshold = error_threshold
        self.error_history = deque(maxlen=1000)
        self.error_patterns = defaultdict(int)

    def detect_error(self, prediction: float, ground_truth: float,
                    confidence: float, context: Dict[str, Any]) -> Dict[str, Any]:
        """
        Detect if a prediction contains an error

        Args:
            prediction: Model's similarity prediction (0-1)
            ground_truth: Actual similarity score (0-1)
            confidence: Model's confidence in prediction (0-1)
            context: Additional context information

        Returns:
            Error analysis dictionary
        """

        error_magnitude = abs(prediction - ground_truth)
        is_error = error_magnitude > self.error_threshold

        # Low confidence + high error = likely error
        confidence_weighted_error = error_magnitude * (1 - confidence)

        error_info = {
            'is_error': is_error,
            'error_magnitude': error_magnitude,
            'confidence_weighted_error': confidence_weighted_error,
            'prediction': prediction,
            'ground_truth': ground_truth,
            'confidence': confidence,
            'context': context,
            'timestamp': datetime.now().isoformat(),
            'error_type': self._classify_error(prediction, ground_truth, confidence)
        }

        if is_error:
            self.error_history.append(error_info)
            self._update_error_patterns(error_info)

        return error_info

    def _classify_error(self, prediction: float, ground_truth: float, confidence: float) -> str:
        """Classify the type of error"""
        error_mag = abs(prediction - ground_truth)

        if confidence < 0.3 and error_mag > 0.5:
            return "low_confidence_high_error"
        elif abs(prediction - 0.5) < 0.1 and abs(ground_truth - 0.5) > 0.3:
            return "neutral_prediction_bias"
        elif (prediction > 0.8 and ground_truth < 0.3) or (prediction < 0.2 and ground_truth > 0.7):
            return "extreme_misclassification"
        elif error_mag > 0.4:
            return "large_error"
        else:
            return "moderate_error"

    def _update_error_patterns(self, error_info: Dict[str, Any]):
        """Update error pattern statistics"""
        error_type = error_info['error_type']
        self.error_patterns[error_type] += 1

        # Analyze context patterns
        context = error_info.get('context', {})
        if 'text1_length' in context and 'text2_length' in context:
            length_ratio = context['text1_length'] / max(context['text2_length'], 1)
            if length_ratio > 3 or length_ratio < 0.33:
                self.error_patterns['length_mismatch'] += 1

        if 'domain' in context:
            domain = context['domain']
            self.error_patterns[f'domain_{domain}'] += 1

    def get_error_statistics(self) -> Dict[str, Any]:
        """Get comprehensive error statistics"""
        if not self.error_history:
            return {'total_errors': 0, 'error_rate': 0.0}

        total_predictions = len(self.error_history)
        errors = sum(1 for e in self.error_history if e['is_error'])
        error_rate = errors / total_predictions

        # Error magnitude statistics
        error_magnitudes = [e['error_magnitude'] for e in self.error_history if e['is_error']]
        avg_error_magnitude = statistics.mean(error_magnitudes) if error_magnitudes else 0

        # Confidence analysis
        confidence_when_wrong = [e['confidence'] for e in self.error_history if e['is_error']]
        avg_confidence_wrong = statistics.mean(confidence_when_wrong) if confidence_when_wrong else 0

        return {
            'total_errors': errors,
            'total_predictions': total_predictions,
            'error_rate': error_rate,
            'average_error_magnitude': avg_error_magnitude,
            'average_confidence_when_wrong': avg_confidence_wrong,
            'error_patterns': dict(self.error_patterns),
            'recent_errors': list(self.error_history)[-10:]  # Last 10 errors
        }

class DecisionAnalyzer:
    """
    Analyzes SOFIA's decision-making process and provides insights
    """

    def __init__(self):
        self.decision_history = deque(maxlen=2000)
        self.decision_patterns = defaultdict(lambda: defaultdict(int))

    def analyze_decision(self, query: str, results: List[Tuple[int, float]],
                        context: Dict[str, Any]) -> Dict[str, Any]:
        """
        Analyze a decision-making process

        Args:
            query: The input query
            results: List of (index, score) tuples
            context: Additional context

        Returns:
            Decision analysis
        """

        analysis = {
            'query': query,
            'top_result_score': results[0][1] if results else 0,
            'result_distribution': self._analyze_score_distribution(results),
            'query_characteristics': self._analyze_query(query),
            'decision_confidence': self._calculate_decision_confidence(results),
            'context': context,
            'timestamp': datetime.now().isoformat()
        }

        self.decision_history.append(analysis)
        self._update_decision_patterns(analysis)

        return analysis

    def _analyze_score_distribution(self, results: List[Tuple[int, float]]) -> Dict[str, Any]:
        """Analyze the distribution of similarity scores"""
        if not results:
            return {'distribution_type': 'no_results'}

        scores = [score for _, score in results]

        # Calculate statistics
        mean_score = statistics.mean(scores)
        std_score = statistics.stdev(scores) if len(scores) > 1 else 0
        max_score = max(scores)
        min_score = min(scores)

        # Classify distribution
        if std_score < 0.1:
            dist_type = 'uniform'
        elif max_score - min_score > 0.5:
            dist_type = 'wide_spread'
        elif mean_score > 0.7:
            dist_type = 'high_similarity'
        elif mean_score < 0.3:
            dist_type = 'low_similarity'
        else:
            dist_type = 'moderate_spread'

        return {
            'distribution_type': dist_type,
            'mean_score': mean_score,
            'std_score': std_score,
            'max_score': max_score,
            'min_score': min_score,
            'score_range': max_score - min_score
        }

    def _analyze_query(self, query: str) -> Dict[str, Any]:
        """Analyze query characteristics"""
        words = query.split()
        sentences = re.split(r'[.!?]+', query)

        return {
            'word_count': len(words),
            'sentence_count': len([s for s in sentences if s.strip()]),
            'avg_word_length': statistics.mean([len(word) for word in words]) if words else 0,
            'contains_questions': '?' in query,
            'contains_numbers': any(char.isdigit() for char in query),
            'is_short': len(words) < 5,
            'is_long': len(words) > 20
        }

    def _calculate_decision_confidence(self, results: List[Tuple[int, float]]) -> float:
        """Calculate confidence in the decision"""
        if not results or len(results) < 2:
            return 0.5  # Neutral confidence

        top_score = results[0][1]
        second_score = results[1][1]

        # Confidence based on margin between top and second result
        margin = top_score - second_score

        if margin > 0.3:
            confidence = 0.9
        elif margin > 0.2:
            confidence = 0.8
        elif margin > 0.1:
            confidence = 0.7
        elif margin > 0.05:
            confidence = 0.6
        else:
            confidence = 0.5

        return confidence

    def _update_decision_patterns(self, analysis: Dict[str, Any]):
        """Update decision pattern statistics"""
        query_chars = analysis['query_characteristics']

        # Track patterns by query type
        if query_chars['contains_questions']:
            self.decision_patterns['question_queries']['count'] += 1
            self.decision_patterns['question_queries']['avg_confidence'] = (
                (self.decision_patterns['question_queries'].get('avg_confidence', 0) * (
                    self.decision_patterns['question_queries']['count'] - 1) + analysis['decision_confidence']) /
                self.decision_patterns['question_queries']['count']
            )

        if query_chars['is_short']:
            self.decision_patterns['short_queries']['count'] += 1

        # Track distribution patterns
        dist_type = analysis['result_distribution']['distribution_type']
        self.decision_patterns['distributions'][dist_type] += 1

    def get_decision_insights(self) -> Dict[str, Any]:
        """Get insights from decision analysis"""
        if not self.decision_history:
            return {'total_decisions': 0}

        total_decisions = len(self.decision_history)
        recent_decisions = list(self.decision_history)[-100:]  # Last 100 decisions

        # Confidence analysis
        confidences = [d['decision_confidence'] for d in recent_decisions]
        avg_confidence = statistics.mean(confidences)

        # Query type analysis
        query_types = defaultdict(int)
        for decision in recent_decisions:
            chars = decision['query_characteristics']
            if chars['contains_questions']:
                query_types['questions'] += 1
            if chars['is_short']:
                query_types['short'] += 1
            if chars['is_long']:
                query_types['long'] += 1

        # Performance patterns
        high_confidence_decisions = sum(1 for c in confidences if c > 0.8)
        low_confidence_decisions = sum(1 for c in confidences if c < 0.6)

        return {
            'total_decisions': total_decisions,
            'average_confidence': avg_confidence,
            'high_confidence_rate': high_confidence_decisions / len(confidences),
            'low_confidence_rate': low_confidence_decisions / len(confidences),
            'query_type_distribution': dict(query_types),
            'decision_patterns': dict(self.decision_patterns)
        }

class MetaCognitiveSOFIA:
    """
    Main meta-cognition system for SOFIA
    """

    def __init__(self, model=None):
        self.model = model
        self.confidence_estimator = ConfidenceEstimator()
        self.error_detector = ErrorDetector()
        self.decision_analyzer = DecisionAnalyzer()

        # Meta-cognitive state
        self.self_awareness_level = 0.0
        self.learning_from_errors = True

    def analyze_prediction(self, text1: str, text2: str,
                          prediction: float, ground_truth: Optional[float] = None,
                          embeddings: Optional[Tuple[torch.Tensor, torch.Tensor]] = None) -> Dict[str, Any]:
        """
        Perform meta-cognitive analysis of a prediction
        """

        # Estimate confidence if embeddings available
        confidence = 0.5  # Default neutral confidence
        if embeddings:
            confidence = self.confidence_estimator(embeddings[0], embeddings[1]).item()

        # Create context
        context = {
            'text1_length': len(text1.split()),
            'text2_length': len(text2.split()),
            'text1': text1[:100],  # Truncated for storage
            'text2': text2[:100],
            'domain': self._infer_domain(text1, text2)
        }

        result = {
            'prediction': prediction,
            'confidence': confidence,
            'context': context
        }

        # Analyze error if ground truth available
        if ground_truth is not None:
            error_analysis = self.error_detector.detect_error(
                prediction, ground_truth, confidence, context
            )
            result['error_analysis'] = error_analysis

            # Update self-awareness based on error patterns
            self._update_self_awareness(error_analysis)

        return result

    def analyze_decision(self, query: str, results: List[Tuple[int, float]],
                        candidates: List[str]) -> Dict[str, Any]:
        """
        Analyze a decision-making process
        """

        context = {
            'num_candidates': len(candidates),
            'query_type': 'search' if len(results) > 1 else 'single',
            'top_candidate': candidates[results[0][0]] if results and candidates else None
        }

        analysis = self.decision_analyzer.analyze_decision(query, results, context)

        # Update self-awareness based on decision patterns
        self._update_self_awareness_from_decision(analysis)

        return analysis

    def _infer_domain(self, text1: str, text2: str) -> str:
        """Infer the domain/topic of the texts"""
        combined_text = (text1 + " " + text2).lower()

        # Simple domain detection
        if any(word in combined_text for word in ['computer', 'software', 'programming', 'code']):
            return 'technology'
        elif any(word in combined_text for word in ['health', 'medical', 'disease', 'treatment']):
            return 'health'
        elif any(word in combined_text for word in ['business', 'company', 'market', 'finance']):
            return 'business'
        elif any(word in combined_text for word in ['science', 'research', 'study', 'experiment']):
            return 'science'
        else:
            return 'general'

    def _update_self_awareness(self, error_analysis: Dict[str, Any]):
        """Update self-awareness based on error analysis"""
        if not error_analysis.get('is_error', False):
            # Correct prediction - slight increase in awareness
            self.self_awareness_level = min(1.0, self.self_awareness_level + 0.01)
        else:
            # Error - analyze and learn
            error_magnitude = error_analysis.get('error_magnitude', 0)
            confidence = error_analysis.get('confidence', 0)

            # Large errors with high confidence decrease awareness more
            awareness_penalty = error_magnitude * confidence * 0.1
            self.self_awareness_level = max(0.0, self.self_awareness_level - awareness_penalty)

            # But learning from errors can help recover
            if self.learning_from_errors:
                recovery = error_magnitude * 0.05  # Learn from mistakes
                self.self_awareness_level = min(1.0, self.self_awareness_level + recovery)

    def _update_self_awareness_from_decision(self, decision_analysis: Dict[str, Any]):
        """Update self-awareness based on decision analysis"""
        confidence = decision_analysis.get('decision_confidence', 0.5)

        # High confidence decisions increase awareness
        if confidence > 0.8:
            self.self_awareness_level = min(1.0, self.self_awareness_level + 0.005)
        elif confidence < 0.4:
            # Low confidence decisions slightly decrease awareness
            self.self_awareness_level = max(0.0, self.self_awareness_level - 0.002)

    def get_meta_cognitive_state(self) -> Dict[str, Any]:
        """Get current meta-cognitive state"""
        return {
            'self_awareness_level': self.self_awareness_level,
            'error_statistics': self.error_detector.get_error_statistics(),
            'decision_insights': self.decision_analyzer.get_decision_insights(),
            'confidence_in_abilities': self._assess_ability_confidence(),
            'learning_active': self.learning_from_errors
        }

    def _assess_ability_confidence(self) -> Dict[str, float]:
        """Assess confidence in different abilities"""
        error_stats = self.error_detector.get_error_statistics()
        decision_insights = self.decision_analyzer.get_decision_insights()

        # Base confidence on error rates and decision patterns
        error_rate = error_stats.get('error_rate', 0.5)
        avg_decision_confidence = decision_insights.get('average_confidence', 0.5)

        return {
            'similarity_prediction': 1.0 - error_rate,
            'decision_making': avg_decision_confidence,
            'error_detection': min(1.0, self.self_awareness_level + 0.3),
            'domain_adaptation': 0.7 if error_stats.get('error_patterns', {}).get('domain_general', 0) < 10 else 0.5
        }

    def reflect_on_performance(self) -> Dict[str, Any]:
        """Perform self-reflection on recent performance"""
        state = self.get_meta_cognitive_state()

        reflection = {
            'overall_assessment': self._assess_overall_performance(state),
            'strengths': self._identify_strengths(state),
            'weaknesses': self._identify_weaknesses(state),
            'improvement_suggestions': self._generate_improvement_suggestions(state),
            'confidence_level': state['self_awareness_level']
        }

        return reflection

    def _assess_overall_performance(self, state: Dict[str, Any]) -> str:
        """Assess overall performance level"""
        awareness = state['self_awareness_level']
        error_rate = state['error_statistics'].get('error_rate', 0.5)
        decision_confidence = state['decision_insights'].get('average_confidence', 0.5)

        overall_score = (awareness + (1 - error_rate) + decision_confidence) / 3

        if overall_score > 0.8:
            return "excellent"
        elif overall_score > 0.6:
            return "good"
        elif overall_score > 0.4:
            return "adequate"
        else:
            return "needs_improvement"

    def _identify_strengths(self, state: Dict[str, Any]) -> List[str]:
        """Identify current strengths"""
        strengths = []

        if state['self_awareness_level'] > 0.7:
            strengths.append("High self-awareness and error detection")

        error_rate = state['error_statistics'].get('error_rate', 0.5)
        if error_rate < 0.2:
            strengths.append("Low error rate in predictions")

        decision_conf = state['decision_insights'].get('average_confidence', 0.5)
        if decision_conf > 0.8:
            strengths.append("High confidence in decision making")

        if not strengths:
            strengths.append("Continuous learning capability")

        return strengths

    def _identify_weaknesses(self, state: Dict[str, Any]) -> List[str]:
        """Identify current weaknesses"""
        weaknesses = []

        if state['self_awareness_level'] < 0.3:
            weaknesses.append("Limited self-awareness")

        error_patterns = state['error_statistics'].get('error_patterns', {})
        if error_patterns.get('large_error', 0) > 5:
            weaknesses.append("Frequent large prediction errors")

        if error_patterns.get('low_confidence_high_error', 0) > 3:
            weaknesses.append("Overconfidence in incorrect predictions")

        decision_insights = state['decision_insights']
        if decision_insights.get('low_confidence_rate', 0) > 0.3:
            weaknesses.append("Low confidence in many decisions")

        if not weaknesses:
            weaknesses.append("Still learning and adapting")

        return weaknesses

    def _generate_improvement_suggestions(self, state: Dict[str, Any]) -> List[str]:
        """Generate suggestions for improvement"""
        suggestions = []

        error_stats = state['error_statistics']
        if error_stats.get('error_rate', 0) > 0.3:
            suggestions.append("Focus on reducing prediction errors through additional training")

        if state['self_awareness_level'] < 0.5:
            suggestions.append("Improve self-awareness by analyzing more prediction outcomes")

        decision_insights = state['decision_insights']
        if decision_insights.get('low_confidence_rate', 0) > 0.2:
            suggestions.append("Work on increasing decision confidence through better calibration")

        error_patterns = error_stats.get('error_patterns', {})
        if error_patterns.get('domain_general', 0) > error_patterns.get('domain_technology', 0):
            suggestions.append("Specialize more in technology domain where errors are lower")

        if len(suggestions) == 0:
            suggestions.append("Continue current learning approach - performance is stable")

        return suggestions


# Example usage
if __name__ == "__main__":
    print("SOFIA Meta-Cognition System")
    print("This system provides self-awareness, error detection, and decision analysis")

    # Example usage would be integrated with SOFIA model
    """
    from sofia_model import SOFIAModel
    from sofia_meta_cognition import MetaCognitiveSOFIA

    sofia = SOFIAModel()
    meta_sofia = MetaCognitiveSOFIA(sofia)

    # Analyze a prediction
    result = meta_sofia.analyze_prediction(
        "Hello world", "Hi there",
        prediction=0.85, ground_truth=0.9
    )

    # Analyze a decision
    decision_analysis = meta_sofia.analyze_decision(
        "What is AI?", [(0, 0.9), (1, 0.7)], ["AI definition", "Weather info"]
    )

    # Get self-reflection
    reflection = meta_sofia.reflect_on_performance()
    print("Self-reflection:", reflection)
    """