Upload 5 files

gemma_utils.py

@@ -0,0 +1,216 @@
+# models/gemma/gemma_utils.py
+
+"""
+Gemma Model Utilities for PENNY Project
+Handles text generation using the Gemma-based core language model via Hugging Face Inference API.
+Provides async generation with structured error handling and logging.
+"""
+
+import os
+import asyncio
+import time
+import httpx
+from typing import Dict, Any, Optional
+
+# --- Logging Imports ---
+from app.logging_utils import log_interaction, sanitize_for_logging
+
+# --- Configuration ---
+HF_API_URL = "https://api-inference.huggingface.co/models/google/gemma-7b-it"
+DEFAULT_TIMEOUT = 30.0  # Gemma can take longer to respond
+MAX_RETRIES = 2
+AGENT_NAME = "penny-core-agent"
+
+
+def is_gemma_available() -> bool:
+    """
+    Check if Gemma service is available.
+    
+    Returns:
+        bool: True if HF_TOKEN is configured.
+    """
+    return bool(os.getenv("HF_TOKEN"))
+
+
+async def generate_response(
+    prompt: str,
+    max_new_tokens: int = 256,
+    temperature: float = 0.7,
+    tenant_id: Optional[str] = None,
+) -> Dict[str, Any]:
+    """
+    Runs text generation using Gemma via Hugging Face Inference API.
+
+    Args:
+        prompt: The conversational or instruction prompt.
+        max_new_tokens: The maximum number of tokens to generate (default: 256).
+        temperature: Controls randomness in generation (default: 0.7).
+        tenant_id: Optional tenant identifier for logging.
+
+    Returns:
+        A dictionary containing:
+            - response (str): The generated text
+            - available (bool): Whether the service was available
+            - error (str, optional): Error message if generation failed
+            - response_time_ms (int, optional): Generation time in milliseconds
+    """
+    start_time = time.time()
+    
+    # Check API token availability
+    HF_TOKEN = os.getenv("HF_TOKEN")
+    if not HF_TOKEN:
+        log_interaction(
+            intent="gemma_generate",
+            tenant_id=tenant_id,
+            success=False,
+            error="HF_TOKEN not configured",
+            fallback_used=True
+        )
+        return {
+            "response": "I'm having trouble accessing my language model right now. Please try again in a moment!",
+            "available": False,
+            "error": "HF_TOKEN not configured"
+        }
+
+    # Validate inputs
+    if not prompt or not isinstance(prompt, str):
+        log_interaction(
+            intent="gemma_generate",
+            tenant_id=tenant_id,
+            success=False,
+            error="Invalid prompt provided"
+        )
+        return {
+            "response": "I didn't receive a valid prompt. Could you try again?",
+            "available": True,
+            "error": "Invalid input"
+        }
+
+    # Configure generation parameters
+    payload = {
+        "inputs": prompt,
+        "parameters": {
+            "max_new_tokens": max_new_tokens,
+            "temperature": temperature,
+            "do_sample": True if temperature > 0.0 else False,
+            "return_full_text": False
+        }
+    }
+    
+    headers = {
+        "Authorization": f"Bearer {HF_TOKEN}",
+        "Content-Type": "application/json"
+    }
+
+    # Retry logic for API calls
+    for attempt in range(MAX_RETRIES):
+        try:
+            async with httpx.AsyncClient(timeout=DEFAULT_TIMEOUT) as client:
+                response = await client.post(HF_API_URL, json=payload, headers=headers)
+                response.raise_for_status()
+                result = response.json()
+            
+            response_time_ms = int((time.time() - start_time) * 1000)
+            
+            # Parse response
+            if isinstance(result, list) and len(result) > 0:
+                generated_text = result[0].get("generated_text", "").strip()
+                
+                # Log slow responses
+                if response_time_ms > 5000:
+                    log_interaction(
+                        intent="gemma_generate_slow",
+                        tenant_id=tenant_id,
+                        success=True,
+                        response_time_ms=response_time_ms,
+                        details="Slow generation detected"
+                    )
+                
+                log_interaction(
+                    intent="gemma_generate",
+                    tenant_id=tenant_id,
+                    success=True,
+                    response_time_ms=response_time_ms,
+                    prompt_preview=sanitize_for_logging(prompt[:100])
+                )
+                
+                return {
+                    "response": generated_text,
+                    "available": True,
+                    "response_time_ms": response_time_ms
+                }
+            
+            # Unexpected output format
+            log_interaction(
+                intent="gemma_generate",
+                tenant_id=tenant_id,
+                success=False,
+                error="Unexpected API response format",
+                response_time_ms=response_time_ms
+            )
+            
+            return {
+                "response": "I got an unexpected response from my language model. Let me try to help you another way!",
+                "available": True,
+                "error": "Unexpected output format"
+            }
+        
+        except httpx.TimeoutException:
+            if attempt < MAX_RETRIES - 1:
+                await asyncio.sleep(1)  # Wait before retry
+                continue
+            
+            response_time_ms = int((time.time() - start_time) * 1000)
+            log_interaction(
+                intent="gemma_generate",
+                tenant_id=tenant_id,
+                success=False,
+                error="API timeout after retries",
+                response_time_ms=response_time_ms
+            )
+            
+            return {
+                "response": "I'm taking too long to respond. Please try again!",
+                "available": False,
+                "error": "Timeout",
+                "response_time_ms": response_time_ms
+            }
+        
+        except httpx.HTTPStatusError as e:
+            response_time_ms = int((time.time() - start_time) * 1000)
+            log_interaction(
+                intent="gemma_generate",
+                tenant_id=tenant_id,
+                success=False,
+                error=f"HTTP {e.response.status_code}",
+                response_time_ms=response_time_ms
+            )
+            
+            return {
+                "response": "I'm having trouble generating a response right now. Please try again!",
+                "available": False,
+                "error": f"HTTP {e.response.status_code}",
+                "response_time_ms": response_time_ms
+            }
+        
+        except Exception as e:
+            if attempt < MAX_RETRIES - 1:
+                await asyncio.sleep(1)
+                continue
+            
+            response_time_ms = int((time.time() - start_time) * 1000)
+            log_interaction(
+                intent="gemma_generate",
+                tenant_id=tenant_id,
+                success=False,
+                error=str(e),
+                response_time_ms=response_time_ms,
+                fallback_used=True
+            )
+            
+            return {
+                "response": "I'm having trouble generating a response right now. Please try again!",
+                "available": False,
+                "error": str(e),
+                "response_time_ms": response_time_ms
+            }

