added AI AQI analysis agent3

ai_agent_tutorials/ai_aqi_analysis_agent/ai_aqi_analysis_agent.py

@@ -1,11 +1,4 @@
-"""
+from typing import Dict, Optional
-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 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 AQIResponse(BaseModel):
-class AQIExtractSchema(BaseModel):
+    success: bool
-    """Schema for AQI data extraction"""
+    data: Dict[str, float]
-    aqi: int = Field(description="Current AQI value")
+    status: str
-    temperature: float = Field(description="Temperature in Celsius")
+    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")
+    wind_speed: float = Field(description="Wind speed in kilometers per hour")
-    pm25: float = Field(description="PM2.5 level in µg/m³")
+    pm25: float = Field(description="Particulate Matter 2.5 micrometers")
-    pm10: float = Field(description="PM10 level in µg/m³")
+    pm10: float = Field(description="Particulate Matter 10 micrometers")
-    co: float = Field(description="CO level in ppb")
+    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 AQIAnalyzer:
-class AQIDataAgent:
-    """Agent responsible for fetching AQI and weather data"""
     
-    def __init__(self, firecrawl_key: str, openai_key: str) -> None:
+    def __init__(self, firecrawl_key: str) -> None:
-        """Initialize with API keys"""
         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:
+    def fetch_aqi_data(self, city: str, state: str, country: str) -> Dict[str, float]:
-        """Fetch weather and AQI data for given location"""
         try:
-            base_url = self._format_url(country, state, city)
+            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.
-            """
             
             response = self.firecrawl.extract(
-                urls=urls,
+                urls=[f"{url}/*"],
                 params={
-                    'prompt': extract_prompt,
+                    'prompt': 'Extract the AQI, temperature, humidity, wind speed, PM2.5, PM10, and CO levels from the page.',
-                    'schema': AQIExtractSchema.model_json_schema()
+                    'schema': ExtractSchema.model_json_schema()
                 }
             )
             
-            if isinstance(response, dict) and 'error' in response:
+            aqi_response = AQIResponse(**response)
-                raise ValueError(f"Firecrawl error: {response['error']}")
+            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 {
-            return AQIExtractSchema(
+                'aqi': 0,
-                aqi=0,
+                'temperature': 0,
-                temperature=0.0,
+                'humidity': 0,
-                humidity=0.0,
+                'wind_speed': 0,
-                wind_speed=0.0,
+                'pm25': 0,
-                pm25=0.0,
+                'pm10': 0,
-                pm10=0.0,
+                'co': 0
-                co=0.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(
+    def get_recommendations(
         self,
-        aqi_data: AQIExtractSchema,
+        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(
+    def _create_prompt(self, aqi_data: Dict[str, float], user_input: UserInput) -> str:
-        self,
-        aqi_data: AQIExtractSchema,
-        user_input: UserInput
-    ) -> str:
-        """Create detailed prompt for health recommendations"""
         return f"""
         Based on the following air quality conditions in {user_input.city}, {user_input.state}, {user_input.country}:
-        - Overall AQI: {aqi_data.aqi}
+        - Overall AQI: {aqi_data['aqi']}
-        - PM2.5 Level: {aqi_data.pm25} µg/m³
+        - PM2.5 Level: {aqi_data['pm25']} µg/m³
-        - PM10 Level: {aqi_data.pm10} µg/m³
+        - PM10 Level: {aqi_data['pm10']} µg/m³
-        - CO Level: {aqi_data.co} ppb
+        - CO Level: {aqi_data['co']} ppb
         
         Weather conditions:
-        - Temperature: {aqi_data.temperature}°C
+        - Temperature: {aqi_data['temperature']}°C
-        - Humidity: {aqi_data.humidity}%
+        - Humidity: {aqi_data['humidity']}%
-        - Wind Speed: {aqi_data.wind_speed} km/h
+        - 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
+def analyze_conditions(
-async def analyze_conditions(
     user_input: UserInput,
     api_keys: Dict[str, str]
 ) -> str:
-    """Main function to analyze conditions and provide recommendations"""
+    aqi_analyzer = AQIAnalyzer(firecrawl_key=api_keys['firecrawl'])
-    # Initialize agents
+    health_agent = HealthRecommendationAgent(openai_key=api_keys['openai'])
-    aqi_agent = AQIDataAgent(
-        firecrawl_key=api_keys['firecrawl'],
-        openai_key=api_keys['openai']
-    )
-    health_agent = HealthRecommendationAgent(
-        openai_key=api_keys['openai']
-    )
     
-    # Get data and recommendations
+    aqi_data = aqi_analyzer.fetch_aqi_data(
-    aqi_data = await aqi_agent.fetch_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("""
+    st.title("🌍 AQI Analysis Assistant")
-        <style>
+    st.info("Get personalized health recommendations based on air quality conditions.")
-        .main { padding: 2rem; }
-        .stButton > button { width: 100%; }
-        .success-message { 
-            padding: 1rem;
-            border-radius: 0.5rem;
-            background-color: #dcfce7;
-            color: #166534;
-        }
-        .error-message {
-            padding: 1rem;
-            border-radius: 0.5rem;
-            background-color: #fee2e2;
-            color: #991b1b;
-        }
-        </style>
-    """, unsafe_allow_html=True)
 
 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.header("📍 Location Details")
-    st.title("🌍 AQI Analysis Assistant")
+    col1, col2 = st.columns(2)
-    st.markdown("Get personalized health recommendations based on air quality conditions.")
-    
-    col1, col2 = st.columns([2, 1])
     
     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(
+                    result = analyze_conditions(
-                        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",

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)