Upload intents.py

app/intents.py

@@ -0,0 +1,516 @@
+# app/intents.py
+"""
+🎯 Penny's Intent Classification System
+Rule-based intent classifier designed for civic engagement queries.
+
+CURRENT: Simple keyword matching (fast, predictable, debuggable)
+FUTURE: Will upgrade to ML/embedding-based classification (Gemma/LayoutLM)
+
+This approach allows Penny to understand resident needs and route them
+to the right civic systems — weather, resources, events, translation, etc.
+"""
+
+import logging
+from typing import Dict, List, Optional
+from dataclasses import dataclass, field
+from enum import Enum
+
+# --- LOGGING SETUP (Azure-friendly) ---
+logger = logging.getLogger(__name__)
+
+
+# --- INTENT CATEGORIES (Enumerated for type safety) ---
+class IntentType(str, Enum):
+    """
+    Penny's supported intent categories.
+    Each maps to a specific civic assistance pathway.
+    """
+    WEATHER = "weather"
+    GREETING = "greeting"
+    LOCAL_RESOURCES = "local_resources"
+    EVENTS = "events"
+    TRANSLATION = "translation"
+    SENTIMENT_ANALYSIS = "sentiment_analysis"
+    BIAS_DETECTION = "bias_detection"
+    DOCUMENT_PROCESSING = "document_processing"
+    HELP = "help"
+    EMERGENCY = "emergency"  # Critical safety routing
+    GOVERNMENT = "government"  # Government officials and representatives
+    UNKNOWN = "unknown"
+
+
+@dataclass
+class IntentMatch:
+    """
+    Structured intent classification result.
+    Includes confidence score and matched keywords for debugging.
+    """
+    intent: IntentType
+    confidence: float  # 0.0 - 1.0
+    matched_keywords: List[str]
+    is_compound: bool = False  # True if query spans multiple intents
+    secondary_intents: List[IntentType] = field(default_factory=list)
+    
+    def to_dict(self) -> Dict:
+        """Convert to dictionary for logging and API responses."""
+        return {
+            "intent": self.intent.value,
+            "confidence": self.confidence,
+            "matched_keywords": self.matched_keywords,
+            "is_compound": self.is_compound,
+            "secondary_intents": [intent.value for intent in self.secondary_intents]
+        }
+
+
+# --- INTENT KEYWORD PATTERNS (Organized by priority) ---
+class IntentPatterns:
+    """
+    Penny's keyword patterns for intent matching.
+    Organized by priority — critical intents checked first.
+    """
+    
+    # 🚨 PRIORITY 1: EMERGENCY & SAFETY (Always check first)
+    EMERGENCY = [
+        "911", "emergency", "urgent", "crisis", "danger", "help me",
+        "suicide", "overdose", "assault", "abuse", "threatening",
+        "hurt myself", "hurt someone", "life threatening"
+    ]
+    
+    # 🌍 PRIORITY 2: TRANSLATION (High civic value)
+    TRANSLATION = [
+        "translate", "in spanish", "in french", "in portuguese", 
+        "in german", "in chinese", "in arabic", "in vietnamese",
+        "in russian", "in korean", "in japanese", "in tagalog",
+        "convert to", "say this in", "how do i say", "what is", "in hindi"
+    ]
+    
+    # 📄 PRIORITY 3: DOCUMENT PROCESSING (Forms, PDFs)
+    DOCUMENT_PROCESSING = [
+        "process this document", "extract data", "analyze pdf",
+        "upload form", "read this file", "scan this", "form help",
+        "fill out", "document", "pdf", "application", "permit"
+    ]
+    
+    # 🔍 PRIORITY 4: ANALYSIS TOOLS
+    SENTIMENT_ANALYSIS = [
+        "how does this sound", "is this positive", "is this negative",
+        "analyze", "sentiment", "feel about", "mood", "tone"
+    ]
+    
+    BIAS_DETECTION = [
+        "is this biased", "check bias", "check fairness", "is this neutral",
+        "biased", "objective", "subjective", "fair", "discriminatory"
+    ]
+    
+    # 🌤️ PRIORITY 5: WEATHER + EVENTS (Compound intent handling)
+    WEATHER = [
+        "weather", "rain", "snow", "sunny", "forecast", "temperature",
+        "hot", "cold", "storm", "wind", "outside", "climate",
+        "degrees", "celsius", "fahrenheit"
+    ]
+    
+    # Specific date/time keywords that suggest event context
+    DATE_TIME = [
+        "today", "tomorrow", "this weekend", "next week",
+        "sunday", "monday", "tuesday", "wednesday", "thursday", "friday", "saturday",
+        "tonight", "this morning", "this afternoon", "this evening"
+    ]
+    
+    EVENTS = [
+        "event", "things to do", "what's happening", "activities",
+        "festival", "concert", "activity", "community event",
+        "show", "performance", "gathering", "meetup", "celebration"
+    ]
+    
+    # 🏛️ PRIORITY 6: GOVERNMENT & OFFICIALS (Before local resources)
+    GOVERNMENT = [
+        "city council", "council member", "council representative", "councilman", "councilwoman",
+        "mayor", "city manager", "commissioner", "representative", "representatives",
+        "who is my", "who represents me", "my representative", "my council",
+        "district", "ward", "elected official", "government official",
+        "city hall", "municipal", "alderman", "aldermanic", "supervisor"
+    ]
+    
+    # 🏛️ PRIORITY 7: LOCAL RESOURCES (Core civic mission)
+    LOCAL_RESOURCES = [
+        "resource", "shelter", "library", "help center",
+        "food bank", "warming center", "cooling center", "csb",
+        "mental health", "housing", "community service",
+        "trash", "recycling", "transit", "bus", "schedule",
+        "clinic", "hospital", "pharmacy", "assistance",
+        "utility", "water", "electric", "gas", "bill"
+    ]
+    
+    # 💬 PRIORITY 8: CONVERSATIONAL
+    GREETING = [
+        "hi", "hello", "hey", "what's up", "good morning",
+        "good afternoon", "good evening", "howdy", "yo",
+        "greetings", "sup", "hiya"
+    ]
+    
+    HELP = [
+        "help", "how do i", "can you help", "i need help",
+        "what can you do", "how does this work", "instructions",
+        "guide", "tutorial", "show me how"
+    ]
+
+
+def classify_intent(message: str) -> str:
+    """
+    🎯 Main classification function (backward-compatible).
+    Returns intent as string for existing API compatibility.
+    
+    Args:
+        message: User's query text
+        
+    Returns:
+        Intent string (e.g., "weather", "events", "translation")
+    """
+    try:
+        result = classify_intent_detailed(message)
+        return result.intent.value
+    except Exception as e:
+        logger.error(f"Intent classification failed: {e}", exc_info=True)
+        return IntentType.UNKNOWN.value
+
+
+def classify_intent_detailed(message: str) -> IntentMatch:
+    """
+    🧠 Enhanced classification with confidence scores and metadata.
+    
+    This function:
+    1. Checks for emergency keywords FIRST (safety routing)
+    2. Detects compound intents (e.g., "weather + events")
+    3. Returns structured result with confidence + matched keywords
+    
+    Args:
+        message: User's query text
+        
+    Returns:
+        IntentMatch object with full classification details
+    """
+    
+    if not message or not message.strip():
+        logger.warning("Empty message received for intent classification")
+        return IntentMatch(
+            intent=IntentType.UNKNOWN,
+            confidence=0.0,
+            matched_keywords=[]
+        )
+    
+    try:
+        text = message.lower().strip()
+        logger.debug(f"Classifying intent for: '{text[:50]}...'")
+        
+        # --- PRIORITY 1: EMERGENCY (Critical safety routing) ---
+        emergency_matches = _find_keyword_matches(text, IntentPatterns.EMERGENCY)
+        if emergency_matches:
+            logger.warning(f"🚨 EMERGENCY intent detected: {emergency_matches}")
+            return IntentMatch(
+                intent=IntentType.EMERGENCY,
+                confidence=1.0,  # Always high confidence for safety
+                matched_keywords=emergency_matches
+            )
+        
+        # --- PRIORITY 2: TRANSLATION ---
+        translation_matches = _find_keyword_matches(text, IntentPatterns.TRANSLATION)
+        if translation_matches:
+            return IntentMatch(
+                intent=IntentType.TRANSLATION,
+                confidence=0.9,
+                matched_keywords=translation_matches
+            )
+        
+        # --- PRIORITY 3: DOCUMENT PROCESSING ---
+        doc_matches = _find_keyword_matches(text, IntentPatterns.DOCUMENT_PROCESSING)
+        if doc_matches:
+            return IntentMatch(
+                intent=IntentType.DOCUMENT_PROCESSING,
+                confidence=0.9,
+                matched_keywords=doc_matches
+            )
+        
+        # --- PRIORITY 4: ANALYSIS TOOLS ---
+        sentiment_matches = _find_keyword_matches(text, IntentPatterns.SENTIMENT_ANALYSIS)
+        if sentiment_matches:
+            return IntentMatch(
+                intent=IntentType.SENTIMENT_ANALYSIS,
+                confidence=0.85,
+                matched_keywords=sentiment_matches
+            )
+        
+        bias_matches = _find_keyword_matches(text, IntentPatterns.BIAS_DETECTION)
+        if bias_matches:
+            return IntentMatch(
+                intent=IntentType.BIAS_DETECTION,
+                confidence=0.85,
+                matched_keywords=bias_matches
+            )
+        
+        # --- PRIORITY 5: GOVERNMENT & OFFICIALS (Check before resources) ---
+        government_matches = _find_keyword_matches(text, IntentPatterns.GOVERNMENT)
+        if government_matches:
+            return IntentMatch(
+                intent=IntentType.GOVERNMENT,
+                confidence=0.9,
+                matched_keywords=government_matches
+            )
+        
+        # --- PRIORITY 6: LOCAL RESOURCES (Check before events to avoid false matches) ---
+        resource_matches = _find_keyword_matches(text, IntentPatterns.LOCAL_RESOURCES)
+        
+        # --- PRIORITY 7: COMPOUND INTENT HANDLING (Weather + Events) ---
+        weather_matches = _find_keyword_matches(text, IntentPatterns.WEATHER)
+        event_matches = _find_keyword_matches(text, IntentPatterns.EVENTS)
+        date_matches = _find_keyword_matches(text, IntentPatterns.DATE_TIME)
+        
+        # If both resource and event keywords match, prioritize resources (more specific)
+        if resource_matches and event_matches:
+            # Check if resource keywords are more specific (e.g., "library" vs generic "show")
+            specific_resource_keywords = ["library", "libraries", "food bank", "shelter", "clinic", "hospital", "pharmacy", "trash", "recycling", "transit", "bus"]
+            has_specific_resource = any(kw in text for kw in specific_resource_keywords)
+            
+            if has_specific_resource:
+                return IntentMatch(
+                    intent=IntentType.LOCAL_RESOURCES,
+                    confidence=0.9,
+                    matched_keywords=resource_matches
+                )
+        
+        # Compound detection: "What events are happening this weekend?"
+        # or "What's the weather like for Sunday's festival?"
+        if event_matches and (weather_matches or date_matches):
+            logger.info("Compound intent detected: events + weather/date")
+            return IntentMatch(
+                intent=IntentType.EVENTS,  # Primary intent
+                confidence=0.85,
+                matched_keywords=event_matches + weather_matches + date_matches,
+                is_compound=True,
+                secondary_intents=[IntentType.WEATHER]
+            )
+        
+        # --- PRIORITY 8: SIMPLE WEATHER INTENT ---
+        if weather_matches:
+            return IntentMatch(
+                intent=IntentType.WEATHER,
+                confidence=0.9,
+                matched_keywords=weather_matches
+            )
+        
+        # --- PRIORITY 9: LOCAL RESOURCES (if not already handled) ---
+        if resource_matches:
+            return IntentMatch(
+                intent=IntentType.LOCAL_RESOURCES,
+                confidence=0.9,
+                matched_keywords=resource_matches
+            )
+        
+        # --- PRIORITY 10: EVENTS (Simple check) ---
+        if event_matches:
+            return IntentMatch(
+                intent=IntentType.EVENTS,
+                confidence=0.85,
+                matched_keywords=event_matches
+            )
+        
+        # --- PRIORITY 11: CONVERSATIONAL ---
+        greeting_matches = _find_keyword_matches(text, IntentPatterns.GREETING)
+        if greeting_matches:
+            return IntentMatch(
+                intent=IntentType.GREETING,
+                confidence=0.8,
+                matched_keywords=greeting_matches
+            )
+        
+        help_matches = _find_keyword_matches(text, IntentPatterns.HELP)
+        if help_matches:
+            return IntentMatch(
+                intent=IntentType.HELP,
+                confidence=0.9,
+                matched_keywords=help_matches
+            )
+        
+        # --- FALLBACK: UNKNOWN ---
+        logger.info(f"No clear intent match for: '{text[:50]}...'")
+        return IntentMatch(
+            intent=IntentType.UNKNOWN,
+            confidence=0.0,
+            matched_keywords=[]
+        )
+        
+    except Exception as e:
+        logger.error(f"Error during intent classification: {e}", exc_info=True)
+        return IntentMatch(
+            intent=IntentType.UNKNOWN,
+            confidence=0.0,
+            matched_keywords=[],
+        )
+
+
+# --- HELPER FUNCTIONS ---
+
+def _find_keyword_matches(text: str, keywords: List[str]) -> List[str]:
+    """
+    Finds which keywords from a pattern list appear in the user's message.
+    
+    Args:
+        text: Normalized user message (lowercase)
+        keywords: List of keywords to search for
+        
+    Returns:
+        List of matched keywords (for debugging/logging)
+    """
+    try:
+        matches = []
+        for keyword in keywords:
+            if keyword in text:
+                matches.append(keyword)
+        return matches
+    except Exception as e:
+        logger.error(f"Error finding keyword matches: {e}", exc_info=True)
+        return []
+
+
+def get_intent_description(intent: IntentType) -> str:
+    """
+    🗣️ Penny's plain-English explanation of what each intent does.
+    Useful for help systems and debugging.
+    
+    Args:
+        intent: IntentType enum value
+        
+    Returns:
+        Human-readable description of the intent
+    """
+    descriptions = {
+        IntentType.WEATHER: "Get current weather conditions and forecasts for your area",
+        IntentType.GREETING: "Start a conversation with Penny",
+        IntentType.LOCAL_RESOURCES: "Find community resources like shelters, libraries, and services",
+        IntentType.EVENTS: "Discover local events and activities happening in your city",
+        IntentType.TRANSLATION: "Translate text between 27 languages",
+        IntentType.SENTIMENT_ANALYSIS: "Analyze the emotional tone of text",
+        IntentType.BIAS_DETECTION: "Check text for potential bias or fairness issues",
+        IntentType.DOCUMENT_PROCESSING: "Process PDFs and forms to extract information",
+        IntentType.HELP: "Learn how to use Penny's features",
+        IntentType.EMERGENCY: "Connect with emergency services and crisis support",
+        IntentType.GOVERNMENT: "Find information about city officials, council members, and representatives",
+        IntentType.UNKNOWN: "I'm not sure what you're asking — can you rephrase?"
+    }
+    return descriptions.get(intent, "Unknown intent type")
+
+
+def get_all_supported_intents() -> Dict[str, str]:
+    """
+    📋 Returns all supported intents with descriptions.
+    Useful for /help endpoints and documentation.
+    
+    Returns:
+        Dictionary mapping intent values to descriptions
+    """
+    try:
+        return {
+            intent.value: get_intent_description(intent)
+            for intent in IntentType
+            if intent != IntentType.UNKNOWN
+        }
+    except Exception as e:
+        logger.error(f"Error getting supported intents: {e}", exc_info=True)
+        return {}
+
+
+# --- FUTURE ML UPGRADE HOOK ---
+def classify_intent_ml(message: str, use_embedding_model: bool = False) -> IntentMatch:
+    """
+    🔮 PLACEHOLDER for future ML-based classification.
+    
+    When ready to upgrade from keyword matching to embeddings:
+    1. Load Gemma-7B or sentence-transformers model
+    2. Generate message embeddings
+    3. Compare to intent prototype embeddings
+    4. Return top match with confidence score
+    
+    Args:
+        message: User's query
+        use_embedding_model: If True, use ML model (not implemented yet)
+        
+    Returns:
+        IntentMatch object (currently falls back to rule-based)
+    """
+    
+    if use_embedding_model:
+        logger.warning("ML-based classification not yet implemented. Falling back to rules.")
+    
+    # Fallback to rule-based for now
+    return classify_intent_detailed(message)
+
+
+# --- TESTING & VALIDATION ---
+def validate_intent_patterns() -> Dict[str, List[str]]:
+    """
+    🧪 Validates that all intent patterns are properly configured.
+    Returns any overlapping keywords that might cause conflicts.
+    
+    Returns:
+        Dictionary of overlapping keywords between intent pairs
+    """
+    try:
+        all_patterns = {
+            "emergency": IntentPatterns.EMERGENCY,
+            "translation": IntentPatterns.TRANSLATION,
+            "document": IntentPatterns.DOCUMENT_PROCESSING,
+            "sentiment": IntentPatterns.SENTIMENT_ANALYSIS,
+            "bias": IntentPatterns.BIAS_DETECTION,
+            "weather": IntentPatterns.WEATHER,
+            "events": IntentPatterns.EVENTS,
+            "resources": IntentPatterns.LOCAL_RESOURCES,
+            "greeting": IntentPatterns.GREETING,
+            "help": IntentPatterns.HELP
+        }
+        
+        overlaps = {}
+        
+        # Check for keyword overlap between different intents
+        for intent1, keywords1 in all_patterns.items():
+            for intent2, keywords2 in all_patterns.items():
+                if intent1 >= intent2:  # Avoid duplicate comparisons
+                    continue
+                
+                overlap = set(keywords1) & set(keywords2)
+                if overlap:
+                    key = f"{intent1}_vs_{intent2}"
+                    overlaps[key] = list(overlap)
+        
+        if overlaps:
+            logger.warning(f"Found keyword overlaps between intents: {overlaps}")
+        
+        return overlaps
+        
+    except Exception as e:
+        logger.error(f"Error validating intent patterns: {e}", exc_info=True)
+        return {}
+
+
+# --- LOGGING SAMPLE CLASSIFICATIONS (For monitoring) ---
+def log_intent_classification(message: str, result: IntentMatch) -> None:
+    """
+    📊 Logs classification results for Azure Application Insights.
+    Helps track intent distribution and confidence patterns.
+    
+    Args:
+        message: Original user message (truncated for PII safety)
+        result: IntentMatch classification result
+    """
+    try:
+        # Truncate message for PII safety
+        safe_message = message[:50] + "..." if len(message) > 50 else message
+        
+        logger.info(
+            f"Intent classified | "
+            f"intent={result.intent.value} | "
+            f"confidence={result.confidence:.2f} | "
+            f"compound={result.is_compound} | "
+            f"keywords={result.matched_keywords[:5]} | "  # Limit logged keywords
+            f"message_preview='{safe_message}'"
+        )
+    except Exception as e:
+        logger.error(f"Error logging intent classification: {e}", exc_info=True)