layoutlm_utils.py

@@ -0,0 +1,359 @@
+# models/layoutlm/layoutlm_utils.py
+
+"""
+LayoutLM Model Utilities for PENNY Project
+Handles document structure extraction and field recognition for civic forms and documents.
+Provides async document processing with structured error handling and logging.
+"""
+
+import asyncio
+import time
+from typing import Dict, Any, Optional, List
+from io import BytesIO
+
+# --- Logging Imports ---
+from app.logging_utils import log_interaction, sanitize_for_logging
+
+# --- Model Loader Import ---
+try:
+    from app.model_loader import load_model_pipeline
+    MODEL_LOADER_AVAILABLE = True
+except ImportError:
+    MODEL_LOADER_AVAILABLE = False
+    import logging
+    logging.getLogger(__name__).warning("Could not import load_model_pipeline. LayoutLM service unavailable.")
+
+# Global variable to store the loaded pipeline for re-use
+LAYOUTLM_PIPELINE: Optional[Any] = None
+AGENT_NAME = "penny-doc-agent"
+INITIALIZATION_ATTEMPTED = False
+
+
+def _initialize_layoutlm_pipeline() -> bool:
+    """
+    Initializes the LayoutLM pipeline only once.
+    
+    Returns:
+        bool: True if initialization succeeded, False otherwise.
+    """
+    global LAYOUTLM_PIPELINE, INITIALIZATION_ATTEMPTED
+    
+    if INITIALIZATION_ATTEMPTED:
+        return LAYOUTLM_PIPELINE is not None
+    
+    INITIALIZATION_ATTEMPTED = True
+    
+    if not MODEL_LOADER_AVAILABLE:
+        log_interaction(
+            intent="layoutlm_initialization",
+            success=False,
+            error="model_loader unavailable"
+        )
+        return False
+    
+    try:
+        log_interaction(
+            intent="layoutlm_initialization",
+            success=None,
+            details=f"Loading {AGENT_NAME}"
+        )
+        
+        LAYOUTLM_PIPELINE = load_model_pipeline(AGENT_NAME)
+        
+        if LAYOUTLM_PIPELINE is None:
+            log_interaction(
+                intent="layoutlm_initialization",
+                success=False,
+                error="Pipeline returned None"
+            )
+            return False
+        
+        log_interaction(
+            intent="layoutlm_initialization",
+            success=True,
+            details=f"Model {AGENT_NAME} loaded successfully"
+        )
+        return True
+        
+    except Exception as e:
+        log_interaction(
+            intent="layoutlm_initialization",
+            success=False,
+            error=str(e)
+        )
+        return False
+
+
+# Attempt initialization at module load
+_initialize_layoutlm_pipeline()
+
+
+def is_layoutlm_available() -> bool:
+    """
+    Check if LayoutLM service is available.
+    
+    Returns:
+        bool: True if LayoutLM pipeline is loaded and ready.
+    """
+    return LAYOUTLM_PIPELINE is not None
+
+
+async def extract_document_data(
+    file_bytes: bytes,
+    file_name: str,
+    tenant_id: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Processes a document (e.g., PDF, image) using LayoutLM to extract structured data.
+
+    Args:
+        file_bytes: The raw bytes of the uploaded file.
+        file_name: The original name of the file (e.g., form.pdf).
+        tenant_id: Optional tenant identifier for logging.
+
+    Returns:
+        A dictionary containing:
+            - status (str): "success" or "error"
+            - extracted_fields (dict, optional): Extracted key-value pairs
+            - available (bool): Whether the service was available
+            - message (str, optional): Error message if extraction failed
+            - response_time_ms (int, optional): Processing time in milliseconds
+    """
+    start_time = time.time()
+    
+    global LAYOUTLM_PIPELINE
+
+    # Check availability
+    if not is_layoutlm_available():
+        log_interaction(
+            intent="layoutlm_extract",
+            tenant_id=tenant_id,
+            success=False,
+            error="LayoutLM pipeline not available",
+            fallback_used=True
+        )
+        return {
+            "status": "error",
+            "available": False,
+            "message": "Document processing is temporarily unavailable. Please try uploading your document again in a moment!"
+        }
+
+    # Validate inputs
+    if not file_bytes or not isinstance(file_bytes, bytes):
+        log_interaction(
+            intent="layoutlm_extract",
+            tenant_id=tenant_id,
+            success=False,
+            error="Invalid file_bytes provided"
+        )
+        return {
+            "status": "error",
+            "available": True,
+            "message": "I didn't receive valid document data. Could you try uploading your file again?"
+        }
+    
+    if not file_name or not isinstance(file_name, str):
+        log_interaction(
+            intent="layoutlm_extract",
+            tenant_id=tenant_id,
+            success=False,
+            error="Invalid file_name provided"
+        )
+        return {
+            "status": "error",
+            "available": True,
+            "message": "I need a valid file name to process your document. Please try again!"
+        }
+
+    # Check file size (prevent processing extremely large files)
+    file_size_mb = len(file_bytes) / (1024 * 1024)
+    if file_size_mb > 50:  # 50 MB limit
+        log_interaction(
+            intent="layoutlm_extract",
+            tenant_id=tenant_id,
+            success=False,
+            error=f"File too large: {file_size_mb:.2f}MB",
+            file_name=sanitize_for_logging(file_name)
+        )
+        return {
+            "status": "error",
+            "available": True,
+            "message": f"Your file is too large ({file_size_mb:.1f}MB). Please upload a document smaller than 50MB."
+        }
+
+    try:
+        # --- Real-world step (PLACEHOLDER) ---
+        # In a real implementation, you would:
+        # 1. Use a library (e.g., PyMuPDF, pdf2image) to convert PDF bytes to image(s).
+        # 2. Use PIL/Pillow to load the image(s) from bytes.
+        # 3. Pass the PIL Image object to the LayoutLM pipeline.
+        
+        # For now, we use a simple mock placeholder for the image object:
+        image_mock = {
+            "file_name": file_name,
+            "byte_size": len(file_bytes)
+        }
+
+        loop = asyncio.get_event_loop()
+        
+        # Run model inference in thread executor
+        results = await loop.run_in_executor(
+            None,
+            lambda: LAYOUTLM_PIPELINE(image_mock)
+        )
+        
+        response_time_ms = int((time.time() - start_time) * 1000)
+        
+        # Validate results
+        if not results or not isinstance(results, list):
+            log_interaction(
+                intent="layoutlm_extract",
+                tenant_id=tenant_id,
+                success=False,
+                error="Unexpected model output format",
+                response_time_ms=response_time_ms,
+                file_name=sanitize_for_logging(file_name)
+            )
+            return {
+                "status": "error",
+                "available": True,
+                "message": "I had trouble understanding the document structure. The file might be corrupted or in an unsupported format."
+            }
+        
+        # Convert model output (list of dicts) into a clean key-value format
+        extracted_data = {}
+        for item in results:
+            if isinstance(item, dict) and 'label' in item and 'text' in item:
+                label_key = item['label'].lower().strip()
+                text_value = str(item['text']).strip()
+                
+                # Avoid empty values
+                if text_value:
+                    extracted_data[label_key] = text_value
+        
+        # Log slow processing
+        if response_time_ms > 10000:  # 10 seconds
+            log_interaction(
+                intent="layoutlm_extract_slow",
+                tenant_id=tenant_id,
+                success=True,
+                response_time_ms=response_time_ms,
+                details="Slow document processing detected",
+                file_name=sanitize_for_logging(file_name)
+            )
+        
+        log_interaction(
+            intent="layoutlm_extract",
+            tenant_id=tenant_id,
+            success=True,
+            response_time_ms=response_time_ms,
+            file_name=sanitize_for_logging(file_name),
+            fields_extracted=len(extracted_data)
+        )
+        
+        return {
+            "status": "success",
+            "extracted_fields": extracted_data,
+            "available": True,
+            "response_time_ms": response_time_ms,
+            "fields_count": len(extracted_data)
+        }
+
+    except asyncio.CancelledError:
+        log_interaction(
+            intent="layoutlm_extract",
+            tenant_id=tenant_id,
+            success=False,
+            error="Processing cancelled",
+            file_name=sanitize_for_logging(file_name)
+        )
+        raise
+        
+    except Exception as e:
+        response_time_ms = int((time.time() - start_time) * 1000)
+        
+        log_interaction(
+            intent="layoutlm_extract",
+            tenant_id=tenant_id,
+            success=False,
+            error=str(e),
+            response_time_ms=response_time_ms,
+            file_name=sanitize_for_logging(file_name),
+            fallback_used=True
+        )
+        
+        return {
+            "status": "error",
+            "available": False,
+            "message": f"I encountered an issue while processing your document. Please try again, or contact support if this continues!",
+            "error": str(e),
+            "response_time_ms": response_time_ms
+        }
+
+
+async def validate_document_fields(
+    extracted_fields: Dict[str, str],
+    required_fields: List[str],
+    tenant_id: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Validates that required fields were successfully extracted from a document.
+    
+    Args:
+        extracted_fields: Dictionary of extracted field names and values.
+        required_fields: List of field names that must be present.
+        tenant_id: Optional tenant identifier for logging.
+    
+    Returns:
+        A dictionary containing:
+            - valid (bool): Whether all required fields are present
+            - missing_fields (list): List of missing required fields
+            - present_fields (list): List of found required fields
+    """
+    if not isinstance(extracted_fields, dict):
+        log_interaction(
+            intent="layoutlm_validate",
+            tenant_id=tenant_id,
+            success=False,
+            error="Invalid extracted_fields type"
+        )
+        return {
+            "valid": False,
+            "missing_fields": required_fields,
+            "present_fields": []
+        }
+    
+    if not isinstance(required_fields, list):
+        log_interaction(
+            intent="layoutlm_validate",
+            tenant_id=tenant_id,
+            success=False,
+            error="Invalid required_fields type"
+        )
+        return {
+            "valid": False,
+            "missing_fields": [],
+            "present_fields": []
+        }
+    
+    # Normalize field names for case-insensitive comparison
+    extracted_keys = {k.lower().strip() for k in extracted_fields.keys()}
+    required_keys = {f.lower().strip() for f in required_fields}
+    
+    present_fields = [f for f in required_fields if f.lower().strip() in extracted_keys]
+    missing_fields = [f for f in required_fields if f.lower().strip() not in extracted_keys]
+    
+    is_valid = len(missing_fields) == 0
+    
+    log_interaction(
+        intent="layoutlm_validate",
+        tenant_id=tenant_id,
+        success=is_valid,
+        details=f"Validated {len(present_fields)}/{len(required_fields)} required fields"
+    )
+    
+    return {
+        "valid": is_valid,
+        "missing_fields": missing_fields,
+        "present_fields": present_fields
+    }

model_config.json

@@ -0,0 +1,47 @@
+{
+  "penny-core-agent": {
+    "model_name": "google/gemma-7b-it",
+    "task": "text-generation",
+    "endpoint": "huggingface-api",
+    "api_url": "https://api-inference.huggingface.co/models/google/gemma-7b-it",
+    "timeout_seconds": 30,
+    "max_retries": 2,
+    "description": "Penny's core conversational AI for civic engagement responses"
+  },
+  "penny-doc-agent": {
+    "model_name": "microsoft/layoutlmv3-base",
+    "task": "pdf-extraction",
+    "endpoint": "huggingface-api",
+    "api_url": "https://api-inference.huggingface.co/models/microsoft/layoutlmv3-base",
+    "timeout_seconds": 45,
+    "max_retries": 2,
+    "description": "Document understanding and PDF extraction for civic documents"
+  },
+  "penny-translate-agent": {
+    "model_name": "facebook/nllb-200-distilled-600M",
+    "task": "translation",
+    "endpoint": "huggingface-api",
+    "api_url": "https://api-inference.huggingface.co/models/facebook/nllb-200-distilled-600M",
+    "timeout_seconds": 20,
+    "max_retries": 2,
+    "description": "Multilingual translation service for accessible civic information"
+  },
+  "penny-sentiment-agent": {
+    "model_name": "cardiffnlp/twitter-roberta-base-sentiment",
+    "task": "sentiment-analysis",
+    "endpoint": "huggingface-api",
+    "api_url": "https://api-inference.huggingface.co/models/cardiffnlp/twitter-roberta-base-sentiment",
+    "timeout_seconds": 15,
+    "max_retries": 2,
+    "description": "Sentiment analysis for community feedback and engagement monitoring"
+  },
+  "penny-bias-checker": {
+    "model_name": "facebook/bart-large-mnli",
+    "task": "bias-detection",
+    "endpoint": "huggingface-api",
+    "api_url": "https://api-inference.huggingface.co/models/facebook/bart-large-mnli",
+    "timeout_seconds": 20,
+    "max_retries": 2,
+    "description": "Bias detection to ensure fair and equitable civic information"
+  }
+}

sentiment_utils.py

@@ -0,0 +1,450 @@
+# models/sentiment/sentiment_utils.py
+
+"""
+Sentiment Analysis Model Utilities for PENNY Project
+Handles text sentiment classification for user input analysis and content moderation.
+Provides async sentiment analysis with structured error handling and logging.
+"""
+
+import asyncio
+import time
+import os
+import httpx
+from typing import Dict, Any, Optional, List
+
+# --- Logging Imports ---
+from app.logging_utils import log_interaction, sanitize_for_logging
+
+# --- Hugging Face API Configuration ---
+HF_API_URL = "https://api-inference.huggingface.co/models/cardiffnlp/twitter-roberta-base-sentiment"
+HF_TOKEN = os.getenv("HF_TOKEN")
+
+AGENT_NAME = "penny-sentiment-agent"
+
+
+def is_sentiment_available() -> bool:
+    """
+    Check if sentiment analysis service is available.
+    
+    Returns:
+        bool: True if sentiment API is configured and ready.
+    """
+    return HF_TOKEN is not None and len(HF_TOKEN) > 0
+
+
+async def get_sentiment_analysis(
+    text: str,
+    tenant_id: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Runs sentiment analysis on the input text using the loaded pipeline.
+
+    Args:
+        text: The string of text to analyze.
+        tenant_id: Optional tenant identifier for logging.
+
+    Returns:
+        A dictionary containing:
+            - label (str): Sentiment label (e.g., "POSITIVE", "NEGATIVE", "NEUTRAL")
+            - score (float): Confidence score for the sentiment prediction
+            - available (bool): Whether the service was available
+            - message (str, optional): Error message if analysis failed
+            - response_time_ms (int, optional): Analysis time in milliseconds
+    """
+    start_time = time.time()
+
+    # Check availability
+    if not is_sentiment_available():
+        log_interaction(
+            intent="sentiment_analysis",
+            tenant_id=tenant_id,
+            success=False,
+            error="Sentiment API not configured (missing HF_TOKEN)",
+            fallback_used=True
+        )
+        return {
+            "label": "UNKNOWN",
+            "score": 0.0,
+            "available": False,
+            "message": "Sentiment analysis is temporarily unavailable."
+        }
+
+    # Validate input
+    if not text or not isinstance(text, str):
+        log_interaction(
+            intent="sentiment_analysis",
+            tenant_id=tenant_id,
+            success=False,
+            error="Invalid text input"
+        )
+        return {
+            "label": "ERROR",
+            "score": 0.0,
+            "available": True,
+            "message": "Invalid text input provided."
+        }
+    
+    # Check text length (prevent processing extremely long texts)
+    if len(text) > 10000:  # 10k character limit
+        log_interaction(
+            intent="sentiment_analysis",
+            tenant_id=tenant_id,
+            success=False,
+            error=f"Text too long: {len(text)} characters",
+            text_preview=sanitize_for_logging(text[:100])
+        )
+        return {
+            "label": "ERROR",
+            "score": 0.0,
+            "available": True,
+            "message": "Text is too long for sentiment analysis (max 10,000 characters)."
+        }
+
+    try:
+        # Prepare API request
+        headers = {"Authorization": f"Bearer {HF_TOKEN}"}
+        payload = {"inputs": text}
+        
+        # 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)
+            
+            response_time_ms = int((time.time() - start_time) * 1000)
+            
+            if response.status_code != 200:
+                log_interaction(
+                    intent="sentiment_analysis",
+                    tenant_id=tenant_id,
+                    success=False,
+                    error=f"API returned status {response.status_code}",
+                    response_time_ms=response_time_ms,
+                    text_preview=sanitize_for_logging(text[:100]),
+                    fallback_used=True
+                )
+                return {
+                    "label": "ERROR",
+                    "score": 0.0,
+                    "available": False,
+                    "message": f"Sentiment API error: {response.status_code}",
+                    "response_time_ms": response_time_ms
+                }
+            
+            results = response.json()
+        
+        # Validate results
+        # API returns: [[{"label": "LABEL_2", "score": 0.95}, ...]]
+        if not results or not isinstance(results, list) or len(results) == 0:
+            log_interaction(
+                intent="sentiment_analysis",
+                tenant_id=tenant_id,
+                success=False,
+                error="Empty or invalid model output",
+                response_time_ms=response_time_ms,
+                text_preview=sanitize_for_logging(text[:100])
+            )
+            return {
+                "label": "ERROR",
+                "score": 0.0,
+                "available": True,
+                "message": "Sentiment analysis returned unexpected format."
+            }
+        
+        # Get the first (highest scoring) result
+        result_list = results[0] if isinstance(results[0], list) else results
+        
+        if not result_list or len(result_list) == 0:
+            log_interaction(
+                intent="sentiment_analysis",
+                tenant_id=tenant_id,
+                success=False,
+                error="Empty result list",
+                response_time_ms=response_time_ms,
+                text_preview=sanitize_for_logging(text[:100])
+            )
+            return {
+                "label": "ERROR",
+                "score": 0.0,
+                "available": True,
+                "message": "Sentiment analysis returned unexpected format."
+            }
+        
+        result = result_list[0]
+        
+        # Validate result structure
+        if not isinstance(result, dict) or 'label' not in result or 'score' not in result:
+            log_interaction(
+                intent="sentiment_analysis",
+                tenant_id=tenant_id,
+                success=False,
+                error="Invalid result structure",
+                response_time_ms=response_time_ms,
+                text_preview=sanitize_for_logging(text[:100])
+            )
+            return {
+                "label": "ERROR",
+                "score": 0.0,
+                "available": True,
+                "message": "Sentiment analysis returned unexpected format."
+            }
+        
+        # Map RoBERTa labels to readable format
+        # LABEL_0 = NEGATIVE, LABEL_1 = NEUTRAL, LABEL_2 = POSITIVE
+        label_mapping = {
+            "LABEL_0": "NEGATIVE",
+            "LABEL_1": "NEUTRAL",
+            "LABEL_2": "POSITIVE"
+        }
+        label = label_mapping.get(result['label'], result['label'])
+        
+        # Log slow analysis
+        if response_time_ms > 3000:  # 3 seconds
+            log_interaction(
+                intent="sentiment_analysis_slow",
+                tenant_id=tenant_id,
+                success=True,
+                response_time_ms=response_time_ms,
+                details="Slow sentiment analysis detected",
+                text_length=len(text)
+            )
+        
+        log_interaction(
+            intent="sentiment_analysis",
+            tenant_id=tenant_id,
+            success=True,
+            response_time_ms=response_time_ms,
+            sentiment_label=label,
+            sentiment_score=result.get('score'),
+            text_length=len(text)
+        )
+        
+        return {
+            "label": label,
+            "score": float(result['score']),
+            "available": True,
+            "response_time_ms": response_time_ms
+        }
+
+    except httpx.TimeoutException:
+        response_time_ms = int((time.time() - start_time) * 1000)
+        log_interaction(
+            intent="sentiment_analysis",
+            tenant_id=tenant_id,
+            success=False,
+            error="Sentiment analysis request timed out",
+            response_time_ms=response_time_ms,
+            text_preview=sanitize_for_logging(text[:100]),
+            fallback_used=True
+        )
+        return {
+            "label": "ERROR",
+            "score": 0.0,
+            "available": False,
+            "message": "Sentiment analysis request timed out.",
+            "response_time_ms": response_time_ms
+        }
+
+    except asyncio.CancelledError:
+        log_interaction(
+            intent="sentiment_analysis",
+            tenant_id=tenant_id,
+            success=False,
+            error="Analysis cancelled"
+        )
+        raise
+        
+    except Exception as e:
+        response_time_ms = int((time.time() - start_time) * 1000)
+        
+        log_interaction(
+            intent="sentiment_analysis",
+            tenant_id=tenant_id,
+            success=False,
+            error=str(e),
+            response_time_ms=response_time_ms,
+            text_preview=sanitize_for_logging(text[:100]),
+            fallback_used=True
+        )
+        
+        return {
+            "label": "ERROR",
+            "score": 0.0,
+            "available": False,
+            "message": "An error occurred during sentiment analysis.",
+            "error": str(e),
+            "response_time_ms": response_time_ms
+        }
+
+
+async def analyze_sentiment_batch(
+    texts: List[str],
+    tenant_id: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Runs sentiment analysis on a batch of texts for efficiency.
+    
+    Args:
+        texts: List of text strings to analyze.
+        tenant_id: Optional tenant identifier for logging.
+    
+    Returns:
+        A dictionary containing:
+            - results (list): List of sentiment analysis results for each text
+            - available (bool): Whether the service was available
+            - total_analyzed (int): Number of texts successfully analyzed
+            - response_time_ms (int, optional): Total batch analysis time
+    """
+    start_time = time.time()
+
+    # Check availability
+    if not is_sentiment_available():
+        log_interaction(
+            intent="sentiment_batch_analysis",
+            tenant_id=tenant_id,
+            success=False,
+            error="Sentiment API not configured (missing HF_TOKEN)",
+            batch_size=len(texts) if texts else 0
+        )
+        return {
+            "results": [],
+            "available": False,
+            "total_analyzed": 0,
+            "message": "Sentiment analysis is temporarily unavailable."
+        }
+
+    # Validate input
+    if not texts or not isinstance(texts, list):
+        log_interaction(
+            intent="sentiment_batch_analysis",
+            tenant_id=tenant_id,
+            success=False,
+            error="Invalid texts input"
+        )
+        return {
+            "results": [],
+            "available": True,
+            "total_analyzed": 0,
+            "message": "Invalid batch input provided."
+        }
+    
+    # Filter valid texts and limit batch size
+    valid_texts = [t for t in texts if isinstance(t, str) and t.strip()]
+    if len(valid_texts) > 100:  # Batch size limit
+        valid_texts = valid_texts[:100]
+    
+    if not valid_texts:
+        log_interaction(
+            intent="sentiment_batch_analysis",
+            tenant_id=tenant_id,
+            success=False,
+            error="No valid texts in batch"
+        )
+        return {
+            "results": [],
+            "available": True,
+            "total_analyzed": 0,
+            "message": "No valid texts provided for analysis."
+        }
+
+    try:
+        # Prepare API request with batch input
+        headers = {"Authorization": f"Bearer {HF_TOKEN}"}
+        payload = {"inputs": valid_texts}
+        
+        # Call Hugging Face Inference API
+        async with httpx.AsyncClient(timeout=60.0) as client:  # Longer timeout for batch
+            response = await client.post(HF_API_URL, json=payload, headers=headers)
+            
+            response_time_ms = int((time.time() - start_time) * 1000)
+            
+            if response.status_code != 200:
+                log_interaction(
+                    intent="sentiment_batch_analysis",
+                    tenant_id=tenant_id,
+                    success=False,
+                    error=f"API returned status {response.status_code}",
+                    response_time_ms=response_time_ms,
+                    batch_size=len(valid_texts)
+                )
+                return {
+                    "results": [],
+                    "available": False,
+                    "total_analyzed": 0,
+                    "message": f"Sentiment API error: {response.status_code}",
+                    "response_time_ms": response_time_ms
+                }
+            
+            results = response.json()
+        
+        # Process results and map labels
+        label_mapping = {
+            "LABEL_0": "NEGATIVE",
+            "LABEL_1": "NEUTRAL",
+            "LABEL_2": "POSITIVE"
+        }
+        
+        processed_results = []
+        if results and isinstance(results, list):
+            for item in results:
+                if isinstance(item, list) and len(item) > 0:
+                    top_result = item[0]
+                    if isinstance(top_result, dict) and 'label' in top_result:
+                        processed_results.append({
+                            "label": label_mapping.get(top_result['label'], top_result['label']),
+                            "score": float(top_result.get('score', 0.0))
+                        })
+        
+        log_interaction(
+            intent="sentiment_batch_analysis",
+            tenant_id=tenant_id,
+            success=True,
+            response_time_ms=response_time_ms,
+            batch_size=len(valid_texts),
+            total_analyzed=len(processed_results)
+        )
+        
+        return {
+            "results": processed_results,
+            "available": True,
+            "total_analyzed": len(processed_results),
+            "response_time_ms": response_time_ms
+        }
+
+    except httpx.TimeoutException:
+        response_time_ms = int((time.time() - start_time) * 1000)
+        log_interaction(
+            intent="sentiment_batch_analysis",
+            tenant_id=tenant_id,
+            success=False,
+            error="Batch sentiment analysis timed out",
+            response_time_ms=response_time_ms,
+            batch_size=len(valid_texts)
+        )
+        return {
+            "results": [],
+            "available": False,
+            "total_analyzed": 0,
+            "message": "Batch sentiment analysis timed out.",
+            "error": "Request timeout",
+            "response_time_ms": response_time_ms
+        }
+
+    except Exception as e:
+        response_time_ms = int((time.time() - start_time) * 1000)
+        
+        log_interaction(
+            intent="sentiment_batch_analysis",
+            tenant_id=tenant_id,
+            success=False,
+            error=str(e),
+            response_time_ms=response_time_ms,
+            batch_size=len(valid_texts)
+        )
+        
+        return {
+            "results": [],
+            "available": False,
+            "total_analyzed": 0,
+            "message": "An error occurred during batch sentiment analysis.",
+            "error": str(e),
+            "response_time_ms": response_time_ms
+        }

translation_utils.py

@@ -0,0 +1,578 @@
+# models/translation/translation_utils.py
+
+"""
+Translation Model Utilities for PENNY Project
+Handles multilingual translation using NLLB-200 for civic engagement accessibility.
+Provides async translation with structured error handling and language code normalization.
+"""
+
+import asyncio
+import time
+import os
+import httpx
+from typing import Dict, Any, Optional, List
+
+# --- Logging Imports ---
+from app.logging_utils import log_interaction, sanitize_for_logging
+
+# --- Hugging Face API Configuration ---
+HF_API_URL = "https://api-inference.huggingface.co/models/facebook/nllb-200-distilled-600M"
+HF_TOKEN = os.getenv("HF_TOKEN")
+
+AGENT_NAME = "penny-translate-agent"
+SERVICE_AVAILABLE = True  # Assume available since we're using API
+
+# NLLB-200 Language Code Mapping (Common languages for civic engagement)
+LANGUAGE_CODES = {
+    # English variants
+    "english": "eng_Latn",
+    "en": "eng_Latn",
+    
+    # Spanish variants
+    "spanish": "spa_Latn",
+    "es": "spa_Latn",
+    "español": "spa_Latn",
+    
+    # French
+    "french": "fra_Latn",
+    "fr": "fra_Latn",
+    "français": "fra_Latn",
+    
+    # Mandarin Chinese
+    "chinese": "zho_Hans",
+    "mandarin": "zho_Hans",
+    "zh": "zho_Hans",
+    
+    # Arabic
+    "arabic": "arb_Arab",
+    "ar": "arb_Arab",
+    
+    # Hindi
+    "hindi": "hin_Deva",
+    "hi": "hin_Deva",
+    
+    # Portuguese
+    "portuguese": "por_Latn",
+    "pt": "por_Latn",
+    
+    # Russian
+    "russian": "rus_Cyrl",
+    "ru": "rus_Cyrl",
+    
+    # German
+    "german": "deu_Latn",
+    "de": "deu_Latn",
+    
+    # Vietnamese
+    "vietnamese": "vie_Latn",
+    "vi": "vie_Latn",
+    
+    # Tagalog
+    "tagalog": "tgl_Latn",
+    "tl": "tgl_Latn",
+    
+    # Urdu
+    "urdu": "urd_Arab",
+    "ur": "urd_Arab",
+    
+    # Swahili
+    "swahili": "swh_Latn",
+    "sw": "swh_Latn",
+}
+
+# Pre-translated civic phrases for common queries
+CIVIC_PHRASES = {
+    "eng_Latn": {
+        "voting_location": "Where is my polling place?",
+        "voter_registration": "How do I register to vote?",
+        "city_services": "What city services are available?",
+        "report_issue": "I want to report a problem.",
+        "contact_city": "How do I contact city hall?",
+    },
+    "spa_Latn": {
+        "voting_location": "¿Dónde está mi lugar de votación?",
+        "voter_registration": "¿Cómo me registro para votar?",
+        "city_services": "¿Qué servicios de la ciudad están disponibles?",
+        "report_issue": "Quiero reportar un problema.",
+        "contact_city": "¿Cómo contacto al ayuntamiento?",
+    }
+}
+
+
+def is_translation_available() -> bool:
+    """
+    Check if translation service is available.
+    
+    Returns:
+        bool: True if translation API is configured and ready.
+    """
+    return HF_TOKEN is not None and len(HF_TOKEN) > 0
+
+
+def normalize_language_code(lang: str) -> str:
+    """
+    Converts common language names/codes to NLLB-200 format.
+    
+    Args:
+        lang: Language name or code (e.g., "spanish", "es", "español")
+        
+    Returns:
+        NLLB-200 language code (e.g., "spa_Latn")
+    """
+    if not lang or not isinstance(lang, str):
+        return "eng_Latn"  # Default to English
+    
+    lang_lower = lang.lower().strip()
+    
+    # Check if it's already in NLLB format (contains underscore)
+    if "_" in lang_lower:
+        return lang_lower
+    
+    # Look up in mapping
+    return LANGUAGE_CODES.get(lang_lower, lang_lower)
+
+
+def get_supported_languages() -> List[str]:
+    """
+    Get list of supported language codes.
+    
+    Returns:
+        List of NLLB-200 language codes supported by PENNY.
+    """
+    return list(set(LANGUAGE_CODES.values()))
+
+
+async def translate_text(
+    text: str,
+    source_language: str = "eng_Latn",
+    target_language: str = "spa_Latn",
+    tenant_id: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Translates text from source language to target language using NLLB-200.
+
+    Args:
+        text: The text to translate.
+        source_language: Source language code (e.g., "eng_Latn", "spanish", "es")
+        target_language: Target language code (e.g., "spa_Latn", "french", "fr")
+        tenant_id: Optional tenant identifier for logging.
+
+    Returns:
+        A dictionary containing:
+            - translated_text (str): The translated text
+            - source_lang (str): Normalized source language code
+            - target_lang (str): Normalized target language code
+            - original_text (str): The input text
+            - available (bool): Whether the service was available
+            - error (str, optional): Error message if translation failed
+            - response_time_ms (int, optional): Translation time in milliseconds
+    """
+    start_time = time.time()
+
+    # Check availability
+    if not is_translation_available():
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=False,
+            error="Translation API not configured (missing HF_TOKEN)",
+            fallback_used=True
+        )
+        return {
+            "translated_text": text,  # Return original text as fallback
+            "source_lang": source_language,
+            "target_lang": target_language,
+            "original_text": text,
+            "available": False,
+            "error": "Translation service is temporarily unavailable."
+        }
+
+    # Validate input
+    if not text or not isinstance(text, str):
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=False,
+            error="Invalid text input"
+        )
+        return {
+            "translated_text": "",
+            "source_lang": source_language,
+            "target_lang": target_language,
+            "original_text": text if isinstance(text, str) else "",
+            "available": True,
+            "error": "Invalid text input provided."
+        }
+    
+    # Check text length (prevent processing extremely long texts)
+    if len(text) > 5000:  # 5k character limit for translation
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=False,
+            error=f"Text too long: {len(text)} characters",
+            text_preview=sanitize_for_logging(text[:100])
+        )
+        return {
+            "translated_text": text,
+            "source_lang": source_language,
+            "target_lang": target_language,
+            "original_text": text,
+            "available": True,
+            "error": "Text is too long for translation (max 5,000 characters)."
+        }
+
+    # Normalize language codes
+    src_lang = normalize_language_code(source_language)
+    tgt_lang = normalize_language_code(target_language)
+    
+    # Skip translation if source and target are the same
+    if src_lang == tgt_lang:
+        log_interaction(
+            intent="translation_skipped",
+            tenant_id=tenant_id,
+            success=True,
+            details="Source and target languages are identical"
+        )
+        return {
+            "translated_text": text,
+            "source_lang": src_lang,
+            "target_lang": tgt_lang,
+            "original_text": text,
+            "available": True,
+            "skipped": True
+        }
+
+    try:
+        # Prepare API request
+        headers = {"Authorization": f"Bearer {HF_TOKEN}"}
+        payload = {
+            "inputs": text,
+            "parameters": {
+                "src_lang": src_lang,
+                "tgt_lang": tgt_lang
+            }
+        }
+        
+        # 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)
+            
+            response_time_ms = int((time.time() - start_time) * 1000)
+            
+            if response.status_code != 200:
+                log_interaction(
+                    intent="translation",
+                    tenant_id=tenant_id,
+                    success=False,
+                    error=f"API returned status {response.status_code}",
+                    response_time_ms=response_time_ms,
+                    source_lang=src_lang,
+                    target_lang=tgt_lang,
+                    fallback_used=True
+                )
+                return {
+                    "translated_text": text,  # Fallback to original
+                    "source_lang": src_lang,
+                    "target_lang": tgt_lang,
+                    "original_text": text,
+                    "available": False,
+                    "error": f"Translation API error: {response.status_code}",
+                    "response_time_ms": response_time_ms
+                }
+            
+            results = response.json()
+        
+        # Validate results
+        if not results or not isinstance(results, list) or len(results) == 0:
+            log_interaction(
+                intent="translation",
+                tenant_id=tenant_id,
+                success=False,
+                error="Empty or invalid model output",
+                response_time_ms=response_time_ms,
+                source_lang=src_lang,
+                target_lang=tgt_lang
+            )
+            return {
+                "translated_text": text,  # Fallback to original
+                "source_lang": src_lang,
+                "target_lang": tgt_lang,
+                "original_text": text,
+                "available": True,
+                "error": "Translation returned unexpected format."
+            }
+        
+        # NLLB returns format: [{'translation_text': '...'}]
+        translated = results[0].get('translation_text', '').strip()
+        
+        if not translated:
+            log_interaction(
+                intent="translation",
+                tenant_id=tenant_id,
+                success=False,
+                error="Empty translation result",
+                response_time_ms=response_time_ms,
+                source_lang=src_lang,
+                target_lang=tgt_lang
+            )
+            return {
+                "translated_text": text,  # Fallback to original
+                "source_lang": src_lang,
+                "target_lang": tgt_lang,
+                "original_text": text,
+                "available": True,
+                "error": "Translation produced empty result."
+            }
+        
+        # Log slow translations
+        if response_time_ms > 5000:  # 5 seconds
+            log_interaction(
+                intent="translation_slow",
+                tenant_id=tenant_id,
+                success=True,
+                response_time_ms=response_time_ms,
+                details="Slow translation detected",
+                source_lang=src_lang,
+                target_lang=tgt_lang,
+                text_length=len(text)
+            )
+        
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=True,
+            response_time_ms=response_time_ms,
+            source_lang=src_lang,
+            target_lang=tgt_lang,
+            text_length=len(text)
+        )
+        
+        return {
+            "translated_text": translated,
+            "source_lang": src_lang,
+            "target_lang": tgt_lang,
+            "original_text": text,
+            "available": True,
+            "response_time_ms": response_time_ms
+        }
+
+    except httpx.TimeoutException:
+        response_time_ms = int((time.time() - start_time) * 1000)
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=False,
+            error="Translation request timed out",
+            response_time_ms=response_time_ms,
+            source_lang=src_lang,
+            target_lang=tgt_lang,
+            fallback_used=True
+        )
+        return {
+            "translated_text": text,  # Fallback to original
+            "source_lang": src_lang,
+            "target_lang": tgt_lang,
+            "original_text": text,
+            "available": False,
+            "error": "Translation request timed out.",
+            "response_time_ms": response_time_ms
+        }
+
+    except asyncio.CancelledError:
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=False,
+            error="Translation cancelled",
+            source_lang=src_lang,
+            target_lang=tgt_lang
+        )
+        raise
+        
+    except Exception as e:
+        response_time_ms = int((time.time() - start_time) * 1000)
+        
+        log_interaction(
+            intent="translation",
+            tenant_id=tenant_id,
+            success=False,
+            error=str(e),
+            response_time_ms=response_time_ms,
+            source_lang=src_lang,
+            target_lang=tgt_lang,
+            text_preview=sanitize_for_logging(text[:100]),
+            fallback_used=True
+        )
+        
+        return {
+            "translated_text": text,  # Fallback to original
+            "source_lang": src_lang,
+            "target_lang": tgt_lang,
+            "original_text": text,
+            "available": False,
+            "error": str(e),
+            "response_time_ms": response_time_ms
+        }
+
+
+async def detect_and_translate(
+    text: str,
+    target_language: str = "eng_Latn",
+    tenant_id: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Attempts to detect the source language and translate to target.
+    
+    Note: This is a simplified heuristic-based detection. For production,
+    consider integrating a dedicated language detection model.
+    
+    Args:
+        text: The text to translate
+        target_language: Target language code
+        tenant_id: Optional tenant identifier for logging
+        
+    Returns:
+        Translation result dictionary
+    """
+    if not text or not isinstance(text, str):
+        return {
+            "translated_text": "",
+            "detected_lang": "unknown",
+            "target_lang": target_language,
+            "original_text": text if isinstance(text, str) else "",
+            "available": True,
+            "error": "Invalid text input."
+        }
+    
+    # Simple heuristic: check for common non-English characters
+    detected_lang = "eng_Latn"  # Default assumption
+    
+    # Check for Spanish characters
+    if any(char in text for char in ['¿', '¡', 'ñ', 'á', 'é', 'í', 'ó', 'ú']):
+        detected_lang = "spa_Latn"
+    # Check for Chinese characters
+    elif any('\u4e00' <= char <= '\u9fff' for char in text):
+        detected_lang = "zho_Hans"
+    # Check for Arabic script
+    elif any('\u0600' <= char <= '\u06ff' for char in text):
+        detected_lang = "arb_Arab"
+    # Check for Cyrillic (Russian)
+    elif any('\u0400' <= char <= '\u04ff' for char in text):
+        detected_lang = "rus_Cyrl"
+    # Check for Devanagari (Hindi)
+    elif any('\u0900' <= char <= '\u097f' for char in text):
+        detected_lang = "hin_Deva"
+    
+    log_interaction(
+        intent="language_detection",
+        tenant_id=tenant_id,
+        success=True,
+        detected_lang=detected_lang,
+        text_preview=sanitize_for_logging(text[:50])
+    )
+    
+    result = await translate_text(text, detected_lang, target_language, tenant_id)
+    result["detected_lang"] = detected_lang
+    
+    return result
+
+
+async def batch_translate(
+    texts: List[str],
+    source_language: str = "eng_Latn",
+    target_language: str = "spa_Latn",
+    tenant_id: Optional[str] = None
+) -> List[Dict[str, Any]]:
+    """
+    Translate multiple texts at once.
+    
+    Args:
+        texts: List of strings to translate
+        source_language: Source language code
+        target_language: Target language code
+        tenant_id: Optional tenant identifier for logging
+        
+    Returns:
+        List of translation result dictionaries
+    """
+    if not texts or not isinstance(texts, list):
+        log_interaction(
+            intent="batch_translation",
+            tenant_id=tenant_id,
+            success=False,
+            error="Invalid texts input"
+        )
+        return []
+    
+    # Filter valid texts and limit batch size
+    valid_texts = [t for t in texts if isinstance(t, str) and t.strip()]
+    if len(valid_texts) > 50:  # Batch size limit
+        valid_texts = valid_texts[:50]
+        log_interaction(
+            intent="batch_translation",
+            tenant_id=tenant_id,
+            success=None,
+            details=f"Batch size limited to 50 texts"
+        )
+    
+    if not valid_texts:
+        log_interaction(
+            intent="batch_translation",
+            tenant_id=tenant_id,
+            success=False,
+            error="No valid texts in batch"
+        )
+        return []
+    
+    start_time = time.time()
+    results = []
+    
+    for text in valid_texts:
+        result = await translate_text(text, source_language, target_language, tenant_id)
+        results.append(result)
+    
+    response_time_ms = int((time.time() - start_time) * 1000)
+    
+    log_interaction(
+        intent="batch_translation",
+        tenant_id=tenant_id,
+        success=True,
+        response_time_ms=response_time_ms,
+        batch_size=len(valid_texts),
+        source_lang=normalize_language_code(source_language),
+        target_lang=normalize_language_code(target_language)
+    )
+    
+    return results
+
+
+def get_civic_phrase(
+    phrase_key: str,
+    language: str = "eng_Latn"
+) -> str:
+    """
+    Get a pre-translated civic phrase for common queries.
+    
+    Args:
+        phrase_key: Key for the civic phrase (e.g., "voting_location")
+        language: Target language code
+        
+    Returns:
+        Translated phrase or empty string if not found
+    """
+    if not phrase_key or not isinstance(phrase_key, str):
+        return ""
+    
+    lang_code = normalize_language_code(language)
+    phrase = CIVIC_PHRASES.get(lang_code, {}).get(phrase_key, "")
+    
+    if phrase:
+        log_interaction(
+            intent="civic_phrase_lookup",
+            success=True,
+            phrase_key=phrase_key,
+            language=lang_code
+        )
+    
+    return phrase