app.py

import os
import gradio as gr
from dotenv import load_dotenv
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.tools import tool
from langchain.pydantic_v1 import BaseModel, Field
import requests
from datetime import datetime
from typing import List
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import PydanticOutputParser
from langchain.memory import ConversationBufferMemory
from langchain.agents import AgentExecutor, create_tool_calling_agent

load_dotenv(dotenv_path='api.env.txt')
Langchain_API_KEY = os.getenv('LANGCHAIN_API')
GOOGLE_API_KEY = os.getenv('GOOGLE_API')
WEATHER_API_KEY = os.getenv('WEATHER_API')

os.environ["GOOGLE_API_KEY"] = GOOGLE_API_KEY

llm = ChatGoogleGenerativeAI(
    model="gemini-1.5-flash",
    temperature=0,
    max_tokens=None,
    timeout=None,
    max_retries=2,
)



class WeatherInput(BaseModel):
    city: str = Field(default=None, description="The city to get the weather for.")


def get_location_from_ip():
    ip = requests.get('https://api.ipify.org').text
    response = requests.get(f"https://ipapi.co/{ip}/json/").json()
    return {
        'city': response.get('city'),
        'latitude': response.get('latitude'),
        'longitude': response.get('longitude')
    }


@tool("get_weather_by_location", args_schema=WeatherInput, return_direct=True)
def get_weather_by_location(city: str = None):
    if not city:
        location = get_location_from_ip()
        city = location['city']

    url = f"https://api.tomorrow.io/v4/timelines?apikey={WEATHER_API_KEY}"
    payload = {
        "location": city,
        "fields": ["temperature", "humidity", "windSpeed"],
        "units": "metric",
        "timesteps": ["1d"],
        "startTime": "now",
        "endTime": "nowPlus5d",
        "timezone": "auto"
    }
    headers = {
        "accept": "application/json",
        "content-type": "application/json"
    }

    response = requests.post(url, json=payload, headers=headers).json()

    return format_weather_response(response, city)


def format_weather_response(weather_data, city):
    intervals = weather_data['data']['timelines'][0]['intervals']
    response = f"Weather forecast for {city}:\n\n"

    for interval in intervals:
        date = datetime.fromisoformat(interval['startTime']).strftime("%A, %B %d")
        temp = round(interval['values']['temperature'], 1)
        humidity = round(interval['values']['humidity'], 1)
        wind_speed = round(interval['values']['windSpeed'], 1)

        response += f"{date}:\n"
        response += f"  Temperature: {temp}°C\n"
        response += f"  Humidity: {humidity}%\n"
        response += f"  Wind Speed: {wind_speed} km/h\n\n"

    return response

class DailyWeather(BaseModel):
    date: str
    temperature: float
    condition: str
    humidity: float
    wind_speed: float
    advice: str

class WeatherOutput(BaseModel):
    location: str = Field(description="The location or the city for which the weather is reported")
    forecast: List[DailyWeather] = Field(description="The weather forecast for multiple days")

parser = PydanticOutputParser(pydantic_object=WeatherOutput)

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a helpful weather assistant. Your primary function is to provide weather information for cities around the world and offer advice based on the weather conditions. Here are your key responsibilities:

1. If a user asks about the weather in a specific city, use the get_weather_by_location tool to fetch and provide that information for today and the next few days.
2. If a user asks about the weather without specifying a city (e.g., "tell me the weather in my city" or "what is the weather in our city/town"), assume they're asking about their current location. Use the get_weather_by_location tool with an empty string as input to get this information.
3. After getting the weather data, always use the format_weather tool to present the information in a user-friendly format and include advice for each day.
4. Based on the weather conditions, provide relevant advice to the user for each day. For example:
   - If it's sunny, suggest outdoor activities or remind them to use sunscreen.
   - If it's rainy, advise them to bring an umbrella or suggest indoor activities.
   - If it's very cold or hot, give appropriate clothing or safety recommendations.
5. If you're unsure about the location or need more information, politely ask the user for clarification.
6. Be prepared to answer follow-up questions about the weather for the rest of the week or for a specific day.

Remember to be friendly and informative in your responses, and focus on providing a full weather forecast when asked. Use the conversation history to provide context-aware responses and avoid repeating information."""),
    ("human", "{input}"),
    ("ai", "Hello! I'd be happy to help you with the weather information for the next few days and provide some helpful advice. What would you like to know?"),
    ("human", "{input}"),
    ("ai", "I understand. Let me fetch that weather information for you and offer some advice based on the conditions."),
    ("placeholder", "{agent_scratchpad}"),
])


# Initialize tools and agent
tools = [get_weather_by_location]

memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

agent = create_tool_calling_agent(llm, tools, prompt=prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    output_parser=PydanticOutputParser(pydantic_object=WeatherOutput)
)


def gradio_interface(user_input):
    result = agent_executor.invoke({"input": user_input})
    return result['output']


# Gradio UI
with gr.Blocks() as demo:
    gr.Markdown("# Weather Assistant")
    chatbot = gr.Chatbot()
    with gr.Row():
        txt = gr.Textbox(show_label=False, placeholder="Ask about the weather...").style(container=False)
        txt.submit(gradio_interface, txt, chatbot)

demo.launch()