Delete bias_utils.py

bias_utils.py

@@ -1,197 +0,0 @@
-# models/bias/bias_utils.py
-
-"""
-Bias Detection Utilities for Penny
-
-Provides zero-shot classification for detecting potential bias in text responses.
-Uses a classification model to identify neutral content vs. biased language patterns.
-"""
-
-import asyncio
-from typing import Dict, Any, Optional, List
-import logging
-
-# --- Logging Setup ---
-logger = logging.getLogger(__name__)
-
-# --- Model Loader Import ---
-try:
-    from app.model_loader import load_model_pipeline
-    MODEL_LOADER_AVAILABLE = True
-except ImportError:
-    MODEL_LOADER_AVAILABLE = False
-    logger.warning("Could not import load_model_pipeline. Bias detection will operate in fallback mode.")
-
-# Global variable to store the loaded pipeline for re-use
-BIAS_PIPELINE: Optional[Any] = None
-AGENT_NAME = "penny-bias-checker"
-
-# Define the labels for Zero-Shot Classification.
-CANDIDATE_LABELS = [
-    "neutral and objective",
-    "contains political bias",
-    "uses emotional language",
-    "is factually biased",
-]
-
-
-def _initialize_bias_pipeline() -> bool:
-    """
-    Initializes the bias detection pipeline only once.
-    
-    Returns:
-        bool: True if pipeline loaded successfully, False otherwise
-    """
-    global BIAS_PIPELINE
-    
-    if BIAS_PIPELINE is not None:
-        return True
-    
-    if not MODEL_LOADER_AVAILABLE:
-        logger.warning(f"{AGENT_NAME}: Model loader not available, pipeline initialization skipped")
-        return False
-    
-    try:
-        logger.info(f"Loading {AGENT_NAME}...")
-        BIAS_PIPELINE = load_model_pipeline(AGENT_NAME)
-        logger.info(f"Model {AGENT_NAME} loaded successfully")
-        return True
-    except Exception as e:
-        logger.error(f"Failed to load {AGENT_NAME}: {e}", exc_info=True)
-        BIAS_PIPELINE = None
-        return False
-
-
-# Attempt to initialize pipeline at module load
-_initialize_bias_pipeline()
-
-
-async def check_bias(text: str) -> Dict[str, Any]:
-    """
-    Runs zero-shot classification to check for bias in the input text.
-    
-    Uses a pre-loaded classification model to analyze text for:
-    - Neutral and objective language
-    - Political bias
-    - Emotional language
-    - Factual bias
-    
-    Args:
-        text: The string of text to analyze for bias
-        
-    Returns:
-        Dictionary containing:
-            - analysis: List of labels with confidence scores, sorted by score
-            - available: Whether the bias detection service is operational
-            - message: Optional error or status message
-            
-    Example:
-        >>> result = await check_bias("This is neutral text.")
-        >>> result['analysis'][0]['label']
-        'neutral and objective'
-    """
-    global BIAS_PIPELINE
-    
-    # Input validation
-    if not text or not isinstance(text, str):
-        logger.warning("check_bias called with invalid text input")
-        return {
-            "analysis": [],
-            "available": False,
-            "message": "Invalid input: text must be a non-empty string"
-        }
-    
-    # Strip text to avoid processing whitespace
-    text = text.strip()
-    if not text:
-        logger.warning("check_bias called with empty text after stripping")
-        return {
-            "analysis": [],
-            "available": False,
-            "message": "Invalid input: text is empty"
-        }
-    
-    # Ensure pipeline is initialized
-    if BIAS_PIPELINE is None:
-        logger.warning(f"{AGENT_NAME} pipeline not available, attempting re-initialization")
-        if not _initialize_bias_pipeline():
-            return {
-                "analysis": [],
-                "available": False,
-                "message": "Bias detection service is currently unavailable"
-            }
-    
-    try:
-        loop = asyncio.get_event_loop()
-        
-        # Run inference in thread pool to avoid blocking
-        results = await loop.run_in_executor(
-            None,
-            lambda: BIAS_PIPELINE(
-                text,
-                CANDIDATE_LABELS,
-                multi_label=True
-            )
-        )
-        
-        # Validate results structure
-        if not results or not isinstance(results, dict):
-            logger.error(f"Bias detection returned unexpected format: {type(results)}")
-            return {
-                "analysis": [],
-                "available": True,
-                "message": "Inference returned unexpected format"
-            }
-        
-        labels = results.get('labels', [])
-        scores = results.get('scores', [])
-        
-        if not labels or not scores:
-            logger.warning("Bias detection returned empty labels or scores")
-            return {
-                "analysis": [],
-                "available": True,
-                "message": "No classification results returned"
-            }
-        
-        # Build analysis results
-        analysis = [
-            {"label": label, "score": float(score)}
-            for label, score in zip(labels, scores)
-        ]
-        
-        # Sort by confidence score (descending)
-        analysis.sort(key=lambda x: x['score'], reverse=True)
-        
-        logger.debug(f"Bias check completed successfully, top result: {analysis[0]['label']} ({analysis[0]['score']:.3f})")
-        
-        return {
-            "analysis": analysis,
-            "available": True
-        }
-    
-    except asyncio.CancelledError:
-        logger.warning("Bias detection task was cancelled")
-        raise
-    
-    except Exception as e:
-        logger.error(f"Error during bias detection inference: {e}", exc_info=True)
-        return {
-            "analysis": [],
-            "available": False,
-            "message": f"Bias detection error: {str(e)}"
-        }
-
-
-def get_bias_pipeline_status() -> Dict[str, Any]:
-    """
-    Returns the current status of the bias detection pipeline.
-    
-    Returns:
-        Dictionary with pipeline availability status
-    """
-    return {
-        "agent_name": AGENT_NAME,
-        "available": BIAS_PIPELINE is not None,
-        "model_loader_available": MODEL_LOADER_AVAILABLE
-    }