|
import time |
|
import re |
|
from typing import Dict, Any, List, Optional |
|
from datetime import datetime |
|
from utils.config import config |
|
|
|
class SmartContextDetector: |
|
"""Intelligently detects when context is relevant to user queries""" |
|
|
|
def __init__(self): |
|
|
|
self.time_indicators = [ |
|
'time', 'date', 'today', 'now', 'current', 'moment', |
|
'morning', 'afternoon', 'evening', 'night', 'weekend', |
|
'what time', 'what day', 'what year', 'what month' |
|
] |
|
|
|
self.weather_indicators = [ |
|
'weather', 'temperature', 'rain', 'snow', 'sunny', 'cloudy', |
|
'forecast', 'climate', 'season', 'hot', 'cold', 'warm', 'cool', |
|
'umbrella', 'jacket', 'outdoor', 'outside' |
|
] |
|
|
|
self.location_indicators = [ |
|
'in', 'at', 'near', 'around', 'local', 'here', 'there' |
|
] |
|
|
|
def should_include_time_context(self, user_query: str, |
|
conversation_history: List[Dict] = None) -> bool: |
|
"""Determine if time context is relevant to the query""" |
|
query_lower = user_query.lower() |
|
|
|
|
|
if any(indicator in query_lower for indicator in self.time_indicators): |
|
return True |
|
|
|
|
|
if conversation_history: |
|
recent_messages = conversation_history[-3:] |
|
context_text = " ".join([msg.get('content', '') for msg in recent_messages]).lower() |
|
if any(word in context_text for word in self.time_indicators): |
|
return True |
|
|
|
return False |
|
|
|
def should_include_weather_context(self, user_query: str, |
|
conversation_history: List[Dict] = None) -> bool: |
|
"""Determine if weather context is relevant to the query""" |
|
query_lower = user_query.lower() |
|
|
|
|
|
if any(indicator in query_lower for indicator in self.weather_indicators): |
|
return True |
|
|
|
|
|
location_mentioned = any(loc.lower() in query_lower for loc in |
|
['new york', 'london', 'tokyo', 'paris', 'los angeles', 'sydney', 'singapore', 'mumbai']) |
|
activity_mentioned = any(activity in query_lower for activity in |
|
['outdoor', 'outside', 'walk', 'run', 'travel', 'trip', 'plans', 'going']) |
|
|
|
if location_mentioned and activity_mentioned: |
|
return True |
|
|
|
|
|
if conversation_history: |
|
recent_messages = conversation_history[-3:] |
|
context_text = " ".join([msg.get('content', '') for msg in recent_messages]).lower() |
|
if any(word in context_text for word in self.weather_indicators): |
|
return True |
|
|
|
return False |
|
|
|
def extract_location_if_relevant(self, user_query: str) -> Optional[str]: |
|
"""Extract location if query suggests location-specific context""" |
|
|
|
locations = { |
|
'new york': ['new york', 'ny', 'nyc'], |
|
'london': ['london', 'uk', 'england'], |
|
'tokyo': ['tokyo', 'japan', 'jp'], |
|
'paris': ['paris', 'france', 'fr'], |
|
'los angeles': ['los angeles', 'la', 'california'], |
|
'sydney': ['sydney', 'australia', 'au'], |
|
'singapore': ['singapore', 'sg'], |
|
'mumbai': ['mumbai', 'india', 'in'] |
|
} |
|
|
|
query_lower = user_query.lower() |
|
for location, aliases in locations.items(): |
|
if any(alias in query_lower for alias in aliases): |
|
return location |
|
|
|
return None |
|
|
|
def get_relevant_context(self, user_query: str, |
|
conversation_history: List[Dict] = None) -> Dict[str, Any]: |
|
"""Get only the context that's actually relevant""" |
|
context = { |
|
'include_time': self.should_include_time_context(user_query, conversation_history), |
|
'include_weather': self.should_include_weather_context(user_query, conversation_history), |
|
'detected_location': self.extract_location_if_relevant(user_query), |
|
'timestamp': time.time() |
|
} |
|
|
|
|
|
if context['include_time']: |
|
now = datetime.now() |
|
context['time_data'] = { |
|
'current_time': now.strftime("%A, %B %d, %Y at %I:%M %p"), |
|
'day_of_week': now.strftime("%A"), |
|
'is_business_hours': 9 <= now.hour <= 17 |
|
} |
|
|
|
return context |
|
|
|
|
|
smart_context = SmartContextDetector() |
|
|