Spaces:
Starting
Starting
File size: 2,086 Bytes
751d628 4701375 751d628 4701375 751d628 4701375 751d628 4701375 751d628 4701375 |
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 |
import aiohttp
import ssl
import logging
import os
from langchain_core.tools import tool
from tenacity import retry, stop_after_attempt, wait_exponential
from dotenv import load_dotenv
logger = logging.getLogger(__name__)
load_dotenv()
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
async def fetch_weather(location: str, api_key: str) -> dict:
url = f"http://api.openweathermap.org/data/2.5/weather?q={location}&appid={api_key}&units=metric"
ssl_context = ssl.create_default_context()
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, ssl=ssl_context) as response:
response.raise_for_status()
return await response.json()
except aiohttp.ClientError as e:
logger.error(f"Failed to fetch weather for {location}: {e}")
raise
@tool
async def weather_info_tool(location: str, query_type: str = "current") -> str:
"""
Fetch weather information for a given location.
Args:
location (str): City or location name.
query_type (str): Type of weather query ('current', 'forecast'; default: 'current').
Returns:
str: Weather information or error message.
"""
try:
api_key = os.getenv("OPENWEATHERMAP_API_KEY")
if not api_key:
logger.error("OPENWEATHERMAP_API_KEY not set")
return "Weather unavailable: API key missing"
if query_type != "current":
logger.warning(f"Query type '{query_type}' not supported; using current weather")
query_type = "current"
response = await fetch_weather(location, api_key)
if response.get("cod") == 200:
condition = response["weather"][0]["description"]
temp = response["main"]["temp"]
return f"Weather in {location}: {condition}, {temp}°C"
return f"Unable to fetch weather for {location}."
except Exception as e:
logger.error(f"Error fetching weather for {location}: {e}")
return f"Error: {str(e)}" |