Spaces:
Sleeping
Sleeping
| import os | |
| from smolagents import CodeAgent, HfApiModel, tool | |
| import datetime | |
| import pytz | |
| import gradio as gr | |
| from geopy.geocoders import Nominatim | |
| from timezonefinder import TimezoneFinder | |
| import requests | |
| from langdetect import detect | |
| def get_timezone_by_city(city: str) -> str: | |
| """Fetches the timezone for a given city using geolocation. | |
| Args: | |
| city: A string representing the city name, optionally with region or country (e.g., 'Бураево', 'Sydney'). | |
| Returns: | |
| A string with the timezone (e.g., 'Europe/Moscow') or an error message. | |
| """ | |
| if not isinstance(city, str): | |
| return f"Error: Expected a string for city, got: {type(city)}" | |
| try: | |
| geolocator = Nominatim(user_agent="smolagents_bot") | |
| location = geolocator.geocode(city, language="en") | |
| if not location: | |
| city_name = city.split(",")[0].strip().title() | |
| location = geolocator.geocode(city_name, language="en") | |
| if not location: | |
| return f"Error: City '{city}' not found. Try a different spelling or provide more context." | |
| tf = TimezoneFinder() | |
| timezone = tf.timezone_at(lat=location.latitude, lng=location.longitude) | |
| if not timezone: | |
| return f"Error: Timezone not found for city '{city}'." | |
| return timezone | |
| except Exception as e: | |
| return f"Error fetching timezone for city '{city}': {str(e)}" | |
| def get_current_time_in_timezone(timezone: str) -> str: | |
| """Fetches the current local time in a specified timezone. | |
| Args: | |
| timezone: A string representing a valid timezone (e.g., 'Europe/Moscow'). | |
| Returns: | |
| A string with the current time (e.g., '2025-05-03 12:00:00') or an error message. | |
| """ | |
| if not isinstance(timezone, str): | |
| return f"Error: Expected a string for timezone, got: {type(timezone)}" | |
| try: | |
| tz = pytz.timezone(timezone) | |
| local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S") | |
| return local_time | |
| except Exception as e: | |
| return f"Error fetching time for timezone '{timezone}': {str(e)}" | |
| def get_air_quality(city: str, lang: str = "en") -> str: | |
| """Fetches air quality data for a given city using OpenWeatherMap API. | |
| Args: | |
| city: A string representing the city name (e.g., 'Бирск', 'Sydney'). | |
| lang: A string representing the language code (e.g., 'en', 'ru'). | |
| Returns: | |
| A string with air quality index (AQI) and pollutant levels or an error message. | |
| """ | |
| if not isinstance(city, str): | |
| return f"Error: Expected a string for city, got: {type(city)}" | |
| try: | |
| geolocator = Nominatim(user_agent="smolagents_bot") | |
| location = geolocator.geocode(city, language="en") | |
| if not location: | |
| city_name = city.split(",")[0].strip().title() | |
| location = geolocator.geocode(city_name, language="en") | |
| if not location: | |
| return f"Error: City '{city}' not found for air quality data." | |
| api_key = os.getenv("OPENWEATHERMAP_API_KEY", "93f518fda8d24cf89aee7050c26b27e9") | |
| url = f"http://api.openweathermap.org/data/2.5/air_pollution?lat={location.latitude}&lon={location.longitude}&appid={api_key}" | |
| response = requests.get(url) | |
| if response.status_code != 200: | |
| return f"Error fetching air quality for '{city}': API returned status {response.status_code}" | |
| data = response.json() | |
| aqi = data["list"][0]["main"]["aqi"] | |
| components = data["list"][0]["components"] | |
| translations = { | |
| "en": { | |
| "aqi_desc": {1: "Good", 2: "Fair", 3: "Moderate", 4: "Poor", 5: "Very Poor"}, | |
| "air_quality": "Air quality", | |
| "pm2_5": "Fine particles (PM2.5)", | |
| "pm10": "Coarse particles (PM10)", | |
| "co": "Carbon monoxide (CO)" | |
| }, | |
| "ru": { | |
| "aqi_desc": {1: "Хорошее", 2: "Удовлетворительное", 3: "Среднее", 4: "Плохое", 5: "Очень плохое"}, | |
| "air_quality": "Качество воздуха", | |
| "pm2_5": "Мелкие частицы (PM2.5)", | |
| "pm10": "Крупные частицы (PM10)", | |
| "co": "Угарный газ (CO)" | |
| } | |
| } | |
| lang = lang if lang in translations else "en" | |
| t = translations[lang] | |
| aqi_emoji = "🌱" if aqi <= 2 else "😷" if aqi <= 4 else "☣️" | |
| return ( | |
| f"😷 {t['air_quality']}: AQI {aqi} ({t['aqi_desc'].get(aqi, 'Unknown')}) {aqi_emoji}\n" | |
| f"{t['pm2_5']}: {components['pm2_5']} µg/m³\n" | |
| f"{t['pm10']}: {components['pm10']} µg/m³\n" | |
| f"{t['co']}: {components['co']} µg/m³" | |
| ) | |
| except Exception as e: | |
| return f"Error fetching air quality for '{city}': {str(e)}" | |
| def get_weather(city: str, lang: str = "en") -> str: | |
| """Fetches weather data for a given city using OpenWeatherMap API. | |
| Args: | |
| city: A string representing the city name (e.g., 'Бирск', 'Sydney'). | |
| lang: A string representing the language code (e.g., 'en', 'ru'). | |
| Returns: | |
| A string with weather conditions, temperature, and humidity or an error message. | |
| """ | |
| if not isinstance(city, str): | |
| return f"Error: Expected a string for city, got: {type(city)}" | |
| try: | |
| geolocator = Nominatim(user_agent="smolagents_bot") | |
| location = geolocator.geocode(city, language="en") | |
| if not location: | |
| city_name = city.split(",")[0].strip().title() | |
| location = geolocator.geocode(city_name, language="en") | |
| if not location: | |
| return f"Error: City '{city}' not found for weather data." | |
| api_key = os.getenv("OPENWEATHERMAP_API_KEY", "93f518fda8d24cf89aee7050c26b27e9") | |
| url = f"http://api.openweathermap.org/data/2.5/weather?lat={location.latitude}&lon={location.longitude}&appid={api_key}&units=metric" | |
| response = requests.get(url) | |
| if response.status_code != 200: | |
| return f"Error fetching weather for '{city}': API returned status {response.status_code}" | |
| data = response.json() | |
| temp = data["main"]["temp"] | |
| weather = data["weather"][0]["main"] | |
| humidity = data["main"]["humidity"] | |
| translations = { | |
| "en": { | |
| "weather": "Weather", | |
| "temp": "Temperature", | |
| "humidity": "Humidity", | |
| "conditions": { | |
| "Clear": "Clear ☀️", | |
| "Clouds": "Cloudy ☁️", | |
| "Rain": "Rain 🌧", | |
| "Snow": "Snow ❄️", | |
| "Thunderstorm": "Thunderstorm ⛈", | |
| "Drizzle": "Drizzle 🌦", | |
| "Mist": "Mist 🌫" | |
| } | |
| }, | |
| "ru": { | |
| "weather": "Погода", | |
| "temp": "Температура", | |
| "humidity": "Влажность", | |
| "conditions": { | |
| "Clear": "Ясно ☀️", | |
| "Clouds": "Облачно ☁️", | |
| "Rain": "Дождь 🌧", | |
| "Snow": "Снег ❄️", | |
| "Thunderstorm": "Гроза ⛈", | |
| "Drizzle": "Морось 🌦", | |
| "Mist": "Туман 🌫" | |
| } | |
| } | |
| } | |
| lang = lang if lang in translations else "en" | |
| t = translations[lang] | |
| condition = t["conditions"].get(weather, weather) | |
| return ( | |
| f"🌡️ {t['weather']}: {condition}\n" | |
| f"{t['temp']}: {temp}°C\n" | |
| f"{t['humidity']}: {humidity}%" | |
| ) | |
| except Exception as e: | |
| return f"Error fetching weather for '{city}': {str(e)}" | |
| def final_answer(answer: str) -> str: | |
| """Returns the final answer. | |
| Args: | |
| answer: A string containing the final answer. | |
| Returns: | |
| The input answer string. | |
| """ | |
| return answer | |
| try: | |
| model = HfApiModel( | |
| max_tokens=1000, | |
| temperature=0.5, | |
| model_id='mistralai/Mixtral-8x7B-Instruct-v0.1', | |
| ) | |
| except Exception as e: | |
| print(f"Error initializing model: {str(e)}") | |
| prompt_templates = { | |
| "system_prompt": ( | |
| "You are a helpful assistant that responds in the same language as the user's query. " | |
| "If the query contains a city name (e.g., 'Бураево', 'Sydney', 'न्यू यॉर्क') or phrases like 'сколько времени', 'what time', " | |
| "or a city in any context (e.g., 'Что там в Бирске?'), return current time, air quality, and weather for that city. " | |
| "Use ONLY the provided functions: `get_timezone_by_city`, `get_current_time_in_timezone`, `get_air_quality`, `get_weather`, `final_answer`. " | |
| "Detect query language with `detect` and pass it to `get_air_quality` and `get_weather` for translated output. " | |
| "Return ONLY the result via `final_answer()` in a code block:\n" | |
| "```py\n" | |
| "final_answer('YOUR ANSWER HERE')\n" | |
| "```\n" | |
| "Do NOT include 'thoughts', 'code', explanations, extra imports, or invalid dates. " | |
| "Do NOT define new functions or modify existing ones. " | |
| "If the city is not found or query is unclear, return an error via `final_answer()`. " | |
| "NEVER generate code without ```py``` block or with invalid syntax." | |
| ), | |
| "default": "Response: {{question}}", | |
| "planning": { | |
| "initial_plan": ( | |
| "Analyze query: {{question}}. If it contains a city, get time, air quality, and weather using provided functions. " | |
| "Return code:\n" | |
| "```py\n" | |
| "final_answer('YOUR ANSWER HERE')\n" | |
| "```" | |
| ), | |
| "update_plan_pre_messages": ( | |
| "Review query: {{question}}. Adjust plan. Return code:\n" | |
| "```py\n" | |
| "final_answer('YOUR ANSWER HERE')\n" | |
| "```" | |
| ), | |
| "update_plan_post_messages": ( | |
| "Review query: {{question}} and results. Adjust plan. Return code:\n" | |
| "```py\n" | |
| "final_answer('YOUR ANSWER HERE')\n" | |
| "```" | |
| ), | |
| }, | |
| "managed_agent": { | |
| "execute": ( | |
| "Execute task: {{question}}. If it contains a city or phrases like 'сколько времени', use provided functions. " | |
| "Return code:\n" | |
| "```py\n" | |
| "from langdetect import detect\n" | |
| "query = '{{question}}'.strip()\n" | |
| "city = query.split(' в ')[-1].strip().title() if ' в ' in query else query.title()\n" | |
| "lang = detect(query) if query else 'en'\n" | |
| "timezone = get_timezone_by_city(city)\n" | |
| "if timezone.startswith('Error'):\n" | |
| " final_answer(timezone)\n" | |
| "else:\n" | |
| " time_result = get_current_time_in_timezone(timezone)\n" | |
| " air_result = get_air_quality(city, lang)\n" | |
| " weather_result = get_weather(city, lang)\n" | |
| " final_answer(f'🕒 Время в {city}: {time_result}\\n' + air_result + '\\n' + weather_result)\n" | |
| "```" | |
| ), | |
| "report": ( | |
| "Summarize task: {{question}}. Return code:\n" | |
| "```py\n" | |
| "final_answer('YOUR ANSWER HERE')\n" | |
| "```" | |
| ), | |
| "task": ( | |
| "Define task: {{question}}. Return code:\n" | |
| "```py\n" | |
| "final_answer('YOUR TASK DEFINITION HERE')\n" | |
| "```" | |
| ), | |
| }, | |
| "final_answer": { | |
| "pre_messages": ( | |
| "Prepare response for: {{question}}. Return code:\n" | |
| "```py\n" | |
| "final_answer('YOUR ANSWER HERE')\n" | |
| "```" | |
| ), | |
| "template": "Final response: {{answer}}", | |
| "post_messages": ( | |
| "Review response for: {{question}}. Return code:\n" | |
| "```py\n" | |
| "final_answer('YOUR ANSWER HERE')\n" | |
| "```" | |
| ), | |
| }, | |
| } | |
| agent = CodeAgent( | |
| model=model, | |
| tools=[final_answer, get_current_time_in_timezone, get_timezone_by_city, get_air_quality, get_weather], | |
| max_steps=3, | |
| verbosity_level=2, | |
| prompt_templates=prompt_templates, | |
| ) | |
| def process_input(user_input): | |
| try: | |
| response = agent.run(user_input) | |
| return response if response else "Error: No response received from the agent." | |
| except Exception as e: | |
| return f"Error: {str(e)}" | |
| gr.Interface( | |
| fn=process_input, | |
| inputs="text", | |
| outputs="text", | |
| title="Helpful Assistant", | |
| description="Hello! What city or town are you from? I can tell you the time, air quality, weather, and more!" | |
| ).launch() |