Spaces:
Sleeping
Sleeping
Download enhanced_stance_classifier.py from infoclassifier/InfoClassifier: direct link, hf CLI and curl.
- Browser
- Download file 34.5 kB
-
https://huggingface.co/spaces/infoclassifier/InfoClassifier/resolve/main/enhanced_stance_classifier.py
- Command line
-
hf download hf://spaces/infoclassifier/InfoClassifier/enhanced_stance_classifier.py
-
curl -L -o enhanced_stance_classifier.py https://huggingface.co/spaces/infoclassifier/InfoClassifier/resolve/main/enhanced_stance_classifier.py
34.5 kB
| """ | |
| enhanced_stance_classifier.py | |
| ============================= | |
| Advanced stance classification system with improved context understanding, | |
| multi-template NLI, and confidence-based decision making. | |
| Key improvements over the original stance_classifier.py: | |
| 1. Enhanced text preprocessing with intelligent chunking | |
| 2. Multiple NLI templates for different claim types | |
| 3. Better context extraction and relevance scoring | |
| 4. Confidence-based aggregation with fallback strategies | |
| 5. Domain-specific semantic rules for common misinformation topics | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import logging | |
| import os | |
| import re | |
| import time | |
| from typing import Dict, List, Optional, Tuple, Any | |
| from dataclasses import dataclass | |
| from enum import Enum | |
| logger = logging.getLogger("enhanced_stance_classifier") | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Enhanced Text Processing | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class TextProcessor: | |
| """Advanced text preprocessing with intelligent chunking and cleaning.""" | |
| def __init__(self): | |
| self.sentence_endings = re.compile(r'[.!?]+\s+') | |
| self.noise_patterns = [ | |
| re.compile(r'\[.*?\]'), # Remove bracketed content | |
| re.compile(r'\(.*?\)'), # Remove parenthetical content (optional) | |
| re.compile(r'https?://\S+'), # Remove URLs | |
| re.compile(r'@\w+'), # Remove mentions | |
| re.compile(r'#\w+'), # Remove hashtags | |
| re.compile(r'\s+'), # Normalize whitespace | |
| ] | |
| def clean_text(self, text: str) -> str: | |
| """Clean and normalize text for better processing.""" | |
| if not text: | |
| return "" | |
| # Remove noise patterns | |
| cleaned = text | |
| for pattern in self.noise_patterns[:-1]: # Don't apply whitespace normalization yet | |
| cleaned = pattern.sub(' ', cleaned) | |
| # Normalize whitespace (last pattern) | |
| cleaned = self.noise_patterns[-1].sub(' ', cleaned).strip() | |
| return cleaned | |
| def intelligent_chunk(self, text: str, max_length: int = 512, overlap: int = 50) -> List[str]: | |
| """ | |
| Split text into chunks at sentence boundaries while preserving context. | |
| Args: | |
| text: Input text to chunk | |
| max_length: Maximum characters per chunk | |
| overlap: Characters to overlap between chunks for context | |
| Returns: | |
| List of text chunks with preserved sentence boundaries | |
| """ | |
| if len(text) <= max_length: | |
| return [text] | |
| sentences = self.sentence_endings.split(text) | |
| chunks = [] | |
| current_chunk = "" | |
| for sentence in sentences: | |
| sentence = sentence.strip() | |
| if not sentence: | |
| continue | |
| # If adding this sentence would exceed max_length, start a new chunk | |
| if len(current_chunk) + len(sentence) > max_length and current_chunk: | |
| chunks.append(current_chunk.strip()) | |
| # Start new chunk with overlap from previous chunk | |
| if overlap > 0 and len(current_chunk) > overlap: | |
| current_chunk = current_chunk[-overlap:] + " " + sentence | |
| else: | |
| current_chunk = sentence | |
| else: | |
| current_chunk += " " + sentence if current_chunk else sentence | |
| # Add the last chunk | |
| if current_chunk.strip(): | |
| chunks.append(current_chunk.strip()) | |
| return chunks if chunks else [text[:max_length]] | |
| def extract_key_entities(self, text: str) -> List[str]: | |
| """Extract key entities and phrases from text.""" | |
| # Simple entity extraction - can be enhanced with NER models | |
| entities = [] | |
| # Extract capitalized phrases (potential proper nouns) | |
| capitalized = re.findall(r'\b[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*\b', text) | |
| entities.extend(capitalized) | |
| # Extract quoted phrases | |
| quoted = re.findall(r'"([^"]+)"', text) | |
| entities.extend(quoted) | |
| # Extract numbers with units | |
| numbers = re.findall(r'\b\d+(?:\.\d+)?\s*(?:%|percent|million|billion|thousand)\b', text, re.I) | |
| entities.extend(numbers) | |
| return list(set(entities)) | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Enhanced Semantic Rules | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class ClaimType(Enum): | |
| HEALTH = "health" | |
| POLITICS = "politics" | |
| SCIENCE = "science" | |
| TECHNOLOGY = "technology" | |
| CLIMATE = "climate" | |
| ECONOMICS = "economics" | |
| GENERAL = "general" | |
| class SemanticRule: | |
| claim_pattern: str | |
| source_pattern: str | |
| stance: str # "supporting" or "contradicting" | |
| confidence: float | |
| claim_types: List[ClaimType] | |
| class EnhancedSemanticLayer: | |
| """ | |
| Advanced semantic heuristics with domain-specific rules and better pattern matching. | |
| """ | |
| def __init__(self): | |
| self.rules = self._build_enhanced_rules() | |
| self.text_processor = TextProcessor() | |
| def _build_enhanced_rules(self) -> List[SemanticRule]: | |
| """Build comprehensive semantic rules for different domains.""" | |
| rules = [] | |
| # Health/Medical misinformation rules | |
| health_rules = [ | |
| SemanticRule( | |
| claim_pattern=r"\bvaccine.{0,30}(cause|causes|causing).{0,30}(autism|death|cancer|infertility|dna)", | |
| source_pattern=r"\b(no\s+evidence|debunked|safe|clinical\s+trial|fda\s+approved|peer\s+reviewed)\b", | |
| stance="contradicting", | |
| confidence=0.9, | |
| claim_types=[ClaimType.HEALTH] | |
| ), | |
| SemanticRule( | |
| claim_pattern=r"\bcovid.{0,20}(hoax|fake|planned|bioweapon)", | |
| source_pattern=r"\b(who|cdc|health\s+officials|scientific\s+consensus|peer\s+reviewed)\b", | |
| stance="contradicting", | |
| confidence=0.85, | |
| claim_types=[ClaimType.HEALTH] | |
| ), | |
| SemanticRule( | |
| claim_pattern=r"\b(cure|treatment|remedy).{0,30}(cancer|covid|aids|diabetes)", | |
| source_pattern=r"\b(no\s+evidence|unproven|not\s+approved|clinical\s+trial|fda)\b", | |
| stance="contradicting", | |
| confidence=0.8, | |
| claim_types=[ClaimType.HEALTH] | |
| ), | |
| ] | |
| # Technology misinformation rules | |
| tech_rules = [ | |
| SemanticRule( | |
| claim_pattern=r"\b5g.{0,30}(cause|spread|transmit).{0,30}(covid|cancer|virus|radiation)", | |
| source_pattern=r"\b(no\s+link|no\s+evidence|debunked|scientifically\s+impossible|who\s+says)\b", | |
| stance="contradicting", | |
| confidence=0.9, | |
| claim_types=[ClaimType.TECHNOLOGY] | |
| ), | |
| SemanticRule( | |
| claim_pattern=r"\bmicrochip.{0,30}(vaccine|implant|track|control)", | |
| source_pattern=r"\b(conspiracy|debunked|no\s+evidence|fact\s+check|impossible)\b", | |
| stance="contradicting", | |
| confidence=0.85, | |
| claim_types=[ClaimType.TECHNOLOGY] | |
| ), | |
| ] | |
| # Climate misinformation rules | |
| climate_rules = [ | |
| SemanticRule( | |
| claim_pattern=r"\bclimate\s+change.{0,20}(hoax|fake|not\s+real|natural)", | |
| source_pattern=r"\b(scientific\s+consensus|97\s*%|ipcc|nasa|noaa|peer\s+reviewed)\b", | |
| stance="contradicting", | |
| confidence=0.9, | |
| claim_types=[ClaimType.CLIMATE] | |
| ), | |
| SemanticRule( | |
| claim_pattern=r"\bglobal\s+warming.{0,20}(stopped|pause|cooling)", | |
| source_pattern=r"\b(temperature\s+records|warmest\s+year|data\s+shows|scientists\s+confirm)\b", | |
| stance="contradicting", | |
| confidence=0.8, | |
| claim_types=[ClaimType.CLIMATE] | |
| ), | |
| ] | |
| # Political misinformation rules | |
| political_rules = [ | |
| SemanticRule( | |
| claim_pattern=r"\belection.{0,30}(stolen|rigged|fraud|fake)", | |
| source_pattern=r"\b(courts?\s+reject|no\s+evidence|audit\s+confirms|official\s+results)\b", | |
| stance="contradicting", | |
| confidence=0.85, | |
| claim_types=[ClaimType.POLITICS] | |
| ), | |
| SemanticRule( | |
| claim_pattern=r"\bgovernment.{0,30}(announce|confirm|official)", | |
| source_pattern=r"\b(official\s+statement|government\s+website|press\s+release|ministry)\b", | |
| stance="supporting", | |
| confidence=0.8, | |
| claim_types=[ClaimType.POLITICS] | |
| ), | |
| ] | |
| rules.extend(health_rules + tech_rules + climate_rules + political_rules) | |
| return rules | |
| def classify_claim_type(self, claim: str) -> ClaimType: | |
| """Classify the type of claim to apply appropriate rules.""" | |
| claim_lower = claim.lower() | |
| health_keywords = ["vaccine", "covid", "virus", "cancer", "health", "medical", "drug", "treatment"] | |
| tech_keywords = ["5g", "microchip", "ai", "technology", "internet", "phone", "radiation"] | |
| climate_keywords = ["climate", "warming", "carbon", "emission", "fossil", "renewable", "temperature"] | |
| political_keywords = ["election", "government", "president", "vote", "congress", "parliament", "policy"] | |
| science_keywords = ["study", "research", "scientist", "experiment", "evidence", "data", "peer review"] | |
| economics_keywords = ["economy", "inflation", "gdp", "unemployment", "market", "recession", "financial"] | |
| keyword_sets = [ | |
| (health_keywords, ClaimType.HEALTH), | |
| (tech_keywords, ClaimType.TECHNOLOGY), | |
| (climate_keywords, ClaimType.CLIMATE), | |
| (political_keywords, ClaimType.POLITICS), | |
| (science_keywords, ClaimType.SCIENCE), | |
| (economics_keywords, ClaimType.ECONOMICS), | |
| ] | |
| max_matches = 0 | |
| best_type = ClaimType.GENERAL | |
| for keywords, claim_type in keyword_sets: | |
| matches = sum(1 for kw in keywords if kw in claim_lower) | |
| if matches > max_matches: | |
| max_matches = matches | |
| best_type = claim_type | |
| return best_type | |
| def classify(self, claim: str, source_text: str) -> Optional[Tuple[str, float]]: | |
| """ | |
| Enhanced semantic classification with domain-specific rules. | |
| Returns: | |
| (stance, confidence) or None if no rule matches | |
| """ | |
| claim_type = self.classify_claim_type(claim) | |
| claim_lower = claim.lower() | |
| source_lower = source_text.lower() | |
| # Apply rules relevant to the claim type | |
| for rule in self.rules: | |
| if claim_type in rule.claim_types or ClaimType.GENERAL in rule.claim_types: | |
| if re.search(rule.claim_pattern, claim_lower) and re.search(rule.source_pattern, source_lower): | |
| logger.debug( | |
| "Enhanced semantic rule matched: %s -> %s (conf=%.2f)", | |
| rule.claim_pattern[:40], rule.stance, rule.confidence | |
| ) | |
| return rule.stance, rule.confidence | |
| return None | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Multi-Template NLI Layer | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class NLITemplateManager: | |
| """Manages multiple NLI templates for different claim types and contexts.""" | |
| def __init__(self): | |
| self.templates = self._build_templates() | |
| def _build_templates(self) -> Dict[ClaimType, List[str]]: | |
| """Build claim-type specific NLI templates.""" | |
| return { | |
| ClaimType.HEALTH: [ | |
| "This text confirms that {}", | |
| "This text disproves that {}", | |
| "According to this source, {} is true", | |
| "According to this source, {} is false", | |
| "Medical evidence shows that {}", | |
| "Medical evidence contradicts that {}", | |
| ], | |
| ClaimType.SCIENCE: [ | |
| "Scientific research confirms that {}", | |
| "Scientific research contradicts that {}", | |
| "Studies show that {}", | |
| "Studies disprove that {}", | |
| "Evidence supports that {}", | |
| "Evidence refutes that {}", | |
| ], | |
| ClaimType.POLITICS: [ | |
| "Official sources confirm that {}", | |
| "Official sources deny that {}", | |
| "Government records show that {}", | |
| "Government records contradict that {}", | |
| "Political analysis supports that {}", | |
| "Political analysis disputes that {}", | |
| ], | |
| ClaimType.TECHNOLOGY: [ | |
| "Technical analysis confirms that {}", | |
| "Technical analysis disproves that {}", | |
| "Technology experts agree that {}", | |
| "Technology experts dispute that {}", | |
| "Research demonstrates that {}", | |
| "Research contradicts that {}", | |
| ], | |
| ClaimType.CLIMATE: [ | |
| "Climate data confirms that {}", | |
| "Climate data contradicts that {}", | |
| "Environmental research shows that {}", | |
| "Environmental research disproves that {}", | |
| "Scientific consensus supports that {}", | |
| "Scientific consensus rejects that {}", | |
| ], | |
| ClaimType.GENERAL: [ | |
| "This text supports the claim that {}", | |
| "This text contradicts the claim that {}", | |
| "Evidence shows that {}", | |
| "Evidence disproves that {}", | |
| "This information confirms that {}", | |
| "This information refutes that {}", | |
| ] | |
| } | |
| def get_templates(self, claim_type: ClaimType) -> List[str]: | |
| """Get appropriate templates for a claim type.""" | |
| return self.templates.get(claim_type, self.templates[ClaimType.GENERAL]) | |
| class EnhancedNLILayer: | |
| """ | |
| Enhanced NLI layer with multiple templates, better preprocessing, and confidence calibration. | |
| """ | |
| def __init__(self): | |
| self.template_manager = NLITemplateManager() | |
| self.text_processor = TextProcessor() | |
| self._cache: Dict[str, Tuple[str, float]] = {} | |
| self._enabled = os.getenv("STANCE_NLI_ENABLED", "true").lower() != "false" | |
| def available(self) -> bool: | |
| return self._enabled | |
| def classify(self, claim: str, title: str, text: str, claim_type: ClaimType = ClaimType.GENERAL) -> Optional[Tuple[str, float]]: | |
| """ | |
| Enhanced NLI classification with multiple templates and better preprocessing. | |
| Returns: | |
| (stance, confidence) or None if classification fails | |
| """ | |
| if not self._enabled: | |
| return None | |
| # Create cache key β use more text to discriminate full articles | |
| cache_key = hashlib.md5( | |
| f"{claim[:120]}||{title[:80]}||{text[:600]}||{claim_type.value}".encode() | |
| ).hexdigest() | |
| if cache_key in self._cache: | |
| return self._cache[cache_key] | |
| # Clean and chunk the text β use larger chunks to handle full articles | |
| combined_text = self.text_processor.clean_text(f"{title} {text}") | |
| text_chunks = self.text_processor.intelligent_chunk(combined_text, max_length=1000) | |
| # Get appropriate templates for this claim type | |
| templates = self.template_manager.get_templates(claim_type) | |
| # Try multiple approaches and aggregate results | |
| results = [] | |
| for chunk in text_chunks[:5]: # Process up to 5 chunks to cover full articles | |
| for template_pair in [templates[i:i+2] for i in range(0, len(templates), 2)]: | |
| if len(template_pair) == 2: | |
| support_template, contradict_template = template_pair | |
| result = self._classify_with_templates(claim, chunk, support_template, contradict_template) | |
| if result: | |
| results.append(result) | |
| if not results: | |
| return None | |
| # Aggregate results with confidence weighting | |
| final_stance, final_confidence = self._aggregate_results(results) | |
| # Cache the result | |
| self._cache[cache_key] = (final_stance, final_confidence) | |
| logger.debug( | |
| "Enhanced NLI: '%s...' -> %s (conf=%.2f) from %d results", | |
| claim[:40], final_stance, final_confidence, len(results) | |
| ) | |
| return final_stance, final_confidence | |
| def _classify_with_templates(self, claim: str, text: str, support_template: str, contradict_template: str) -> Optional[Tuple[str, float]]: | |
| """Classify using a pair of support/contradict templates.""" | |
| try: | |
| # Try HF API first, then local transformers | |
| result = self._hf_api_classify(claim, text, support_template, contradict_template) | |
| if result: | |
| return result | |
| result = self._local_classify(claim, text, support_template, contradict_template) | |
| return result | |
| except Exception as exc: | |
| logger.debug("Template classification failed: %s", exc) | |
| return None | |
| def _hf_api_classify(self, claim: str, text: str, support_template: str, contradict_template: str) -> Optional[Tuple[str, float]]: | |
| """Use HuggingFace API for classification.""" | |
| try: | |
| import requests | |
| except ImportError: | |
| return None | |
| hf_token = os.getenv("HF_TOKEN", "") | |
| if not hf_token or hf_token.startswith("your_"): | |
| return None | |
| headers = { | |
| "Content-Type": "application/json", | |
| "Authorization": f"Bearer {hf_token}", | |
| } | |
| # Use zero-shot classification with custom templates | |
| payload = { | |
| "inputs": text, | |
| "parameters": { | |
| "candidate_labels": [ | |
| support_template.format(claim), | |
| contradict_template.format(claim), | |
| "This text is unrelated to the claim" | |
| ], | |
| "multi_label": False, | |
| }, | |
| } | |
| try: | |
| resp = requests.post( | |
| "https://api-inference.huggingface.co/models/facebook/bart-large-mnli", | |
| headers=headers, | |
| json=payload, | |
| timeout=15 | |
| ) | |
| if resp.status_code in (503, 502): | |
| return None | |
| if not resp.ok: | |
| return None | |
| data = resp.json() | |
| return self._parse_hf_response(data) | |
| except Exception: | |
| return None | |
| def _local_classify(self, claim: str, text: str, support_template: str, contradict_template: str) -> Optional[Tuple[str, float]]: | |
| """Use local transformers for classification.""" | |
| try: | |
| from transformers import pipeline | |
| # Initialize pipeline if not already done | |
| if not hasattr(self, '_local_pipeline'): | |
| self._local_pipeline = pipeline( | |
| "zero-shot-classification", | |
| model="facebook/bart-large-mnli", | |
| device=-1 # CPU | |
| ) | |
| result = self._local_pipeline( | |
| text, | |
| candidate_labels=[ | |
| support_template.format(claim), | |
| contradict_template.format(claim), | |
| "unrelated" | |
| ] | |
| ) | |
| labels = result["labels"] | |
| scores = result["scores"] | |
| if "unrelated" in labels[0].lower(): | |
| return "neutral", float(scores[0]) | |
| elif support_template.format("").strip() in labels[0]: | |
| return "supporting", float(scores[0]) | |
| else: | |
| return "contradicting", float(scores[0]) | |
| except Exception: | |
| return None | |
| def _parse_hf_response(self, data: dict) -> Optional[Tuple[str, float]]: | |
| """Parse HuggingFace API response.""" | |
| if not data or not isinstance(data, dict): | |
| return None | |
| labels = data.get("labels", []) | |
| scores = data.get("scores", []) | |
| if not labels or not scores: | |
| return None | |
| top_label = labels[0].lower() | |
| top_score = float(scores[0]) | |
| if "unrelated" in top_label: | |
| return "neutral", top_score | |
| elif any(word in top_label for word in ["confirms", "supports", "true", "shows"]): | |
| return "supporting", top_score | |
| else: | |
| return "contradicting", top_score | |
| def _aggregate_results(self, results: List[Tuple[str, float]]) -> Tuple[str, float]: | |
| """Aggregate multiple NLI results with confidence weighting.""" | |
| if not results: | |
| return "neutral", 0.2 | |
| # Weight results by confidence | |
| stance_scores = {"supporting": 0.0, "contradicting": 0.0, "neutral": 0.0} | |
| total_weight = 0.0 | |
| for stance, confidence in results: | |
| stance_scores[stance] += confidence | |
| total_weight += confidence | |
| if total_weight == 0: | |
| return "neutral", 0.2 | |
| # Normalize scores | |
| for stance in stance_scores: | |
| stance_scores[stance] /= total_weight | |
| # Find the highest scoring stance | |
| best_stance = max(stance_scores, key=stance_scores.get) | |
| best_score = stance_scores[best_stance] | |
| # Adjust confidence based on margin and number of results | |
| margin = best_score - max(score for stance, score in stance_scores.items() if stance != best_stance) | |
| confidence = min(0.95, best_score + margin * 0.1 + len(results) * 0.05) | |
| return best_stance, confidence | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Enhanced Keyword Layer | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class EnhancedKeywordLayer: | |
| """Enhanced keyword matching with context awareness and better thresholds.""" | |
| def __init__(self): | |
| self.strong_contra = { | |
| "false", "debunked", "disproven", "misinformation", "misleading", | |
| "no evidence", "lacks evidence", "unfounded", "refuted", "myth", | |
| "conspiracy theory", "fact-check", "factcheck", "incorrect", | |
| "not true", "inaccurate", "unproven", "pseudoscience", "hoax", | |
| "wrong", "fake", "fabricated", "baseless", "discredited", | |
| "impossible", "scientifically impossible", "medically impossible" | |
| } | |
| self.weak_contra = { | |
| "disputed", "controversial", "unverified", "claims", "alleged", | |
| "questioned", "skeptics", "critics", "challenged", "contradicts", | |
| "refutes", "rejects", "warns against", "cautioned", "doubtful", | |
| "uncertain", "unclear", "mixed evidence", "conflicting" | |
| } | |
| self.strong_support = { | |
| "confirms", "confirmed", "verified", "validated", "proven", | |
| "evidence shows", "data shows", "research shows", "study finds", | |
| "scientists confirm", "officially announced", "official statement", | |
| "press release", "gazette notification", "government confirms", | |
| "approved", "endorsed", "supported by evidence", "peer reviewed", | |
| "clinical trial", "fda approved", "who approved", "scientifically proven" | |
| } | |
| self.weak_support = { | |
| "suggests", "indicates", "shows", "demonstrates", "according to", | |
| "reports", "states", "announced", "published", "found that", | |
| "appears", "seems", "likely", "probably", "research indicates", | |
| "studies suggest", "data indicates", "preliminary results" | |
| } | |
| self.context_modifiers = { | |
| "negation": ["not", "no", "never", "none", "neither", "without"], | |
| "uncertainty": ["may", "might", "could", "possibly", "perhaps", "maybe"], | |
| "emphasis": ["definitely", "certainly", "absolutely", "clearly", "obviously"] | |
| } | |
| def classify(self, text: str, claim: str = "") -> Tuple[str, float]: | |
| """ | |
| Enhanced keyword classification with context awareness. | |
| Returns: | |
| (stance, confidence) | |
| """ | |
| text_lower = text.lower() | |
| # Count keyword matches | |
| strong_contra_count = sum(1 for kw in self.strong_contra if kw in text_lower) | |
| weak_contra_count = sum(1 for kw in self.weak_contra if kw in text_lower) | |
| strong_support_count = sum(1 for kw in self.strong_support if kw in text_lower) | |
| weak_support_count = sum(1 for kw in self.weak_support if kw in text_lower) | |
| # Check for context modifiers | |
| negation_count = sum(1 for kw in self.context_modifiers["negation"] if kw in text_lower) | |
| uncertainty_count = sum(1 for kw in self.context_modifiers["uncertainty"] if kw in text_lower) | |
| emphasis_count = sum(1 for kw in self.context_modifiers["emphasis"] if kw in text_lower) | |
| # Calculate weighted scores | |
| contra_score = strong_contra_count * 3 + weak_contra_count * 1 | |
| support_score = strong_support_count * 3 + weak_support_count * 1 | |
| # Apply context modifiers | |
| if negation_count > 0: | |
| # Negation can flip the meaning | |
| if contra_score > support_score: | |
| contra_score *= 0.5 # Reduce contradiction confidence | |
| elif support_score > contra_score: | |
| support_score *= 0.5 # Reduce support confidence | |
| # Uncertainty reduces confidence | |
| uncertainty_penalty = min(0.3, uncertainty_count * 0.1) | |
| # Emphasis increases confidence | |
| emphasis_bonus = min(0.2, emphasis_count * 0.05) | |
| # Determine stance and confidence | |
| if strong_contra_count >= 1 and contra_score > support_score: | |
| base_confidence = min(0.9, 0.6 + strong_contra_count * 0.1) | |
| confidence = base_confidence - uncertainty_penalty + emphasis_bonus | |
| return "contradicting", max(0.3, confidence) | |
| if strong_support_count >= 1 and support_score > contra_score: | |
| base_confidence = min(0.9, 0.6 + strong_support_count * 0.1) | |
| confidence = base_confidence - uncertainty_penalty + emphasis_bonus | |
| return "supporting", max(0.3, confidence) | |
| if contra_score >= 2 and contra_score > support_score: | |
| confidence = 0.5 - uncertainty_penalty + emphasis_bonus | |
| return "contradicting", max(0.2, confidence) | |
| if support_score >= 2 and support_score > contra_score: | |
| confidence = 0.5 - uncertainty_penalty + emphasis_bonus | |
| return "supporting", max(0.2, confidence) | |
| # Default to neutral with low confidence | |
| return "neutral", 0.2 | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Main Enhanced Stance Classifier | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| class EnhancedStanceClassifier: | |
| """ | |
| Advanced stance classifier combining enhanced semantic rules, improved NLI, | |
| and better keyword matching with confidence-based decision making. | |
| """ | |
| def __init__(self): | |
| self.semantic_layer = EnhancedSemanticLayer() | |
| self.keyword_layer = EnhancedKeywordLayer() | |
| self.nli_layer = EnhancedNLILayer() | |
| self.text_processor = TextProcessor() | |
| # Confidence thresholds for each layer | |
| self.semantic_threshold = 0.7 | |
| self.keyword_threshold = 0.6 | |
| self.nli_threshold = 0.5 | |
| # Minimum source reliability for NLI layer | |
| self.nli_min_reliability = float(os.getenv("STANCE_NLI_MIN_REL", "0.3")) | |
| def classify(self, text: str, claim: str) -> str: | |
| """ | |
| Backward-compatible interface for simple stance classification. | |
| Returns: | |
| stance string: "supporting" | "contradicting" | "neutral" | |
| """ | |
| stance, _, _ = self.classify_detailed( | |
| title="", text=text, claim=claim, source_reliability=0.5 | |
| ) | |
| return stance | |
| def classify_detailed( | |
| self, | |
| title: str, | |
| text: str, | |
| claim: str, | |
| source_reliability: float = 0.5, | |
| ) -> Tuple[str, float, str]: | |
| """ | |
| Enhanced stance classification with detailed output. | |
| Args: | |
| title: Source title | |
| text: Source text content | |
| claim: Claim to verify against | |
| source_reliability: Reliability score of the source (0.0-1.0) | |
| Returns: | |
| (stance, confidence, winning_layer) | |
| """ | |
| # Clean and preprocess text | |
| combined_text = self.text_processor.clean_text(f"{title} {text}") | |
| claim_type = self.semantic_layer.classify_claim_type(claim) | |
| # Layer 1: Enhanced Semantic Heuristics | |
| semantic_result = self.semantic_layer.classify(claim, combined_text) | |
| if semantic_result and semantic_result[1] >= self.semantic_threshold: | |
| stance, confidence = semantic_result | |
| logger.debug("Enhanced semantic layer: %s (%.2f)", stance, confidence) | |
| return stance, confidence, "semantic" | |
| # Layer 2: Enhanced Keyword Matching | |
| keyword_stance, keyword_conf = self.keyword_layer.classify(combined_text, claim) | |
| if keyword_conf >= self.keyword_threshold: | |
| logger.debug("Enhanced keyword layer: %s (%.2f)", keyword_stance, keyword_conf) | |
| return keyword_stance, keyword_conf, "keyword" | |
| # Layer 3: Enhanced NLI (only for reliable sources) | |
| if (self.nli_layer.available() and | |
| source_reliability >= self.nli_min_reliability): | |
| nli_result = self.nli_layer.classify(claim, title, text, claim_type) | |
| if nli_result and nli_result[1] >= self.nli_threshold: | |
| stance, confidence = nli_result | |
| logger.debug("Enhanced NLI layer: %s (%.2f)", stance, confidence) | |
| return stance, confidence, "nli" | |
| # Fallback: Use best available result with adjusted confidence | |
| best_result = None | |
| best_confidence = 0.0 | |
| best_layer = "fallback" | |
| if semantic_result and semantic_result[1] > best_confidence: | |
| best_result = semantic_result | |
| best_confidence = semantic_result[1] | |
| best_layer = "semantic_fallback" | |
| if keyword_conf > best_confidence: | |
| best_result = (keyword_stance, keyword_conf) | |
| best_confidence = keyword_conf | |
| best_layer = "keyword_fallback" | |
| if best_result: | |
| stance, confidence = best_result | |
| # Reduce confidence for fallback results | |
| adjusted_confidence = confidence * 0.7 | |
| logger.debug("Fallback layer (%s): %s (%.2f -> %.2f)", | |
| best_layer, stance, confidence, adjusted_confidence) | |
| return stance, adjusted_confidence, best_layer | |
| # Ultimate fallback | |
| return "neutral", 0.1, "default" | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| # Factory function for backward compatibility | |
| # βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def build_enhanced_stance_classifier(scorer=None) -> EnhancedStanceClassifier: | |
| """Factory function to create enhanced stance classifier.""" | |
| return EnhancedStanceClassifier() | |
| # Alias for easy import | |
| StanceClassifier = EnhancedStanceClassifier |