diff --git a/ai_agent_tutorials/ai_aqi_analysis_agent/README.md b/ai_agent_tutorials/ai_aqi_analysis_agent/README.md new file mode 100644 index 0000000..cae38eb --- /dev/null +++ b/ai_agent_tutorials/ai_aqi_analysis_agent/README.md @@ -0,0 +1,81 @@ +# 🌍 AQI Analysis Agent + +The AQI Analysis Agent is a powerful air quality monitoring and health recommendation tool powered by Firecrawl and Agno's AI Agent framework. This app helps users make informed decisions about outdoor activities by analyzing real-time air quality data and providing personalized health recommendations. + +## Features + +- **Multi-Agent System** + - **AQI Analyzer**: Fetches and processes real-time air quality data + - **Health Recommendation Agent**: Generates personalized health advice + +- **Air Quality Metrics**: + - Overall Air Quality Index (AQI) + - Particulate Matter (PM2.5 and PM10) + - Carbon Monoxide (CO) levels + - Temperature + - Humidity + - Wind Speed + +- **Comprehensive Analysis**: + - Real-time data visualization + - Health impact assessment + - Activity safety recommendations + - Best time suggestions for outdoor activities + - Weather condition correlations + +- **Interactive Features**: + - Location-based analysis + - Medical condition considerations + - Activity-specific recommendations + - Downloadable reports + - Example queries for quick testing + +## How to Run + +Follow these steps to set up and run the application: + +1. **Clone the Repository**: + ```bash + git clone https://github.com/Shubhamsaboo/awesome-llm-apps.git + cd ai_agent_tutorials/ai_aqi_analysis_agent + ``` + +2. **Install the dependencies**: + ```bash + pip install -r requirements.txt + ``` + +3. **Set up your API keys**: + - Get an OpenAI API key from: https://platform.openai.com/api-keys + - Get a Firecrawl API key from: [Firecrawl website](https://www.firecrawl.dev/app/api-keys) + +4. **Run the Gradio app**: + ```bash + python ai_aqi_analysis_agent.py + ``` + +5. **Access the Web Interface**: + - The terminal will display two URLs: + - Local URL: `http://127.0.0.1:7860` (for local access) + - Public URL: `https://xxx-xxx-xxx.gradio.live` (for temporary public access) + - Click on either URL to open the web interface in your browser + +## Usage + +1. Enter your API keys in the API Configuration section +2. Input location details: + - City name + - State (optional for Union Territories/US cities) + - Country +3. Provide personal information: + - Medical conditions (optional) + - Planned outdoor activity +4. Click "Analyze & Get Recommendations" to receive: + - Current air quality data + - Health impact analysis + - Activity safety recommendations +5. Try the example queries for quick testing + +## Note + +The air quality data is fetched using Firecrawl's web scraping capabilities. Due to caching and rate limiting, the data might not always match real-time values on the website. For the most accurate real-time data, consider checking the source website directly. diff --git a/ai_agent_tutorials/ai_aqi_analysis_agent/ai_aqi_analysis_agent_gradio.py b/ai_agent_tutorials/ai_aqi_analysis_agent/ai_aqi_analysis_agent_gradio.py new file mode 100644 index 0000000..609d027 --- /dev/null +++ b/ai_agent_tutorials/ai_aqi_analysis_agent/ai_aqi_analysis_agent_gradio.py @@ -0,0 +1,272 @@ +from typing import Dict, Optional +from dataclasses import dataclass +from pydantic import BaseModel, Field +from agno.agent import Agent +from agno.models.openai import OpenAIChat +from firecrawl import FirecrawlApp +import gradio as gr +import json + +class AQIResponse(BaseModel): + success: bool + data: Dict[str, float] + status: str + expiresAt: str + +class ExtractSchema(BaseModel): + aqi: float = Field(description="Air Quality Index") + temperature: float = Field(description="Temperature in degrees Celsius") + humidity: float = Field(description="Humidity percentage") + wind_speed: float = Field(description="Wind speed in kilometers per hour") + pm25: float = Field(description="Particulate Matter 2.5 micrometers") + pm10: float = Field(description="Particulate Matter 10 micrometers") + co: float = Field(description="Carbon Monoxide level") + +@dataclass +class UserInput: + city: str + state: str + country: str + medical_conditions: Optional[str] + planned_activity: str + +class AQIAnalyzer: + + def __init__(self, firecrawl_key: str) -> None: + self.firecrawl = FirecrawlApp(api_key=firecrawl_key) + + def _format_url(self, country: str, state: str, city: str) -> str: + """Format URL based on location, handling cases with and without state""" + country_clean = country.lower().replace(' ', '-') + city_clean = city.lower().replace(' ', '-') + + if not state or state.lower() == 'none': + return f"https://www.aqi.in/dashboard/{country_clean}/{city_clean}" + + state_clean = state.lower().replace(' ', '-') + return f"https://www.aqi.in/dashboard/{country_clean}/{state_clean}/{city_clean}" + + def fetch_aqi_data(self, city: str, state: str, country: str) -> tuple[Dict[str, float], str]: + """Fetch AQI data using Firecrawl""" + try: + url = self._format_url(country, state, city) + info_msg = f"Accessing URL: {url}" + + response = self.firecrawl.extract( + urls=[f"{url}/*"], + params={ + 'prompt': 'Extract the current real-time AQI, temperature, humidity, wind speed, PM2.5, PM10, and CO levels from the page. Also extract the timestamp of the data.', + 'schema': ExtractSchema.model_json_schema() + } + ) + + aqi_response = AQIResponse(**response) + if not aqi_response.success: + raise ValueError(f"Failed to fetch AQI data: {aqi_response.status}") + + return aqi_response.data, info_msg + + except Exception as e: + error_msg = f"Error fetching AQI data: {str(e)}" + return { + 'aqi': 0, + 'temperature': 0, + 'humidity': 0, + 'wind_speed': 0, + 'pm25': 0, + 'pm10': 0, + 'co': 0 + }, error_msg + +class HealthRecommendationAgent: + + def __init__(self, openai_key: str) -> None: + self.agent = Agent( + model=OpenAIChat( + id="gpt-4o", + name="Health Recommendation Agent", + api_key=openai_key + ) + ) + + def get_recommendations( + self, + aqi_data: Dict[str, float], + user_input: UserInput + ) -> str: + prompt = self._create_prompt(aqi_data, user_input) + response = self.agent.run(prompt) + return response.content + + def _create_prompt(self, aqi_data: Dict[str, float], user_input: UserInput) -> str: + return f""" + Based on the following air quality conditions in {user_input.city}, {user_input.state}, {user_input.country}: + - Overall AQI: {aqi_data['aqi']} + - PM2.5 Level: {aqi_data['pm25']} µg/m³ + - PM10 Level: {aqi_data['pm10']} µg/m³ + - CO Level: {aqi_data['co']} ppb + + Weather conditions: + - Temperature: {aqi_data['temperature']}°C + - Humidity: {aqi_data['humidity']}% + - Wind Speed: {aqi_data['wind_speed']} km/h + + User's Context: + - Medical Conditions: {user_input.medical_conditions or 'None'} + - Planned Activity: {user_input.planned_activity} + **Comprehensive Health Recommendations:** + 1. **Impact of Current Air Quality on Health:** + 2. **Necessary Safety Precautions for Planned Activity:** + 3. **Advisability of Planned Activity:** + 4. **Best Time to Conduct the Activity:** + """ + +def analyze_conditions( + city: str, + state: str, + country: str, + medical_conditions: str, + planned_activity: str, + firecrawl_key: str, + openai_key: str +) -> tuple[str, str, str, str]: + """Analyze conditions and return AQI data, recommendations, and status messages""" + try: + # Initialize analyzers + aqi_analyzer = AQIAnalyzer(firecrawl_key=firecrawl_key) + health_agent = HealthRecommendationAgent(openai_key=openai_key) + + # Create user input + user_input = UserInput( + city=city, + state=state, + country=country, + medical_conditions=medical_conditions, + planned_activity=planned_activity + ) + + # Get AQI data + aqi_data, info_msg = aqi_analyzer.fetch_aqi_data( + city=user_input.city, + state=user_input.state, + country=user_input.country + ) + + # Format AQI data for display + aqi_json = json.dumps({ + "Air Quality Index (AQI)": aqi_data['aqi'], + "PM2.5": f"{aqi_data['pm25']} µg/m³", + "PM10": f"{aqi_data['pm10']} µg/m³", + "Carbon Monoxide (CO)": f"{aqi_data['co']} ppb", + "Temperature": f"{aqi_data['temperature']}°C", + "Humidity": f"{aqi_data['humidity']}%", + "Wind Speed": f"{aqi_data['wind_speed']} km/h" + }, indent=2) + + # Get recommendations + recommendations = health_agent.get_recommendations(aqi_data, user_input) + + warning_msg = """ + ⚠️ Note: The data shown may not match real-time values on the website. + This could be due to: + - Cached data in Firecrawl + - Rate limiting + - Website updates not being captured + + Consider refreshing or checking the website directly for real-time values. + """ + + return aqi_json, recommendations, info_msg, warning_msg + + except Exception as e: + error_msg = f"Error occurred: {str(e)}" + return "", "Analysis failed", error_msg, "" + +def create_demo() -> gr.Blocks: + """Create and configure the Gradio interface""" + with gr.Blocks(title="AQI Analysis Agent") as demo: + gr.Markdown( + """ + # 🌍 AQI Analysis Agent + Get personalized health recommendations based on air quality conditions. + """ + ) + + # API Configuration + with gr.Accordion("API Configuration", open=False): + firecrawl_key = gr.Textbox( + label="Firecrawl API Key", + type="password", + placeholder="Enter your Firecrawl API key" + ) + openai_key = gr.Textbox( + label="OpenAI API Key", + type="password", + placeholder="Enter your OpenAI API key" + ) + + # Location Details + with gr.Row(): + with gr.Column(): + city = gr.Textbox(label="City", placeholder="e.g., Mumbai") + state = gr.Textbox( + label="State", + placeholder="Leave blank for Union Territories or US cities", + value="" + ) + country = gr.Textbox(label="Country", value="India") + + # Personal Details + with gr.Row(): + with gr.Column(): + medical_conditions = gr.Textbox( + label="Medical Conditions (optional)", + placeholder="e.g., asthma, allergies", + lines=2 + ) + planned_activity = gr.Textbox( + label="Planned Activity", + placeholder="e.g., morning jog for 2 hours", + lines=2 + ) + + # Status Messages + info_box = gr.Textbox(label="ℹ️ Status", interactive=False) + warning_box = gr.Textbox(label="⚠️ Warning", interactive=False) + + # Output Areas + aqi_data_json = gr.JSON(label="📊 Current Air Quality Data") + recommendations = gr.Markdown(label="🏥 Health Recommendations") + + # Analyze Button + analyze_btn = gr.Button("🔍 Analyze & Get Recommendations", variant="primary") + analyze_btn.click( + fn=analyze_conditions, + inputs=[ + city, + state, + country, + medical_conditions, + planned_activity, + firecrawl_key, + openai_key + ], + outputs=[aqi_data_json, recommendations, info_box, warning_box] + ) + + # Examples + gr.Examples( + examples=[ + ["Mumbai", "Maharashtra", "India", "asthma", "morning walk for 30 minutes"], + ["Delhi", "", "India", "", "outdoor yoga session"], + ["New York", "", "United States", "allergies", "afternoon run"], + ["Kakinada", "Andhra Pradesh", "India", "none", "Tennis for 2 hours"] + ], + inputs=[city, state, country, medical_conditions, planned_activity] + ) + + return demo + +if __name__ == "__main__": + demo = create_demo() + demo.launch(share=True) diff --git a/ai_agent_tutorials/ai_aqi_analysis_agent/ai_aqi_analysis_agent_streamlit.py b/ai_agent_tutorials/ai_aqi_analysis_agent/ai_aqi_analysis_agent_streamlit.py new file mode 100644 index 0000000..b8acaeb --- /dev/null +++ b/ai_agent_tutorials/ai_aqi_analysis_agent/ai_aqi_analysis_agent_streamlit.py @@ -0,0 +1,265 @@ +from typing import Dict, Optional +from dataclasses import dataclass +from pydantic import BaseModel, Field +from agno.agent import Agent +from agno.models.openai import OpenAIChat +from firecrawl import FirecrawlApp +import streamlit as st + +class AQIResponse(BaseModel): + success: bool + data: Dict[str, float] + status: str + expiresAt: str + +class ExtractSchema(BaseModel): + aqi: float = Field(description="Air Quality Index") + temperature: float = Field(description="Temperature in degrees Celsius") + humidity: float = Field(description="Humidity percentage") + wind_speed: float = Field(description="Wind speed in kilometers per hour") + pm25: float = Field(description="Particulate Matter 2.5 micrometers") + pm10: float = Field(description="Particulate Matter 10 micrometers") + co: float = Field(description="Carbon Monoxide level") + +@dataclass +class UserInput: + city: str + state: str + country: str + medical_conditions: Optional[str] + planned_activity: str + +class AQIAnalyzer: + + def __init__(self, firecrawl_key: str) -> None: + self.firecrawl = FirecrawlApp(api_key=firecrawl_key) + + def _format_url(self, country: str, state: str, city: str) -> str: + """Format URL based on location, handling cases with and without state""" + country_clean = country.lower().replace(' ', '-') + city_clean = city.lower().replace(' ', '-') + + if not state or state.lower() == 'none': + return f"https://www.aqi.in/dashboard/{country_clean}/{city_clean}" + + state_clean = state.lower().replace(' ', '-') + return f"https://www.aqi.in/dashboard/{country_clean}/{state_clean}/{city_clean}" + + def fetch_aqi_data(self, city: str, state: str, country: str) -> Dict[str, float]: + """Fetch AQI data using Firecrawl""" + try: + url = self._format_url(country, state, city) + st.info(f"Accessing URL: {url}") # Display URL being accessed + + response = self.firecrawl.extract( + urls=[f"{url}/*"], + params={ + 'prompt': 'Extract the current real-time AQI, temperature, humidity, wind speed, PM2.5, PM10, and CO levels from the page. Also extract the timestamp of the data.', + 'schema': ExtractSchema.model_json_schema() + } + ) + + aqi_response = AQIResponse(**response) + if not aqi_response.success: + raise ValueError(f"Failed to fetch AQI data: {aqi_response.status}") + + with st.expander("📦 Raw AQI Data", expanded=True): + st.json({ + "url_accessed": url, + "timestamp": aqi_response.expiresAt, + "data": aqi_response.data + }) + + st.warning(""" + ⚠️ Note: The data shown may not match real-time values on the website. + This could be due to: + - Cached data in Firecrawl + - Rate limiting + - Website updates not being captured + + Consider refreshing or checking the website directly for real-time values. + """) + + return aqi_response.data + + except Exception as e: + st.error(f"Error fetching AQI data: {str(e)}") + return { + 'aqi': 0, + 'temperature': 0, + 'humidity': 0, + 'wind_speed': 0, + 'pm25': 0, + 'pm10': 0, + 'co': 0 + } + +class HealthRecommendationAgent: + + def __init__(self, openai_key: str) -> None: + self.agent = Agent( + model=OpenAIChat( + id="gpt-4o", + name="Health Recommendation Agent", + api_key=openai_key + ) + ) + + def get_recommendations( + self, + aqi_data: Dict[str, float], + user_input: UserInput + ) -> str: + prompt = self._create_prompt(aqi_data, user_input) + response = self.agent.run(prompt) + return response.content + + def _create_prompt(self, aqi_data: Dict[str, float], user_input: UserInput) -> str: + return f""" + Based on the following air quality conditions in {user_input.city}, {user_input.state}, {user_input.country}: + - Overall AQI: {aqi_data['aqi']} + - PM2.5 Level: {aqi_data['pm25']} µg/m³ + - PM10 Level: {aqi_data['pm10']} µg/m³ + - CO Level: {aqi_data['co']} ppb + + Weather conditions: + - Temperature: {aqi_data['temperature']}°C + - Humidity: {aqi_data['humidity']}% + - Wind Speed: {aqi_data['wind_speed']} km/h + + User's Context: + - Medical Conditions: {user_input.medical_conditions or 'None'} + - Planned Activity: {user_input.planned_activity} + **Comprehensive Health Recommendations:** + 1. **Impact of Current Air Quality on Health:** + 2. **Necessary Safety Precautions for Planned Activity:** + 3. **Advisability of Planned Activity:** + 4. **Best Time to Conduct the Activity:** + """ + +def analyze_conditions( + user_input: UserInput, + api_keys: Dict[str, str] +) -> str: + aqi_analyzer = AQIAnalyzer(firecrawl_key=api_keys['firecrawl']) + health_agent = HealthRecommendationAgent(openai_key=api_keys['openai']) + + aqi_data = aqi_analyzer.fetch_aqi_data( + city=user_input.city, + state=user_input.state, + country=user_input.country + ) + + return health_agent.get_recommendations(aqi_data, user_input) + +def initialize_session_state(): + if 'api_keys' not in st.session_state: + st.session_state.api_keys = { + 'firecrawl': '', + 'openai': '' + } + +def setup_page(): + st.set_page_config( + page_title="AQI Analysis Agent", + page_icon="🌍", + layout="wide" + ) + + st.title("🌍 AQI Analysis Agent") + st.info("Get personalized health recommendations based on air quality conditions.") + +def render_sidebar(): + """Render sidebar with API configuration""" + with st.sidebar: + st.header("🔑 API Configuration") + + new_firecrawl_key = st.text_input( + "Firecrawl API Key", + type="password", + value=st.session_state.api_keys['firecrawl'], + help="Enter your Firecrawl API key" + ) + new_openai_key = st.text_input( + "OpenAI API Key", + type="password", + value=st.session_state.api_keys['openai'], + help="Enter your OpenAI API key" + ) + + if (new_firecrawl_key and new_openai_key and + (new_firecrawl_key != st.session_state.api_keys['firecrawl'] or + new_openai_key != st.session_state.api_keys['openai'])): + st.session_state.api_keys.update({ + 'firecrawl': new_firecrawl_key, + 'openai': new_openai_key + }) + st.success("✅ API keys updated!") + +def render_main_content(): + st.header("📍 Location Details") + col1, col2 = st.columns(2) + + with col1: + city = st.text_input("City", placeholder="e.g., Mumbai") + state = st.text_input("State", placeholder="If it's a Union Territory or a city in the US, leave it blank") + country = st.text_input("Country", value="India", placeholder="United States") + + with col2: + st.header("👤 Personal Details") + medical_conditions = st.text_area( + "Medical Conditions (optional)", + placeholder="e.g., asthma, allergies" + ) + planned_activity = st.text_area( + "Planned Activity", + placeholder="e.g., morning jog for 2 hours" + ) + + return UserInput( + city=city, + state=state, + country=country, + medical_conditions=medical_conditions, + planned_activity=planned_activity + ) + +def main(): + """Main application entry point""" + initialize_session_state() + setup_page() + render_sidebar() + user_input = render_main_content() + + result = None + + if st.button("🔍 Analyze & Get Recommendations"): + if not all([user_input.city, user_input.planned_activity]): + st.error("Please fill in all required fields (state and medical conditions are optional)") + elif not all(st.session_state.api_keys.values()): + st.error("Please provide both API keys in the sidebar") + else: + try: + with st.spinner("🔄 Analyzing conditions..."): + result = analyze_conditions( + user_input=user_input, + api_keys=st.session_state.api_keys + ) + st.success("✅ Analysis completed!") + + except Exception as e: + st.error(f"❌ Error: {str(e)}") + + if result: + st.markdown("### 📦 Recommendations") + st.markdown(result) + + st.download_button( + "💾 Download Recommendations", + data=result, + file_name=f"aqi_recommendations_{user_input.city}_{user_input.state}.txt", + mime="text/plain" + ) + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/ai_agent_tutorials/ai_aqi_analysis_agent/requirements.txt b/ai_agent_tutorials/ai_aqi_analysis_agent/requirements.txt new file mode 100644 index 0000000..034b309 --- /dev/null +++ b/ai_agent_tutorials/ai_aqi_analysis_agent/requirements.txt @@ -0,0 +1,6 @@ +agno +openai +firecrawl-py==1.9.0 +gradio==5.9.1 +pydantic +dataclasses