Spaces:
Paused
Paused
| # 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" | |
| 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) |