Upload bias_utils.py

bias_utils.py

@@ -0,0 +1,190 @@
+# 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
+    }