File size: 2,346 Bytes
e441606 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
import requests
from typing import Dict, Any, Optional
from utils.config import config
from src.services.smart_context import smart_context
class ContextProvider:
"""Provides context data only when relevant"""
def __init__(self):
self.openweather_api_key = getattr(config, 'openweather_api_key', None)
def get_context_for_llm(self, user_query: str,
conversation_history: list = None) -> Optional[str]:
"""Generate context string only when relevant for LLM consumption"""
# Get smart context detection
context_info = smart_context.get_relevant_context(user_query, conversation_history)
context_parts = []
# Add time context if relevant
if context_info['include_time'] and 'time_data' in context_info:
time_data = context_info['time_data']
context_parts.append(f"Current time: {time_data['current_time']}")
# Add weather context if relevant
if context_info['include_weather']:
weather_data = self._get_weather_data(context_info['detected_location'] or 'New York')
if weather_data:
context_parts.append(weather_data)
# Only return context if there's something relevant
if context_parts:
return f"[Context: {', '.join(context_parts)}]"
return None # No context needed
def _get_weather_data(self, location: str) -> Optional[str]:
"""Get weather data for a specific location"""
if not self.openweather_api_key:
return None
try:
url = "http://api.openweathermap.org/data/2.5/weather"
params = {
'q': location,
'appid': self.openweather_api_key,
'units': 'metric'
}
response = requests.get(url, params=params, timeout=5)
if response.status_code == 200:
data = response.json()
return (f"Weather in {data['name']}: {data['weather'][0]['description']}, "
f"{data['main']['temp']}°C, humidity {data['main']['humidity']}%")
except Exception:
pass
return None
# Global instance
context_provider = ContextProvider()
|