|
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""" |
|
|
|
context_info = smart_context.get_relevant_context(user_query, conversation_history) |
|
|
|
context_parts = [] |
|
|
|
|
|
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']}") |
|
|
|
|
|
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) |
|
|
|
|
|
if context_parts: |
|
return f"[Context: {', '.join(context_parts)}]" |
|
|
|
return None |
|
|
|
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 |
|
|
|
|
|
context_provider = ContextProvider() |
|
|