Delete bias_utils.py

bias_utils.py

@@ -1,190 +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
-import os
-import httpx
-from typing import Dict, Any, Optional, List
-import logging
-
-# --- Logging Setup ---
-logger = logging.getLogger(__name__)
-
-# --- Hugging Face API Configuration ---
-HF_API_URL = "https://api-inference.huggingface.co/models/facebook/bart-large-mnli"
-HF_TOKEN = os.getenv("HF_TOKEN")
-
-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 _is_bias_available() -> bool:
-    """
-    Check if bias detection service is available.
-    
-    Returns:
-        bool: True if HF_TOKEN is configured
-    """
-    return HF_TOKEN is not None and len(HF_TOKEN) > 0
-
-
-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'
-    """
-    
-    # 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"
-        }
-    
-    # Check API availability
-    if not _is_bias_available():
-        logger.warning(f"{AGENT_NAME}: API not configured (missing HF_TOKEN)")
-        return {
-            "analysis": [],
-            "available": False,
-            "message": "Bias detection service is currently unavailable"
-        }
-    
-    try:
-        # Prepare API request for zero-shot classification
-        headers = {"Authorization": f"Bearer {HF_TOKEN}"}
-        payload = {
-            "inputs": text,
-            "parameters": {
-                "candidate_labels": CANDIDATE_LABELS,
-                "multi_label": True
-            }
-        }
-        
-        # Call Hugging Face Inference API
-        async with httpx.AsyncClient(timeout=30.0) as client:
-            response = await client.post(HF_API_URL, json=payload, headers=headers)
-            
-            if response.status_code != 200:
-                logger.error(f"Bias detection API returned status {response.status_code}")
-                return {
-                    "analysis": [],
-                    "available": False,
-                    "message": f"Bias detection API error: {response.status_code}"
-                }
-            
-            results = response.json()
-        
-        # 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 httpx.TimeoutException:
-        logger.error("Bias detection request timed out")
-        return {
-            "analysis": [],
-            "available": False,
-            "message": "Bias detection request timed out"
-        }
-    
-    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": _is_bias_available(),
-        "api_configured": HF_TOKEN is not None
-    }