""" quality_weighted_aggregator.py ============================== Advanced aggregation system that combines multiple stance signals with quality weighting, confidence calibration, and uncertainty handling. This module provides sophisticated methods for combining stance classifications from multiple sources, taking into account source reliability, confidence scores, and contextual factors to produce more accurate final decisions. """ from __future__ import annotations import logging import math from typing import Dict, List, Tuple, Optional, Any from dataclasses import dataclass from enum import Enum logger = logging.getLogger("quality_weighted_aggregator") @dataclass class StanceEvidence: """Represents a single piece of stance evidence with metadata.""" stance: str # "supporting", "contradicting", "neutral" confidence: float # 0.0 to 1.0 source_reliability: float # 0.0 to 1.0 source_type: str # "official", "news", "academic", etc. method: str # "semantic", "keyword", "nli", etc. text_length: int # Length of analyzed text claim_relevance: float = 1.0 # How relevant the source is to the claim temporal_weight: float = 1.0 # Recency weight if applicable class AggregationStrategy(Enum): """Different strategies for aggregating stance evidence.""" WEIGHTED_MAJORITY = "weighted_majority" CONFIDENCE_THRESHOLD = "confidence_threshold" RELIABILITY_WEIGHTED = "reliability_weighted" BAYESIAN_UPDATE = "bayesian_update" ENSEMBLE = "ensemble" class QualityWeightedAggregator: """ Advanced aggregation system for combining multiple stance classifications with quality weighting and uncertainty handling. """ def __init__(self, strategy: AggregationStrategy = AggregationStrategy.ENSEMBLE): self.strategy = strategy # Source type weights (can be configured) self.source_type_weights = { "official": 1.2, "academic": 1.1, "news": 1.0, "fact_check": 1.3, "social": 0.7, "blog": 0.6, "unknown": 0.8, } # Method confidence multipliers self.method_weights = { "semantic": 1.1, "nli": 1.0, "keyword": 0.8, "fallback": 0.6, "default": 0.3, } # Minimum evidence thresholds self.min_evidence_count = 2 self.min_total_weight = 1.0 self.uncertainty_threshold = 0.3 def aggregate_stance_evidence( self, evidence_list: List[StanceEvidence], claim_text: str = "", context: Optional[Dict[str, Any]] = None ) -> Tuple[str, float, Dict[str, Any]]: """ Aggregate multiple stance evidence pieces into a final decision. Args: evidence_list: List of StanceEvidence objects claim_text: The original claim text for context context: Additional context information Returns: (final_stance, confidence, metadata) """ if not evidence_list: return "neutral", 0.1, {"reason": "no_evidence"} # Apply the selected aggregation strategy if self.strategy == AggregationStrategy.WEIGHTED_MAJORITY: return self._weighted_majority_aggregation(evidence_list) elif self.strategy == AggregationStrategy.CONFIDENCE_THRESHOLD: return self._confidence_threshold_aggregation(evidence_list) elif self.strategy == AggregationStrategy.RELIABILITY_WEIGHTED: return self._reliability_weighted_aggregation(evidence_list) elif self.strategy == AggregationStrategy.BAYESIAN_UPDATE: return self._bayesian_update_aggregation(evidence_list) else: # ENSEMBLE return self._ensemble_aggregation(evidence_list, claim_text, context) def _weighted_majority_aggregation( self, evidence_list: List[StanceEvidence] ) -> Tuple[str, float, Dict[str, Any]]: """Simple weighted majority voting.""" stance_weights = {"supporting": 0.0, "contradicting": 0.0, "neutral": 0.0} total_weight = 0.0 for evidence in evidence_list: weight = self._calculate_evidence_weight(evidence) stance_weights[evidence.stance] += weight total_weight += weight if total_weight == 0: return "neutral", 0.1, {"reason": "zero_weight"} # Normalize weights for stance in stance_weights: stance_weights[stance] /= total_weight best_stance = max(stance_weights, key=stance_weights.get) confidence = stance_weights[best_stance] return best_stance, confidence, { "method": "weighted_majority", "stance_weights": stance_weights, "total_weight": total_weight } def _confidence_threshold_aggregation( self, evidence_list: List[StanceEvidence] ) -> Tuple[str, float, Dict[str, Any]]: """Aggregation based on confidence thresholds.""" high_confidence_evidence = [ e for e in evidence_list if e.confidence >= 0.7 and e.source_reliability >= 0.6 ] if high_confidence_evidence: return self._weighted_majority_aggregation(high_confidence_evidence) # Fall back to all evidence if no high-confidence evidence medium_confidence_evidence = [ e for e in evidence_list if e.confidence >= 0.4 ] if medium_confidence_evidence: result = self._weighted_majority_aggregation(medium_confidence_evidence) # Reduce confidence for medium-confidence aggregation return result[0], result[1] * 0.8, {**result[2], "fallback": "medium_confidence"} return "neutral", 0.2, {"reason": "low_confidence_evidence"} def _reliability_weighted_aggregation( self, evidence_list: List[StanceEvidence] ) -> Tuple[str, float, Dict[str, Any]]: """Aggregation heavily weighted by source reliability.""" stance_scores = {"supporting": 0.0, "contradicting": 0.0, "neutral": 0.0} total_reliability = 0.0 for evidence in evidence_list: # Weight primarily by reliability, with confidence as secondary factor weight = evidence.source_reliability * (0.5 + 0.5 * evidence.confidence) weight *= self.source_type_weights.get(evidence.source_type, 1.0) stance_scores[evidence.stance] += weight total_reliability += evidence.source_reliability if total_reliability == 0: return "neutral", 0.1, {"reason": "no_reliable_sources"} best_stance = max(stance_scores, key=stance_scores.get) # Confidence based on reliability and consensus max_score = stance_scores[best_stance] second_max = max(score for stance, score in stance_scores.items() if stance != best_stance) margin = max_score - second_max if second_max > 0 else max_score confidence = min(0.95, (max_score / sum(stance_scores.values())) + margin * 0.1) return best_stance, confidence, { "method": "reliability_weighted", "stance_scores": stance_scores, "avg_reliability": total_reliability / len(evidence_list) } def _bayesian_update_aggregation( self, evidence_list: List[StanceEvidence] ) -> Tuple[str, float, Dict[str, Any]]: """Bayesian updating approach to stance aggregation.""" # Prior probabilities (neutral prior) priors = {"supporting": 0.33, "contradicting": 0.33, "neutral": 0.34} posteriors = priors.copy() for evidence in evidence_list: # Calculate likelihood based on evidence quality reliability_factor = evidence.source_reliability confidence_factor = evidence.confidence method_factor = self.method_weights.get(evidence.method, 1.0) # Likelihood of observing this evidence given each stance if evidence.stance == "supporting": likelihood = { "supporting": 0.8 * reliability_factor * confidence_factor * method_factor, "contradicting": 0.1, "neutral": 0.3 } elif evidence.stance == "contradicting": likelihood = { "supporting": 0.1, "contradicting": 0.8 * reliability_factor * confidence_factor * method_factor, "neutral": 0.3 } else: # neutral likelihood = { "supporting": 0.2, "contradicting": 0.2, "neutral": 0.6 * reliability_factor * confidence_factor * method_factor } # Bayesian update evidence_prob = sum(posteriors[stance] * likelihood[stance] for stance in posteriors) if evidence_prob > 0: for stance in posteriors: posteriors[stance] = (posteriors[stance] * likelihood[stance]) / evidence_prob best_stance = max(posteriors, key=posteriors.get) confidence = posteriors[best_stance] return best_stance, confidence, { "method": "bayesian_update", "posteriors": posteriors, "evidence_count": len(evidence_list) } def _ensemble_aggregation( self, evidence_list: List[StanceEvidence], claim_text: str = "", context: Optional[Dict[str, Any]] = None ) -> Tuple[str, float, Dict[str, Any]]: """ Ensemble method combining multiple aggregation strategies. """ # Get results from different strategies strategies = [ AggregationStrategy.WEIGHTED_MAJORITY, AggregationStrategy.RELIABILITY_WEIGHTED, AggregationStrategy.BAYESIAN_UPDATE ] results = [] for strategy in strategies: old_strategy = self.strategy self.strategy = strategy try: result = self.aggregate_stance_evidence(evidence_list, claim_text, context) results.append(result) except Exception as e: logger.warning("Strategy %s failed: %s", strategy, e) finally: self.strategy = old_strategy if not results: return "neutral", 0.1, {"reason": "all_strategies_failed"} # Combine results with weighted voting stance_votes = {"supporting": 0.0, "contradicting": 0.0, "neutral": 0.0} total_confidence = 0.0 strategy_weights = [1.0, 1.2, 0.8] # Weight reliability-weighted higher for i, (stance, confidence, metadata) in enumerate(results): weight = strategy_weights[i] * confidence stance_votes[stance] += weight total_confidence += confidence if total_confidence == 0: return "neutral", 0.1, {"reason": "zero_ensemble_confidence"} # Final decision best_stance = max(stance_votes, key=stance_votes.get) # Ensemble confidence calculation max_vote = stance_votes[best_stance] total_votes = sum(stance_votes.values()) consensus_strength = max_vote / total_votes if total_votes > 0 else 0 avg_confidence = total_confidence / len(results) final_confidence = min(0.95, consensus_strength * avg_confidence + self._calculate_ensemble_bonus(evidence_list)) return best_stance, final_confidence, { "method": "ensemble", "strategy_results": results, "stance_votes": stance_votes, "consensus_strength": consensus_strength, "avg_confidence": avg_confidence } def _calculate_evidence_weight(self, evidence: StanceEvidence) -> float: """Calculate the weight of a piece of evidence.""" base_weight = evidence.confidence * evidence.source_reliability # Apply source type multiplier type_multiplier = self.source_type_weights.get(evidence.source_type, 1.0) # Apply method multiplier method_multiplier = self.method_weights.get(evidence.method, 1.0) # Apply relevance and temporal weights relevance_weight = evidence.claim_relevance temporal_weight = evidence.temporal_weight # Text length bonus (longer analysis generally more reliable) length_bonus = min(1.2, 1.0 + (evidence.text_length / 1000) * 0.1) final_weight = (base_weight * type_multiplier * method_multiplier * relevance_weight * temporal_weight * length_bonus) return max(0.01, final_weight) # Minimum weight to avoid zero def _calculate_ensemble_bonus(self, evidence_list: List[StanceEvidence]) -> float: """Calculate bonus confidence for ensemble methods.""" # Bonus for having multiple high-quality sources high_quality_count = sum( 1 for e in evidence_list if e.source_reliability >= 0.7 and e.confidence >= 0.6 ) # Bonus for method diversity unique_methods = len(set(e.method for e in evidence_list)) # Bonus for source type diversity unique_source_types = len(set(e.source_type for e in evidence_list)) quality_bonus = min(0.15, high_quality_count * 0.03) method_bonus = min(0.1, unique_methods * 0.02) type_bonus = min(0.1, unique_source_types * 0.02) return quality_bonus + method_bonus + type_bonus def assess_uncertainty(self, evidence_list: List[StanceEvidence]) -> Dict[str, Any]: """ Assess the uncertainty in the stance classification. Returns: Dictionary with uncertainty metrics """ if not evidence_list: return {"high_uncertainty": True, "reason": "no_evidence"} # Calculate stance distribution stance_weights = {"supporting": 0.0, "contradicting": 0.0, "neutral": 0.0} total_weight = 0.0 for evidence in evidence_list: weight = self._calculate_evidence_weight(evidence) stance_weights[evidence.stance] += weight total_weight += weight if total_weight == 0: return {"high_uncertainty": True, "reason": "zero_weight"} # Normalize for stance in stance_weights: stance_weights[stance] /= total_weight # Calculate entropy (uncertainty measure) entropy = -sum(p * math.log2(p) for p in stance_weights.values() if p > 0) max_entropy = math.log2(3) # Maximum entropy for 3 classes normalized_entropy = entropy / max_entropy # Calculate margin between top two stances sorted_weights = sorted(stance_weights.values(), reverse=True) margin = sorted_weights[0] - sorted_weights[1] if len(sorted_weights) > 1 else sorted_weights[0] # Assess evidence quality avg_reliability = sum(e.source_reliability for e in evidence_list) / len(evidence_list) avg_confidence = sum(e.confidence for e in evidence_list) / len(evidence_list) # Determine if uncertainty is high high_uncertainty = ( normalized_entropy > 0.8 or # High entropy margin < 0.2 or # Small margin avg_reliability < 0.4 or # Low reliability avg_confidence < 0.5 or # Low confidence len(evidence_list) < self.min_evidence_count # Insufficient evidence ) return { "high_uncertainty": high_uncertainty, "entropy": entropy, "normalized_entropy": normalized_entropy, "margin": margin, "avg_reliability": avg_reliability, "avg_confidence": avg_confidence, "evidence_count": len(evidence_list), "stance_distribution": stance_weights } def create_stance_evidence_from_classification( stance: str, confidence: float, source_reliability: float, source_type: str, method: str, text_length: int, claim_relevance: float = 1.0, temporal_weight: float = 1.0 ) -> StanceEvidence: """Helper function to create StanceEvidence from classification results.""" return StanceEvidence( stance=stance, confidence=confidence, source_reliability=source_reliability, source_type=source_type, method=method, text_length=text_length, claim_relevance=claim_relevance, temporal_weight=temporal_weight )