- unwind ai
- Posts
- Build an AQI Analysis Agent
Build an AQI Analysis Agent
Fully functional AI agent app with step-by-step instructions (100% opensource)
Air quality has become a crucial health factor, especially in urban areas where pollution levels can significantly impact our daily lives. While many air quality monitoring tools exist, there's a gap when it comes to personalized health recommendations based on real-time air quality data.
In this tutorial, we'll walk you through building a multi-agent AQI Analysis App that gives personalized health recommendations based on real-time air quality data. This system will analyze current air conditions and provide tailored advice based on your health conditions and planned activities.
For this project, we'll be using a powerful stack of technologies:
Firecrawl serves as our web scraping tool, extracting real-time air quality data from online sources without requiring a complex setup.
Agno (formerly Phidata) provides the agent framework that allows us to create and coordinate our specialized AI agents.
OpenAI GPT-4o powers our health recommendation agent with advanced reasoning capabilities for personalized advice.
Streamlit creates our user-friendly interface, making it easy to input location data and display recommendations.
What Weโre Building
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, and wind speed
Comprehensive Analysis
Real-time data visualization
Health impact assessment
Activity safety recommendations
Best time suggestions for outdoor activities
Interactive Features
Location-based analysis
Medical condition considerations
Activity-specific recommendations
Downloadable reports
Prerequisites
Before we begin, make sure you have the following:
Step-by-Step Instructions
Setting Up the Environment
First, let's get our development environment ready:
Clone the GitHub repository:
git clone https://github.com/Shubhamsaboo/awesome-llm-apps.git
Go to the ai_aqi_analysis_agent folder:
cd ai_agent_tutorials/ai_aqi_analysis_agent
Install the required dependencies:
pip install -r requirements.txt
Creating the Streamlit App
Letโs create our app. Create a new file ai_aqi_analysis_agent_streamlit.py
and add the following code:
Let's import our libraries and define our data models:
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
Let's define our data structures:
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
Now let's create the AQIAnalyzer class to fetch real-time air quality data:
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}"
Add the method to fetch AQI data using Firecrawl:
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
}
Now let's create our Health Recommendation Agent:
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
Add the prompt creation method for our agent:
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:**
"""
Create our main analysis function:
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': ''
}
Set up the Streamlit UI:
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!")
Create the main content rendering function:
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
)
Implement the main application function:
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()
Running the App
With our code in place, it's time to launch the app.
In your terminal, navigate to the project folder, and run the following command
streamlit run ai_aqi_analysis_agent_streamlit.py
Streamlit will provide a local URL (typically http://localhost:8501).
How the App works:
Enter your API keys in the API Configuration section
Input location details:
City name
State (optional for Union Territories/US cities)
Country
Provide personal information:
Medical conditions (optional)
Planned outdoor activity
The AQI Analyzer uses Firecrawl to scrape real-time air quality data from aqi.in based on the user's location.
The scraped data is parsed into a structured format containing AQI, particulate matter levels, weather conditions, and other relevant metrics.
The Health Recommendation Agent, powered by GPT-4o, combines the air quality data with user-provided information about medical conditions and planned activities.
Based on all inputs, the agent generates comprehensive health recommendations, including:
Health impact assessment
Safety precautions
Activity advisability
Optimal timing suggestions
The recommendations are displayed in the interface and can be downloaded as a text file.
Working Application Demo
Conclusion
Your AQI Analysis Agent is ready to provide personalized health recommendations based on real-time air quality data. This multi-agent system combines the power of web scraping, AI language models, and user context to deliver actionable insights.
For further enhancements, consider:
Custom Activity Profiles: Allow users to create and save activity profiles with different sensitivity levels and requirements.
Pollution Source Attribution: Identify and report likely sources of current pollution (traffic, industrial, wildfire, etc.) with relevant avoidance strategies for each type.
Historical Data Analysis: Incorporate historical AQI trends to provide forecasting and pattern recognition.
Location Auto-detection: Implement geolocation to automatically detect the user's current location.
Exposure Budget Calculator: Implement a feature that calculates a "pollution exposure budget" for users with respiratory conditions.
Travel Planning Mode: Develop functionality for comparing air quality between multiple destinations to help users make informed travel decisions.
Keep experimenting with different agent configurations and features to build more sophisticated AI applications.
We share hands-on tutorials like this 2-3 times a week, to help you stay ahead in the world of AI. If you're serious about leveling up your AI skills and staying ahead of the curve, subscribe now and be the first to access our latest tutorials.
Reply