Penny_V2 / app /intents.py
pythonprincess's picture
Upload 9 files
22eeb7e verified
# 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
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: 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 7: 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: 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)
# 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 6: SIMPLE WEATHER INTENT ---
if weather_matches:
return IntentMatch(
intent=IntentType.WEATHER,
confidence=0.9,
matched_keywords=weather_matches
)
# --- PRIORITY 7: LOCAL RESOURCES ---
resource_matches = _find_keyword_matches(text, IntentPatterns.LOCAL_RESOURCES)
if resource_matches:
return IntentMatch(
intent=IntentType.LOCAL_RESOURCES,
confidence=0.9,
matched_keywords=resource_matches
)
# --- PRIORITY 8: EVENTS (Simple check) ---
if event_matches:
return IntentMatch(
intent=IntentType.EVENTS,
confidence=0.85,
matched_keywords=event_matches
)
# --- PRIORITY 9: 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.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)