diff --git a/ai_agent_tutorials/ai_aqi_analysis_agent/ai_aqi_analysis_agent.py b/ai_agent_tutorials/ai_aqi_analysis_agent/ai_aqi_analysis_agent.py index 91bdd34..885d832 100644 --- a/ai_agent_tutorials/ai_aqi_analysis_agent/ai_aqi_analysis_agent.py +++ b/ai_agent_tutorials/ai_aqi_analysis_agent/ai_aqi_analysis_agent.py @@ -1,11 +1,4 @@ -""" -AQI Analysis Assistant ---------------------- -A Streamlit application that provides health recommendations based on air quality conditions. -Uses Firecrawl for AQI data and OpenAI for health recommendations. -""" - -from typing import Dict, Optional, TypedDict +from typing import Dict, Optional from dataclasses import dataclass from pydantic import BaseModel, Field from agno.agent import Agent @@ -14,130 +7,99 @@ from firecrawl import FirecrawlApp import streamlit as st import asyncio -# Data Models -class AQIExtractSchema(BaseModel): - """Schema for AQI data extraction""" - aqi: int = Field(description="Current AQI value") - temperature: float = Field(description="Temperature in Celsius") +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 km/h") - pm25: float = Field(description="PM2.5 level in µg/m³") - pm10: float = Field(description="PM10 level in µg/m³") - co: float = Field(description="CO level in ppb") + 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: - """Structure for user input data""" city: str state: str country: str medical_conditions: Optional[str] planned_activity: str -# Agent Classes -class AQIDataAgent: - """Agent responsible for fetching AQI and weather data""" +class AQIAnalyzer: - def __init__(self, firecrawl_key: str, openai_key: str) -> None: - """Initialize with API keys""" + def __init__(self, firecrawl_key: str) -> None: self.firecrawl = FirecrawlApp(api_key=firecrawl_key) - self.agent = Agent( - model=OpenAIChat( - id="gpt-4o", - api_key=openai_key - ), - description="Expert in analyzing air quality data and weather conditions" - ) def _format_url(self, country: str, state: str, city: str) -> str: - """Format location URL with proper formatting for multi-word locations""" return f"https://www.aqi.in/dashboard/{country.lower().replace(' ', '-')}/{state.lower().replace(' ', '-')}/{city.lower().replace(' ', '-')}" - async def fetch_data(self, city: str, state: str, country: str) -> AQIExtractSchema: - """Fetch weather and AQI data for given location""" + def fetch_aqi_data(self, city: str, state: str, country: str) -> Dict[str, float]: try: - base_url = self._format_url(country, state, city) - urls = [base_url, f"{base_url}/pm", f"{base_url}/co", f"{base_url}/pm10"] - - extract_prompt = """ - Extract the following air quality and weather metrics from the page: - - Current AQI value as an integer - - Temperature in Celsius as a float - - Humidity percentage as a float - - Wind speed in km/h as a float - - PM2.5 level in µg/m³ as a float - - PM10 level in µg/m³ as a float - - CO level in ppb as a float - - Return these exact metrics with their specified types. - """ + url = self._format_url(country, state, city) response = self.firecrawl.extract( - urls=urls, + urls=[f"{url}/*"], params={ - 'prompt': extract_prompt, - 'schema': AQIExtractSchema.model_json_schema() + 'prompt': 'Extract the AQI, temperature, humidity, wind speed, PM2.5, PM10, and CO levels from the page.', + 'schema': ExtractSchema.model_json_schema() } ) - if isinstance(response, dict) and 'error' in response: - raise ValueError(f"Firecrawl error: {response['error']}") + aqi_response = AQIResponse(**response) + if not aqi_response.success: + raise ValueError(f"Failed to fetch AQI data: {aqi_response.status}") - return AQIExtractSchema(**response) + return aqi_response.data except Exception as e: st.error(f"Error fetching AQI data: {str(e)}") - # Return default values if fetch fails - return AQIExtractSchema( - aqi=0, - temperature=0.0, - humidity=0.0, - wind_speed=0.0, - pm25=0.0, - pm10=0.0, - co=0.0 - ) + return { + 'aqi': 0, + 'temperature': 0, + 'humidity': 0, + 'wind_speed': 0, + 'pm25': 0, + 'pm10': 0, + 'co': 0 + } class HealthRecommendationAgent: - """Agent responsible for providing health recommendations""" def __init__(self, openai_key: str) -> None: - """Initialize with OpenAI API key""" self.agent = Agent( model=OpenAIChat( id="gpt-4o", + name="Health Recommendation Agent", api_key=openai_key - ), - description="Health recommendation expert for air quality conditions" + ) ) - async def get_recommendations( - self, - aqi_data: AQIExtractSchema, + def get_recommendations( + self, + aqi_data: Dict[str, float], user_input: UserInput ) -> str: - """Generate health recommendations based on conditions""" prompt = self._create_prompt(aqi_data, user_input) - response = await self.agent.run(prompt) + response = self.agent.run(prompt) return response.content - def _create_prompt( - self, - aqi_data: AQIExtractSchema, - user_input: UserInput - ) -> str: - """Create detailed prompt for health recommendations""" + 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 + - 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 + - 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'} @@ -151,33 +113,22 @@ class HealthRecommendationAgent: 5. Best time to conduct the activity if applicable """ -# Main Analysis Function -async def analyze_conditions( +def analyze_conditions( user_input: UserInput, api_keys: Dict[str, str] ) -> str: - """Main function to analyze conditions and provide recommendations""" - # Initialize agents - aqi_agent = AQIDataAgent( - firecrawl_key=api_keys['firecrawl'], - openai_key=api_keys['openai'] - ) - health_agent = HealthRecommendationAgent( - openai_key=api_keys['openai'] - ) + aqi_analyzer = AQIAnalyzer(firecrawl_key=api_keys['firecrawl']) + health_agent = HealthRecommendationAgent(openai_key=api_keys['openai']) - # Get data and recommendations - aqi_data = await aqi_agent.fetch_data( + aqi_data = aqi_analyzer.fetch_aqi_data( city=user_input.city, state=user_input.state, country=user_input.country ) - return await health_agent.get_recommendations(aqi_data, user_input) + return health_agent.get_recommendations(aqi_data, user_input) -# Streamlit UI Components def initialize_session_state(): - """Initialize Streamlit session state""" if 'api_keys' not in st.session_state: st.session_state.api_keys = { 'firecrawl': '', @@ -185,34 +136,16 @@ def initialize_session_state(): } def setup_page(): - """Configure page settings and styles""" st.set_page_config( page_title="AQI Analysis Assistant", page_icon="🌍", layout="wide" ) - st.markdown(""" - - """, unsafe_allow_html=True) + st.title("🌍 AQI Analysis Assistant") + 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") @@ -238,18 +171,15 @@ def render_sidebar(): st.success("✅ API keys updated!") def render_main_content(): - """Render main content area""" - st.title("🌍 AQI Analysis Assistant") - st.markdown("Get personalized health recommendations based on air quality conditions.") - - col1, col2 = st.columns([2, 1]) + st.header("📍 Location Details") + col1, col2 = st.columns(2) with col1: - st.header("📍 Location Details") city = st.text_input("City", placeholder="e.g., Mumbai") state = st.text_input("State", placeholder="e.g., Maharashtra") country = st.text_input("Country", value="India") - + + with col2: st.header("👤 Personal Details") medical_conditions = st.text_area( "Medical Conditions (optional)", @@ -269,7 +199,6 @@ def render_main_content(): ) def main(): - """Main application entry point""" initialize_session_state() setup_page() render_sidebar() @@ -283,16 +212,14 @@ def main(): else: try: with st.spinner("🔄 Analyzing conditions..."): - result = asyncio.run( - analyze_conditions( - user_input=user_input, - api_keys=st.session_state.api_keys - ) + result = analyze_conditions( + user_input=user_input, + api_keys=st.session_state.api_keys ) st.success("✅ Analysis completed!") st.markdown("### 📊 Recommendations") - st.markdown(result) + st.info(result) st.download_button( "💾 Download Recommendations", diff --git a/ai_agent_tutorials/ai_aqi_analysis_agent/test_firecrawl.py b/ai_agent_tutorials/ai_aqi_analysis_agent/test_firecrawl.py new file mode 100644 index 0000000..3829df3 --- /dev/null +++ b/ai_agent_tutorials/ai_aqi_analysis_agent/test_firecrawl.py @@ -0,0 +1,22 @@ +from firecrawl import FirecrawlApp +from pydantic import BaseModel, Field + +# Initialize the FirecrawlApp with your API key +app = FirecrawlApp(api_key='') + +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") + +data = app.extract([ + 'https://www.aqi.in/dashboard/india/andhra-pradesh/kakinada/*' +], { + 'prompt': 'Extract the AQI, temperature, humidity, wind speed, PM2.5, PM10, and CO levels from the page.', + 'schema': ExtractSchema.model_json_schema(), +}) +print(data) \ No newline at end of file