Spaces:

infoclassifier
/

InfoClassifier

Sleeping

InfoClassifier / quality_weighted_aggregator.py

UTKARSH MISHRA

Initial commit: InfoClassifier project

e5e36db 7 months ago

17.2 kB

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

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