Climate Impact

Environmental monitoring with impact prediction, adaptation strategies, and sustainability planning

Multi-agent climate analysis: environmental monitoring → urban analysis → impact prediction → strategy generation. Includes SQLite data storage, structured output, and API deployment.

Setup

# Create environment
python3 -m venv venv && source venv/bin/activate

# Install packages
pip install praisonaiagents praisonai

# Set API keys
export OPENAI_API_KEY="your-key"
export TAVILY_API_KEY="your-key"  # For climate news

Create Sample Data

# Create environmental readings
cat > environment.json << 'EOF'
{
  "temperature": {"current": 32.5, "avg_historical": 28.0, "trend": "increasing"},
  "humidity": {"current": 45, "avg_historical": 55, "trend": "decreasing"},
  "air_quality": {"aqi": 125, "pm25": 45, "co2_ppm": 420},
  "precipitation": {"monthly_mm": 25, "avg_historical": 50, "trend": "decreasing"}
}
EOF

# Create urban data
cat > urban.json << 'EOF'
{
  "zones": [
    {"name": "downtown", "green_space_pct": 5, "building_density": 0.85, "heat_island_index": 4.2},
    {"name": "residential", "green_space_pct": 25, "building_density": 0.45, "heat_island_index": 1.8},
    {"name": "industrial", "green_space_pct": 2, "building_density": 0.60, "heat_island_index": 3.5}
  ]
}
EOF

Run: Python Code

from praisonaiagents import Agent, AgentTeam, Task, tool
from pydantic import BaseModel
from typing import List
import json

# Structured output
class ClimateReport(BaseModel):
    zone: str
    risk_level: str  # critical, high, medium, low
    temperature_anomaly: float
    health_impact_score: int
    adaptation_strategies: List[str]
    estimated_cost: str

# Database persistence is configured via memory={} parameter

# Tools
@tool
def get_environmental_data() -> str:
    """Get current environmental readings."""
    with open("environment.json") as f:
        return f.read()

@tool
def get_urban_zones() -> str:
    """Get urban zone data."""
    with open("urban.json") as f:
        return f.read()

@tool
def calculate_heat_risk(temperature: float, humidity: float, aqi: int) -> str:
    """Calculate heat risk index."""
    heat_index = temperature + (0.5 * (temperature - 14.5) * (1 - humidity/100))
    aqi_factor = 1 + (aqi / 500)
    risk_score = heat_index * aqi_factor
    
    if risk_score > 45:
        level = "critical"
    elif risk_score > 38:
        level = "high"
    elif risk_score > 32:
        level = "medium"
    else:
        level = "low"
    
    return json.dumps({"heat_index": round(heat_index, 1), "risk_score": round(risk_score, 1), "level": level})

@tool
def search_climate_news(region: str) -> str:
    """Search for climate news in a region."""
    from tavily import TavilyClient
    import os
    client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
    results = client.search(f"climate change impact {region} 2024", max_results=3)
    return json.dumps([{"title": r["title"], "content": r["content"][:200]} for r in results["results"]])

# Agents
monitor = Agent(
    name="EnvironmentMonitor",
    instructions="Collect environmental data. Use get_environmental_data tool.",
    tools=[get_environmental_data],
    memory={
        "db": "sqlite:///climate.db",
        "session_id": "climate-analysis"
    }
)

analyzer = Agent(
    name="UrbanAnalyzer",
    instructions="Analyze urban zones. Use get_urban_zones tool.",
    tools=[get_urban_zones]
)

predictor = Agent(
    name="ImpactPredictor",
    instructions="Calculate climate risks. Use calculate_heat_risk tool.",
    tools=[calculate_heat_risk, search_climate_news]
)

strategist = Agent(
    name="AdaptationStrategist",
    instructions="Generate adaptation strategies based on risk levels."
)

# Tasks
monitor_task = Task(
    description="Collect current environmental readings",
    agent=monitor,
    expected_output="Environmental data with trends"
)

analyze_task = Task(
    description="Analyze urban zones for climate vulnerability",
    agent=analyzer,
    expected_output="Zone vulnerability assessment"
)

predict_task = Task(
    description="Predict climate impacts for each zone",
    agent=predictor,
    expected_output="Risk assessment per zone"
)

strategy_task = Task(
    description="Generate adaptation strategies",
    agent=strategist,
    expected_output="Structured climate report",
    output_pydantic=ClimateReport
)

# Run
agents = AgentTeam(agents=[monitor, analyzer, predictor, strategist], tasks=[monitor_task, analyze_task, predict_task, strategy_task])
result = agents.start()
print(result)

Run: CLI

# Analyze current conditions
praisonai "Analyze climate impact for downtown area" --verbose

# With web search for news
praisonai "Climate risk assessment with latest news" --web-search --memory --user-id climate_team

# Full report
praisonai "Generate climate adaptation report" --telemetry

Run: agents.yaml

Create agents.yaml:

framework: praisonai
topic: "climate impact analysis"
roles:
  monitor:
    role: Environmental Monitor
    goal: Collect climate data
    backstory: Expert at environmental sensor networks
    tools:
      - tavily_search
    tasks:
      collect:
        description: |
          Collect data:
          - Temperature and humidity
          - Air quality (AQI, PM2.5)
          - Precipitation trends
        expected_output: Environmental readings
        
  analyzer:
    role: Urban Analyst
    goal: Assess urban vulnerability
    backstory: Expert at urban climate analysis
    tasks:
      analyze:
        description: |
          Analyze zones:
          - Heat island effect
          - Green space coverage
          - Building density impact
        expected_output: Vulnerability assessment
        
  strategist:
    role: Adaptation Planner
    goal: Develop climate strategies
    backstory: Expert at climate adaptation
    tasks:
      plan:
        description: |
          Generate strategies:
          - Green infrastructure
          - Cool roofs/pavements
          - Urban forestry
          - Water management
        expected_output: Adaptation plan with costs

Run:

praisonai agents.yaml --web-search --verbose

Monitor & Verify

# View analysis history
praisonai --history 10 --user-id climate_team

# Check metrics
praisonai --metrics

# Export report
praisonai --save climate_report

Serve API

from praisonaiagents import Agent, tool
import json

@tool
def quick_climate_check(temperature: float, aqi: int, green_space_pct: float) -> str:
    """Quick climate risk assessment."""
    risk_score = (temperature - 25) * 2 + (aqi / 50) - (green_space_pct / 5)
    
    if risk_score > 15:
        level, action = "critical", "immediate_intervention"
    elif risk_score > 10:
        level, action = "high", "priority_planning"
    elif risk_score > 5:
        level, action = "medium", "monitoring"
    else:
        level, action = "low", "maintenance"
    
    return json.dumps({"risk_score": round(risk_score, 1), "level": level, "action": action})

agent = Agent(
    name="ClimateAPI",
    instructions="Assess climate risk for urban areas.",
    tools=[quick_climate_check]
)

agent.launch(path="/climate-risk", port=8000)

Test:

curl -X POST http://localhost:8000/climate-risk \
  -H "Content-Type: application/json" \
  -d '{"message": "Check climate risk: temperature 35, AQI 150, green space 5%"}'

Cleanup

rm -f climate.db environment.json urban.json
deactivate

Features Demonstrated

Feature	Implementation
Multi-agent	Monitor → Analyzer → Predictor → Strategist
Web Search	Tavily for climate news
Structured Output	Pydantic ClimateReport
Risk Calculation	Heat index + AQI scoring
DB Persistence	SQLite via db()
CLI	--web-search for news
YAML Config	3-agent climate pipeline
API Endpoint	agent.launch()