nbws/disaster-response-api
0
1"""2Disaster Response Coordinator - orchestrates multi-agent workflows.3"""4 5from crewai import Crew, Agent, Task6from crewai.project import CrewBase, agent, crew, task7from groq import Groq8from typing import Dict, List, Optional, Any9import asyncio10import logging11from datetime import datetime, timedelta12import json13 14from tools.weather_tools import WeatherTool15from tools.social_tools import SocialTool 16from tools.routing_tools import RoutingTool17from tools.communication_tools import CommunicationTool18from tools.risk_tools import RiskTool19from utils.config import settings20from utils.database import get_db_session21from models.incident import Incident, RiskAssessment, ResourceAllocation22 23import os24os.environ['CREWAI_DB_PATH'] = '/tmp/crewai_db'25 26# In the _create_crew method27def _create_crew(self) -> Crew:28 """Create the main crew with all agents."""29 return Crew(30 agents=list(self.agents.values()),31 verbose=True,32 memory=True33 # Removed embedder config34 )35 36logger = logging.getLogger(__name__)37 38class DisasterCoordinator:39 """Coordinates all disaster response agents and workflows."""40 41 def __init__(self):42 self.groq_client = Groq(api_key=settings.groq_api_key)43 self.tools = self._initialize_tools()44 self.agents = {}45 self.crew = None46 self.active_incidents = {}47 self.is_running = False48 49 def _initialize_tools(self):50 """Initialize all external API tools."""51 return {52 'weather': WeatherTool(),53 'social': SocialTool(),54 'routing': RoutingTool(), 55 'communication': CommunicationTool(),56 'risk': RiskTool()57 }58 59 async def initialize(self):60 """Initialize the coordinator and all agents."""61 try:62 logger.info("Initializing Disaster Response Coordinator...")63 64 # Initialize tools65 for tool_name, tool in self.tools.items():66 await tool.initialize()67 logger.info(f"Initialized {tool_name} tool")68 69 # Create agents70 self.agents = {71 'data_ingestion': self._create_data_ingestion_agent(),72 'signal_triage': self._create_signal_triage_agent(),73 'risk_scoring': self._create_risk_scoring_agent(),74 'resource_allocation': self._create_resource_allocation_agent(),75 'communications': self._create_communications_agent()76 }77 78 # Create crew with all agents79 self.crew = self._create_crew()80 81 self.is_running = True82 logger.info("Coordinator initialized successfully")83 84 except Exception as e:85 logger.error(f"Failed to initialize coordinator: {e}")86 raise87 88 def _create_data_ingestion_agent(self) -> Agent:89 """Create the data ingestion agent."""90 return Agent(91 role='Data Ingestion Specialist',92 goal='Continuously monitor and ingest data from multiple sources including weather, social media, and sensor feeds',93 backstory="""You are an experienced data engineer specializing in real-time disaster monitoring.94 Your expertise lies in collecting, validating, and preprocessing data from diverse sources.95 You ensure data quality and handle rate limiting and API failures gracefully.""",96 verbose=True,97 allow_delegation=False,98 tools=[99 self.tools['weather'],100 self.tools['social']101 ],102 llm=self.groq_client103 )104 105 def _create_signal_triage_agent(self) -> Agent:106 """Create the signal triage agent."""107 return Agent(108 role='Signal Triage Analyst', 109 goal='Analyze incoming signals for credibility, urgency, and relevance to filter out noise and false alarms',110 backstory="""You are a crisis analyst with years of experience in emergency management.111 You excel at quickly assessing the credibility of reports and determining their urgency.112 Your decisions help prevent resource waste on false alarms while ensuring real emergencies get immediate attention.""",113 verbose=True,114 allow_delegation=False,115 tools=[self.tools['risk']],116 llm=self.groq_client117 )118 119 def _create_risk_scoring_agent(self) -> Agent:120 """Create the risk scoring agent."""121 return Agent(122 role='Risk Assessment Specialist',123 goal='Evaluate and score risk levels for different geographic zones based on hazard, exposure, and vulnerability data',124 backstory="""You are a disaster risk reduction expert with deep knowledge of hazard assessment methodologies.125 You combine meteorological, geological, and social vulnerability data to produce accurate risk scores.126 Your assessments directly inform evacuation and resource allocation decisions.""",127 verbose=True,128 allow_delegation=False,129 tools=[130 self.tools['weather'],131 self.tools['risk']132 ],133 llm=self.groq_client134 )135 136 def _create_resource_allocation_agent(self) -> Agent:137 """Create the resource allocation agent.""" 138 return Agent(139 role='Resource Allocation Coordinator',140 goal='Optimize evacuation routes, shelter assignments, and emergency resource distribution based on current conditions',141 backstory="""You are an operations research specialist with expertise in emergency logistics.142 You excel at solving complex optimization problems under time pressure.143 Your route and resource recommendations save lives by ensuring efficient emergency response.""",144 verbose=True,145 allow_delegation=False,146 tools=[147 self.tools['routing'],148 self.tools['risk']149 ],150 llm=self.groq_client151 )152 153 def _create_communications_agent(self) -> Agent:154 """Create the communications agent."""155 return Agent(156 role='Emergency Communications Specialist',157 goal='Draft clear, actionable emergency communications and coordinate notifications to stakeholders',158 backstory="""You are a crisis communications expert who understands how to communicate effectively during emergencies.159 You craft messages that are clear, culturally sensitive, and appropriate for different audiences.160 Your communications help coordinate response efforts and keep communities informed.""",161 verbose=True,162 allow_delegation=False,163 tools=[self.tools['communication']],164 llm=self.groq_client165 )166 167 def _create_crew(self) -> Crew:168 """Create the main crew with all agents."""169 return Crew(170 agents=list(self.agents.values()),171 verbose=True,172 memory=True,173 embedder={174 "provider": "huggingface",175 "config": {176 "model": "sentence-transformers/all-MiniLM-L6-v2"177 }178 }179 )180 181 async def run_emergency_assessment(self, location: Dict[str, float], incident_type: str, severity: int):182 """Run complete emergency assessment workflow."""183 try:184 incident_id = f"incident_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}"185 logger.info(f"Starting emergency assessment {incident_id}")186 187 # Create incident record188 incident = Incident(189 id=incident_id,190 location=location,191 incident_type=incident_type,192 severity=severity,193 timestamp=datetime.utcnow(),194 status="processing"195 )196 197 self.active_incidents[incident_id] = incident198 199 # Define workflow tasks200 tasks = [201 self._create_ingestion_task(location, incident_type),202 self._create_triage_task(incident_id),203 self._create_risk_assessment_task(location),204 self._create_resource_allocation_task(location),205 self._create_communication_task(incident_id)206 ]207 208 # Execute workflow209 result = await self._execute_crew_tasks(tasks)210 211 # Update incident with results212 incident.status = "completed"213 incident.results = result214 215 logger.info(f"Emergency assessment {incident_id} completed")216 return result217 218 except Exception as e:219 logger.error(f"Error in emergency assessment: {e}")220 if incident_id in self.active_incidents:221 self.active_incidents[incident_id].status = "failed"222 self.active_incidents[incident_id].error = str(e)223 raise224 225 def _create_ingestion_task(self, location: Dict[str, float], incident_type: str) -> Task:226 """Create data ingestion task."""227 return Task(228 description=f"""229 Ingest and process real-time data for location {location} regarding {incident_type} incident.230 231 Your tasks:232 1. Collect current weather conditions and alerts233 2. Monitor social media for relevant reports 234 3. Check for any seismic activity or related hazards235 4. Validate and geocode all collected data236 5. Return structured data with confidence scores237 238 Focus on data quality and include uncertainty estimates.239 """,240 agent=self.agents['data_ingestion'],241 expected_output="Structured JSON with weather, social, and hazard data including confidence scores"242 )243 244 def _create_triage_task(self, incident_id: str) -> Task:245 """Create signal triage task."""246 return Task(247 description=f"""248 Analyze collected signals for incident {incident_id} to determine credibility and urgency.249 250 Your tasks:251 1. Assess credibility of social media reports (1-10 scale)252 2. Cross-reference multiple data sources253 3. Identify potential false alarms or duplicate reports254 4. Determine incident urgency level (low/medium/high/critical)255 5. Flag any reports requiring immediate human attention256 257 Provide clear reasoning for your assessments.258 """,259 agent=self.agents['signal_triage'],260 expected_output="Triage report with credibility scores, urgency level, and recommendations"261 )262 263 def _create_risk_assessment_task(self, location: Dict[str, float]) -> Task:264 """Create risk assessment task."""265 return Task(266 description=f"""267 Conduct comprehensive risk assessment for location {location}.268 269 Your tasks:270 1. Analyze hazard intensity and spatial extent271 2. Assess population exposure in affected areas272 3. Evaluate social vulnerability factors273 4. Calculate composite risk scores for geographic zones274 5. Estimate affected population and potential impacts275 6. Include uncertainty bounds in all assessments276 277 Use standard disaster risk assessment frameworks.278 """,279 agent=self.agents['risk_scoring'],280 expected_output="Risk assessment with zone-based scores, impact estimates, and confidence intervals"281 )282 283 def _create_resource_allocation_task(self, location: Dict[str, float]) -> Task:284 """Create resource allocation task."""285 return Task(286 description=f"""287 Develop optimal resource allocation plan for location {location}.288 289 Your tasks:290 1. Calculate evacuation routes avoiding hazard areas291 2. Match at-risk populations with appropriate shelters292 3. Assign emergency responders to priority zones293 4. Optimize supply distribution based on needs assessment294 5. Consider transportation constraints and capacity limits295 6. Provide alternative plans for different scenarios296 297 Minimize total evacuation time and maximize coverage.298 """,299 agent=self.agents['resource_allocation'],300 expected_output="Resource allocation plan with routes, assignments, and alternatives"301 )302 303 def _create_communication_task(self, incident_id: str) -> Task:304 """Create communication task."""305 return Task(306 description=f"""307 Prepare emergency communications for incident {incident_id}.308 309 Your tasks:310 1. Draft public alert message (clear, actionable, culturally appropriate)311 2. Create briefing for emergency responders312 3. Prepare social media updates with key information313 4. Generate notifications for relevant stakeholders314 5. Ensure messages are appropriate for different audiences315 6. Include all necessary safety instructions316 317 All communications require human approval before sending.318 """,319 agent=self.agents['communications'],320 expected_output="Draft communications package for human review and approval"321 )322 323 async def _execute_crew_tasks(self, tasks: List[Task]) -> Dict[str, Any]:324 """Execute crew tasks and return results."""325 try:326 # For now, run tasks sequentially327 # In production, some could run in parallel328 results = {}329 330 for task in tasks:331 logger.info(f"Executing task: {task.description[:50]}...")332 333 # Simulate task execution with mock data for development334 task_result = await self._mock_task_execution(task)335 results[task.agent.role] = task_result336 337 logger.info(f"Completed task for {task.agent.role}")338 339 return results340 341 except Exception as e:342 logger.error(f"Error executing crew tasks: {e}")343 raise344 345 async def _mock_task_execution(self, task: Task) -> Dict[str, Any]:346 """Mock task execution for development."""347 # This will be replaced with actual CrewAI execution348 agent_role = task.agent.role349 350 if "Data Ingestion" in agent_role:351 return {352 "weather": {"temperature": 15, "wind_speed": 25, "alerts": ["High Wind Warning"]},353 "social": {"relevant_posts": 3, "credibility_avg": 7.5},354 "hazards": {"earthquake_risk": "low", "flood_risk": "medium"}355 }356 elif "Signal Triage" in agent_role:357 return {358 "credibility_score": 8.2,359 "urgency_level": "high", 360 "false_alarm_probability": 0.15,361 "requires_human_review": False362 }363 elif "Risk Assessment" in agent_role:364 return {365 "zone_scores": {"zone_1": 6.5, "zone_2": 8.1, "zone_3": 4.2},366 "affected_population": 15000,367 "confidence_interval": [0.7, 0.9]368 }369 elif "Resource Allocation" in agent_role:370 return {371 "evacuation_routes": [{"route_id": "R1", "capacity": 5000, "time_estimate": "45min"}],372 "shelter_assignments": {"shelter_A": 3000, "shelter_B": 2000},373 "responder_deployment": {"team_1": "zone_2", "team_2": "zone_1"}374 }375 elif "Communications" in agent_role:376 return {377 "public_alert": "Severe weather warning in effect. Seek shelter immediately.",378 "responder_brief": "High priority incident in zones 1-2. Deploy teams as assigned.",379 "social_updates": ["#WeatherAlert: Stay indoors", "#EmergencyUpdate: Shelters open"],380 "stakeholder_notifications": ["Mayor", "Emergency Director", "Red Cross"]381 }382 else:383 return {"status": "completed", "timestamp": datetime.utcnow().isoformat()}384 385 async def get_system_status(self) -> Dict[str, Any]:386 """Get current system status."""387 return {388 "status": "operational" if self.is_running else "offline",389 "agents_count": len(self.agents),390 "active_incidents": len(self.active_incidents),391 "tools_status": {name: "active" for name in self.tools.keys()},392 "last_updated": datetime.utcnow().isoformat()393 }394 395 async def shutdown(self):396 """Shutdown coordinator and cleanup resources."""397 logger.info("Shutting down coordinator...")398 self.is_running = False399 400 # Cleanup tools401 for tool in self.tools.values():402 if hasattr(tool, 'cleanup'):403 await tool.cleanup()404 405 logger.info("Coordinator shutdown complete")