# 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)