# CareLoop Hackathon Submission - Main Application # AI-Powered Family Caregiving Assistant import asyncio import json from datetime import datetime, timedelta from typing import Dict, List, Any, TypedDict, Optional from dataclasses import dataclass, asdict from enum import Enum import random import uuid # LangGraph and LangChain imports from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langchain_openai import ChatOpenAI from langchain_core.tools import tool # FastAPI for web interface from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.staticfiles import StaticFiles from fastapi.responses import HTMLResponse import uvicorn # ============== MOCK DATA MODELS ============== class AlertLevel(Enum): LOW = "low" MEDIUM = "medium" HIGH = "high" URGENT = "urgent" @dataclass class Parent: id: str name: str age: int conditions: List[str] medications: List[Dict] emergency_contacts: List[Dict] preferences: Dict @dataclass class FamilyMember: id: str name: str relationship: str phone: str email: str role: str notification_preferences: Dict @dataclass class HealthMetric: timestamp: datetime metric_type: str value: float unit: str source: str normal_range: tuple # ============== COMPREHENSIVE MOCK DATA ============== class MockDataGenerator: def __init__(self): self.parents = {} self.families = {} self.health_data = {} self.medication_data = {} self.setup_families() def setup_families(self): # Family 1: Margaret Chen (active, multiple conditions) margaret = Parent( id="parent_001", name="Margaret Chen", age=78, conditions=["Type 2 Diabetes", "Hypertension", "Mild Cognitive Impairment"], medications=[ {"name": "Metformin", "dosage": "500mg", "frequency": "2x daily", "times": ["08:00", "20:00"]}, {"name": "Lisinopril", "dosage": "10mg", "frequency": "1x daily", "times": ["08:00"]}, {"name": "Donepezil", "dosage": "5mg", "frequency": "1x daily", "times": ["20:00"]} ], emergency_contacts=[ {"name": "Dr. Sarah Kim", "phone": "555-MED-1234", "type": "primary_care"}, {"name": "Mercy General Hospital", "phone": "555-911-HELP", "type": "emergency"} ], preferences={ "preferred_contact_time": "morning", "communication_style": "gentle", "daily_check_in_time": "19:00" } ) margaret_family = [ FamilyMember( id="family_001", name="David Chen", relationship="son", phone="555-123-4567", email="david.chen@email.com", role="primary_caregiver", notification_preferences={ "urgent_alerts": True, "daily_summary": True, "medication_reminders": True, "health_trends": True } ), FamilyMember( id="family_002", name="Lisa Chen-Rodriguez", relationship="daughter", phone="555-987-6543", email="lisa.rodriguez@email.com", role="backup_caregiver", notification_preferences={ "urgent_alerts": True, "daily_summary": False, "medication_reminders": False, "health_trends": True } ), FamilyMember( id="family_003", name="Jennifer Chen", relationship="daughter-in-law", phone="555-456-7890", email="jen.chen@email.com", role="support", notification_preferences={ "urgent_alerts": True, "daily_summary": False, "medication_reminders": False, "health_trends": False } ) ] # Family 2: Robert Johnson (recent stroke recovery) robert = Parent( id="parent_002", name="Robert Johnson", age=72, conditions=["Stroke Recovery", "Atrial Fibrillation", "Arthritis"], medications=[ {"name": "Warfarin", "dosage": "5mg", "frequency": "1x daily", "times": ["18:00"]}, {"name": "Atorvastatin", "dosage": "20mg", "frequency": "1x daily", "times": ["20:00"]}, {"name": "Aspirin", "dosage": "81mg", "frequency": "1x daily", "times": ["08:00"]} ], emergency_contacts=[ {"name": "Dr. James Miller", "phone": "555-MED-8765", "type": "primary_care"}, {"name": "Central Hospital", "phone": "555-911-9999", "type": "emergency"} ], preferences={ "preferred_contact_time": "afternoon", "communication_style": "direct", "daily_check_in_time": "17:00" } ) robert_family = [ FamilyMember( id="family_101", name="Michael Johnson", relationship="son", phone="555-333-4444", email="michael.johnson@email.com", role="primary_caregiver", notification_preferences={ "urgent_alerts": True, "daily_summary": True, "medication_reminders": True, "health_trends": True } ), FamilyMember( id="family_102", name="Susan Johnson", relationship="daughter", phone="555-555-6666", email="susan.johnson@email.com", role="backup_caregiver", notification_preferences={ "urgent_alerts": True, "daily_summary": True, "medication_reminders": True, "health_trends": True } ) ] # Family 3: Elena Gonzalez (early Alzheimer's) elena = Parent( id="parent_003", name="Elena Gonzalez", age=81, conditions=["Early Alzheimer's", "Osteoporosis", "COPD"], medications=[ {"name": "Aricept", "dosage": "10mg", "frequency": "1x daily", "times": ["20:00"]}, {"name": "Alendronate", "dosage": "70mg", "frequency": "1x weekly", "times": ["08:00"]}, {"name": "Albuterol", "dosage": "90mcg", "frequency": "as needed", "times": ["08:00", "20:00"]} ], emergency_contacts=[ {"name": "Dr. Maria Lopez", "phone": "555-MED-9876", "type": "primary_care"}, {"name": "Sunset Medical Center", "phone": "555-911-4321", "type": "emergency"} ], preferences={ "preferred_contact_time": "morning", "communication_style": "simple", "daily_check_in_time": "10:00" } ) elena_family = [ FamilyMember( id="family_201", name="Carlos Gonzalez", relationship="son", phone="555-777-8888", email="carlos.gonzalez@email.com", role="primary_caregiver", notification_preferences={ "urgent_alerts": True, "daily_summary": True, "medication_reminders": True, "health_trends": True } ), FamilyMember( id="family_202", name="Isabella Martinez", relationship="granddaughter", phone="555-999-0000", email="isabella.martinez@email.com", role="support", notification_preferences={ "urgent_alerts": True, "daily_summary": False, "medication_reminders": False, "health_trends": True } ) ] # Store in parent dictionary self.parents = { "parent_001": margaret, "parent_002": robert, "parent_003": elena } # Store in family dictionary self.families = { "parent_001": margaret_family, "parent_002": robert_family, "parent_003": elena_family } # Generate health data for each parent for parent_id in self.parents: self.health_data[parent_id] = self.generate_health_timeline(parent_id) self.medication_data[parent_id] = self.generate_medication_timeline(parent_id) # For backward compatibility self.margaret = self.parents["parent_001"] self.margaret_family = self.families["parent_001"] def get_parent_by_id(self, parent_id: str) -> Optional[Parent]: """Get parent data by ID""" return self.parents.get(parent_id) def get_family_by_parent_id(self, parent_id: str) -> List[FamilyMember]: """Get family members for a specific parent""" return self.families.get(parent_id, []) def get_health_data_by_parent_id(self, parent_id: str) -> List[HealthMetric]: """Get health data for a specific parent""" return self.health_data.get(parent_id, []) def get_medication_data_by_parent_id(self, parent_id: str) -> List[Dict]: """Get medication data for a specific parent""" return self.medication_data.get(parent_id, []) def generate_health_timeline(self, parent_id: str) -> List[HealthMetric]: """Generate realistic health data for the past week""" metrics = [] base_date = datetime.now() - timedelta(days=7) parent = self.parents.get(parent_id) if not parent: return [] for day in range(7): current_date = base_date + timedelta(days=day) # Morning metrics morning = current_date.replace(hour=8, minute=0) # Customize metrics based on parent conditions if parent_id == "parent_001": # Margaret (diabetes, hypertension) metrics.extend([ HealthMetric(morning, "heart_rate", 72 + random.randint(-8, 8), "bpm", "apple_watch", (60, 90)), HealthMetric(morning, "blood_pressure_systolic", 135 + random.randint(-15, 15), "mmHg", "omron_cuff", (90, 140)), HealthMetric(morning, "blood_pressure_diastolic", 82 + random.randint(-10, 10), "mmHg", "omron_cuff", (60, 90)), HealthMetric(morning, "weight", 145.2 + random.uniform(-1, 1), "lbs", "smart_scale", (140, 150)), ]) # Blood glucose (Type 2 diabetes) for hour in [8, 12, 18]: glucose_time = current_date.replace(hour=hour, minute=random.randint(0, 30)) base_glucose = 140 if hour == 12 else 120 # Higher after lunch metrics.append( HealthMetric(glucose_time, "blood_glucose", base_glucose + random.randint(-20, 30), "mg/dL", "glucose_meter", (80, 130)) ) elif parent_id == "parent_002": # Robert (stroke recovery, AFib) metrics.extend([ HealthMetric(morning, "heart_rate", 78 + random.randint(-5, 12), "bpm", "apple_watch", (60, 90)), HealthMetric(morning, "blood_pressure_systolic", 142 + random.randint(-10, 20), "mmHg", "omron_cuff", (90, 140)), HealthMetric(morning, "blood_pressure_diastolic", 88 + random.randint(-5, 10), "mmHg", "omron_cuff", (60, 90)), HealthMetric(morning, "weight", 190.5 + random.uniform(-1, 1.5), "lbs", "smart_scale", (185, 195)), HealthMetric(morning, "heart_rhythm", 1 if random.random() < 0.2 else 0, "irregularity", "ecg_monitor", (0, 0)), ]) elif parent_id == "parent_003": # Elena (Alzheimer's, osteoporosis, COPD) metrics.extend([ HealthMetric(morning, "heart_rate", 75 + random.randint(-8, 10), "bpm", "apple_watch", (60, 90)), HealthMetric(morning, "blood_pressure_systolic", 128 + random.randint(-10, 15), "mmHg", "omron_cuff", (90, 140)), HealthMetric(morning, "blood_pressure_diastolic", 78 + random.randint(-8, 8), "mmHg", "omron_cuff", (60, 90)), HealthMetric(morning, "weight", 118.7 + random.uniform(-0.5, 0.5), "lbs", "smart_scale", (115, 120)), HealthMetric(morning, "oxygen_saturation", 94 + random.randint(-3, 2), "%", "pulse_oximeter", (95, 100)), ]) # Cognitive function test (for Alzheimer's) if day % 2 == 0: # Every other day test_time = current_date.replace(hour=11, minute=random.randint(0, 30)) metrics.append( HealthMetric(test_time, "cognitive_score", 78 + random.randint(-8, 5), "points", "cognitive_test", (85, 100)) ) # Daily activity metrics - common for all metrics.extend([ HealthMetric(current_date.replace(hour=23, minute=59), "steps", 2800 + random.randint(-800, 1200), "steps", "apple_watch", (2000, 8000)), HealthMetric(current_date.replace(hour=23, minute=59), "sleep_hours", 6.5 + random.uniform(-1, 1.5), "hours", "apple_watch", (6, 9)), ]) return sorted(metrics, key=lambda x: x.timestamp) def generate_medication_timeline(self, parent_id: str) -> List[Dict]: """Generate medication compliance data""" compliance_data = [] base_date = datetime.now() - timedelta(days=7) parent = self.parents.get(parent_id) if not parent: return [] for day in range(7): current_date = base_date + timedelta(days=day) for med in parent.medications: for time_str in med["times"]: hour, minute = map(int, time_str.split(':')) scheduled_time = current_date.replace(hour=hour, minute=minute) # Simulate realistic compliance patterns compliance_rate = 0.9 # Default rate if parent_id == "parent_001": # Margaret if med["name"] == "Metformin": compliance_rate = 0.95 if hour < 12 else 0.85 elif med["name"] == "Lisinopril": compliance_rate = 0.92 else: # Donepezil compliance_rate = 0.88 elif parent_id == "parent_002": # Robert if med["name"] == "Warfarin": compliance_rate = 0.97 # Critical medication else: compliance_rate = 0.9 elif parent_id == "parent_003": # Elena (Alzheimer's) # More forgetful due to cognitive issues compliance_rate = 0.75 if hour >= 18 else 0.85 was_taken = random.random() < compliance_rate actual_time = None if was_taken: # Add realistic delay (5-30 minutes) delay_minutes = random.randint(5, 30) actual_time = scheduled_time + timedelta(minutes=delay_minutes) compliance_data.append({ "medication": med["name"], "scheduled_time": scheduled_time, "actual_time": actual_time, "was_taken": was_taken, "source": "pill_dispenser" if was_taken else "missed" }) return compliance_data # ============== LANGGRAPH STATE DEFINITION ============== class CareState(TypedDict): parent_id: str date: str health_metrics: List[Dict] medication_status: List[Dict] concerns: List[str] alerts: List[Dict] daily_summary: str action_items: List[str] family_notifications: List[Dict] emergency_level: str # ============== SPECIALIZED CARE AGENTS ============== class HealthMonitorAgent: def __init__(self, llm): self.llm = llm self.mock_data = MockDataGenerator() def analyze_health_patterns(self, state: CareState) -> CareState: """Analyze health trends and identify concerns""" recent_metrics = [m for m in self.mock_data.health_data[state["parent_id"]] if m.timestamp >= datetime.now() - timedelta(days=3)] concerns = [] alerts = [] # Analyze blood glucose trends (diabetes management) glucose_readings = [m for m in recent_metrics if m.metric_type == "blood_glucose"] if glucose_readings: avg_glucose = sum(m.value for m in glucose_readings) / len(glucose_readings) if avg_glucose > 160: concerns.append("Blood glucose levels elevated - average 165 mg/dL over 3 days") alerts.append({ "type": "health_concern", "severity": AlertLevel.HIGH.value, "message": "Diabetes management needs attention - glucose levels trending high", "recommended_action": "Contact diabetes care team" }) # Analyze activity levels step_data = [m for m in recent_metrics if m.metric_type == "steps"] if step_data: avg_steps = sum(m.value for m in step_data) / len(step_data) if avg_steps < 2000: concerns.append(f"Low activity level - averaging {int(avg_steps)} steps/day") alerts.append({ "type": "activity_concern", "severity": AlertLevel.MEDIUM.value, "message": "Activity levels below recommended for health maintenance", "recommended_action": "Encourage gentle walks or physical therapy" }) # Analyze sleep patterns sleep_data = [m for m in recent_metrics if m.metric_type == "sleep_hours"] if sleep_data: avg_sleep = sum(m.value for m in sleep_data) / len(sleep_data) if avg_sleep < 6: concerns.append(f"Insufficient sleep - averaging {avg_sleep:.1f} hours/night") # Blood pressure monitoring bp_systolic = [m for m in recent_metrics if m.metric_type == "blood_pressure_systolic"] if bp_systolic: recent_bp = bp_systolic[-1].value if recent_bp > 150: concerns.append(f"Blood pressure elevated - last reading {recent_bp} mmHg") alerts.append({ "type": "vital_concern", "severity": AlertLevel.HIGH.value, "message": "Blood pressure significantly elevated", "recommended_action": "Schedule urgent medical appointment" }) state["health_metrics"] = [asdict(m) for m in recent_metrics] state["concerns"].extend(concerns) state["alerts"].extend(alerts) return state class MedicationAgent: def __init__(self, llm): self.llm = llm self.mock_data = MockDataGenerator() def check_medication_compliance(self, state: CareState) -> CareState: """Monitor medication adherence and identify issues""" recent_meds = [m for m in self.mock_data.medication_data[state["parent_id"]] if m["scheduled_time"] >= datetime.now() - timedelta(days=3)] # Calculate compliance rates compliance_summary = {} for med_record in recent_meds: med_name = med_record["medication"] if med_name not in compliance_summary: compliance_summary[med_name] = {"total": 0, "taken": 0, "missed_times": []} compliance_summary[med_name]["total"] += 1 if med_record["was_taken"]: compliance_summary[med_name]["taken"] += 1 else: compliance_summary[med_name]["missed_times"].append(med_record["scheduled_time"]) # Generate alerts for poor compliance alerts = [] concerns = [] for med_name, data in compliance_summary.items(): compliance_rate = data["taken"] / data["total"] if compliance_rate < 0.8: concerns.append(f"Poor medication compliance: {med_name} - {compliance_rate:.0%} over 3 days") alerts.append({ "type": "medication_compliance", "severity": AlertLevel.HIGH.value, "message": f"{med_name} missed multiple doses - compliance at {compliance_rate:.0%}", "recommended_action": "Review medication routine with family" }) elif compliance_rate < 0.9: concerns.append(f"Medication adherence concern: {med_name} - {compliance_rate:.0%}") # Check for recent missed doses recent_missed = [m for m in recent_meds if not m["was_taken"] and m["scheduled_time"] >= datetime.now() - timedelta(hours=24)] if recent_missed: for missed in recent_missed: alerts.append({ "type": "missed_medication", "severity": AlertLevel.MEDIUM.value, "message": f"Missed {missed['medication']} at {missed['scheduled_time'].strftime('%I:%M %p')}", "recommended_action": "Remind about missed dose" }) state["medication_status"] = recent_meds state["concerns"].extend(concerns) state["alerts"].extend(alerts) return state class FamilyCommunicationAgent: def __init__(self, llm): self.llm = llm self.mock_data = MockDataGenerator() def generate_family_updates(self, state: CareState) -> CareState: """Create personalized updates for each family member""" notifications = [] # Get urgency level alert_levels = [alert["severity"] for alert in state["alerts"]] max_severity = AlertLevel.LOW.value if AlertLevel.URGENT.value in alert_levels: max_severity = AlertLevel.URGENT.value elif AlertLevel.HIGH.value in alert_levels: max_severity = AlertLevel.HIGH.value elif AlertLevel.MEDIUM.value in alert_levels: max_severity = AlertLevel.MEDIUM.value # Generate notifications for each family member for family_member in self.mock_data.get_family_by_parent_id(state["parent_id"]): prefs = family_member.notification_preferences # Determine what to include based on preferences and severity should_send_immediate = False message_parts = [] if max_severity in [AlertLevel.URGENT.value, AlertLevel.HIGH.value] and prefs["urgent_alerts"]: should_send_immediate = True urgent_alerts = [a for a in state["alerts"] if a["severity"] in [AlertLevel.URGENT.value, AlertLevel.HIGH.value]] message_parts.append("šŸšØ **Urgent Update Needed**") for alert in urgent_alerts[:2]: # Limit to top 2 urgent items message_parts.append(f"ā€¢ {alert['message']}") if prefs["daily_summary"]: message_parts.append(f"\nšŸ“‹ **Daily Summary for {self.mock_data.get_parent_by_id(state['parent_id']).name}**") # Health highlights if state["health_metrics"]: latest_metrics = {} for metric in state["health_metrics"]: if metric["metric_type"] not in latest_metrics or metric["timestamp"] > latest_metrics[metric["metric_type"]]["timestamp"]: latest_metrics[metric["metric_type"]] = metric message_parts.append("šŸ“Š **Today's Health Data:**") if "steps" in latest_metrics: steps = int(latest_metrics["steps"]["value"]) message_parts.append(f"ā€¢ Activity: {steps:,} steps") if "blood_glucose" in latest_metrics: glucose = latest_metrics["blood_glucose"]["value"] message_parts.append(f"ā€¢ Blood Sugar: {glucose} mg/dL") # Medication status if prefs["medication_reminders"] and state["medication_status"]: today_meds = [m for m in state["medication_status"] if m["scheduled_time"].date() == datetime.now().date()] taken_count = sum(1 for m in today_meds if m["was_taken"]) total_count = len(today_meds) message_parts.append(f"šŸ’Š **Medications:** {taken_count}/{total_count} doses taken today") if message_parts: notifications.append({ "recipient": family_member.name, "recipient_id": family_member.id, "phone": family_member.phone, "email": family_member.email, "message": "\n".join(message_parts), "urgency": max_severity, "send_immediately": should_send_immediate, "channels": ["sms"] if should_send_immediate else ["email"] }) state["family_notifications"] = notifications return state def create_daily_summary(self, state: CareState) -> CareState: """Generate comprehensive daily summary""" summary_parts = [] # Header with date and overall status today = datetime.now().strftime("%A, %B %d, %Y") summary_parts.append(f"# Daily Care Report - {today}") summary_parts.append(f"**Parent:** {self.mock_data.get_parent_by_id(state['parent_id']).name}") # Overall status indicator if any(alert["severity"] == AlertLevel.URGENT.value for alert in state["alerts"]): summary_parts.append("šŸ”“ **Status: Needs Immediate Attention**") elif any(alert["severity"] == AlertLevel.HIGH.value for alert in state["alerts"]): summary_parts.append("šŸŸ” **Status: Some Concerns**") elif state["concerns"]: summary_parts.append("šŸŸ¢ **Status: Monitoring**") else: summary_parts.append("āœ… **Status: All Good**") # Key metrics summary summary_parts.append("\n## Today's Highlights") if state["health_metrics"]: # Get today's key metrics today_metrics = [m for m in state["health_metrics"] if m["timestamp"].date() == datetime.now().date()] if today_metrics: latest_by_type = {} for metric in today_metrics: metric_type = metric["metric_type"] if metric_type not in latest_by_type: latest_by_type[metric_type] = metric summary_parts.append("### Health Metrics") for metric_type, data in latest_by_type.items(): value = data["value"] unit = data["unit"] if metric_type == "steps": summary_parts.append(f"ā€¢ **Activity:** {int(value):,} {unit}") elif metric_type == "blood_glucose": status = "šŸ”“ High" if value > 150 else "šŸŸ” Elevated" if value > 130 else "āœ… Normal" summary_parts.append(f"ā€¢ **Blood Sugar:** {value} {unit} {status}") elif metric_type == "blood_pressure_systolic": status = "šŸ”“ High" if value > 140 else "šŸŸ” Elevated" if value > 130 else "āœ… Normal" summary_parts.append(f"ā€¢ **Blood Pressure:** {value} {unit} {status}") # Medication summary if state["medication_status"]: today_meds = [m for m in state["medication_status"] if m["scheduled_time"].date() == datetime.now().date()] if today_meds: taken_count = sum(1 for m in today_meds if m["was_taken"]) total_count = len(today_meds) summary_parts.append(f"\n### Medication Compliance") summary_parts.append(f"ā€¢ **Today:** {taken_count}/{total_count} doses taken") if taken_count < total_count: missed = [m for m in today_meds if not m["was_taken"]] summary_parts.append("ā€¢ **Missed doses:**") for m in missed: time_str = m["scheduled_time"].strftime("%I:%M %p") summary_parts.append(f" - {m['medication']} at {time_str}") # Concerns and alerts if state["concerns"]: summary_parts.append(f"\n## Areas of Attention ({len(state['concerns'])})") for concern in state["concerns"]: summary_parts.append(f"ā€¢ {concern}") # Action items if state["action_items"]: summary_parts.append(f"\n## Recommended Actions") for action in state["action_items"]: summary_parts.append(f"ā€¢ {action}") state["daily_summary"] = "\n".join(summary_parts) return state class ActionPlannerAgent: def __init__(self, llm): self.llm = llm def generate_action_items(self, state: CareState) -> CareState: """Generate prioritized action items based on concerns and alerts""" actions = [] # Process alerts to create specific actions urgent_alerts = [a for a in state["alerts"] if a["severity"] == AlertLevel.URGENT.value] high_alerts = [a for a in state["alerts"] if a["severity"] == AlertLevel.HIGH.value] # Urgent actions first for alert in urgent_alerts: if alert["type"] == "vital_concern": actions.append("šŸšØ URGENT: Schedule emergency medical appointment for blood pressure management") elif alert["type"] == "medication_compliance": actions.append(f"šŸšØ URGENT: Call parent immediately about missed {alert['message'].split()[0]} medications") # High priority actions for alert in high_alerts: if alert["type"] == "health_concern": actions.append("šŸ“ž HIGH: Contact diabetes care team about elevated glucose levels") elif alert["type"] == "medication_compliance": actions.append("šŸ“ž HIGH: Review medication routine - consider pill dispenser or reminder system") # Medium priority actions based on concerns concern_keywords = { "low activity": "šŸ’Ŗ Encourage gentle exercise - suggest short walks or chair exercises", "insufficient sleep": "šŸ˜“ Discuss sleep hygiene - check for pain or anxiety issues", "blood pressure": "šŸ©ŗ Monitor blood pressure daily and log readings for doctor", "medication": "šŸ’Š Set up automated medication reminders or family check-ins" } for concern in state["concerns"]: concern_lower = concern.lower() for keyword, action in concern_keywords.items(): if keyword in concern_lower and action not in actions: actions.append(action) # Always include routine actions routine_actions = [ "šŸ“‹ Review tomorrow's medication schedule", "šŸ“± Confirm evening check-in call is scheduled", "šŸ“Š Check health metrics sync from devices" ] # Only add routine actions if no urgent/high priority actions if not urgent_alerts and not high_alerts: actions.extend(routine_actions) # Limit to top 5 actions for manageability state["action_items"] = actions[:5] return state # ============== MAIN LANGGRAPH WORKFLOW ============== class CareLoopOrchestrator: def __init__(self): # For demo purposes, we'll mock the LLM self.llm = None # ChatOpenAI would go here in production # Initialize mock data self.mock_data = MockDataGenerator() # Initialize specialized agents self.health_monitor = HealthMonitorAgent(self.llm) self.medication_agent = MedicationAgent(self.llm) self.family_communicator = FamilyCommunicationAgent(self.llm) self.action_planner = ActionPlannerAgent(self.llm) # Build the workflow graph self.graph = self._build_workflow() def _build_workflow(self) -> StateGraph: """Build the LangGraph workflow for daily care monitoring""" workflow = StateGraph(CareState) # Add nodes for each step workflow.add_node("health_analysis", self._health_analysis_node) workflow.add_node("medication_check", self._medication_check_node) workflow.add_node("generate_summary", self._generate_summary_node) workflow.add_node("send_family_notifications", self._family_notifications_node) workflow.add_node("action_planning", self._action_planning_node) workflow.add_node("emergency_check", self._emergency_check_node) # Define the workflow workflow.set_entry_point("health_analysis") workflow.add_edge("health_analysis", "medication_check") workflow.add_edge("medication_check", "emergency_check") workflow.add_conditional_edges( "emergency_check", self._should_trigger_emergency, { "emergency": "send_family_notifications", "continue": "generate_summary" } ) workflow.add_edge("generate_summary", "send_family_notifications") workflow.add_edge("send_family_notifications", "action_planning") workflow.add_edge("action_planning", END) return workflow.compile() def _health_analysis_node(self, state: CareState) -> CareState: return self.health_monitor.analyze_health_patterns(state) def _medication_check_node(self, state: CareState) -> CareState: return self.medication_agent.check_medication_compliance(state) def _generate_summary_node(self, state: CareState) -> CareState: return self.family_communicator.create_daily_summary(state) def _family_notifications_node(self, state: CareState) -> CareState: return self.family_communicator.generate_family_updates(state) def _action_planning_node(self, state: CareState) -> CareState: return self.action_planner.generate_action_items(state) def _emergency_check_node(self, state: CareState) -> CareState: """Check if emergency response is needed""" urgent_alerts = [a for a in state["alerts"] if a["severity"] == AlertLevel.URGENT.value] if urgent_alerts: state["emergency_level"] = "urgent" else: state["emergency_level"] = "normal" return state def _should_trigger_emergency(self, state: CareState) -> str: """Decide whether to trigger emergency protocol""" return "emergency" if state["emergency_level"] == "urgent" else "continue" def run_daily_care_cycle(self, parent_id: str) -> Dict[str, Any]: """Execute the complete daily care monitoring workflow""" # Validate parent_id exists if parent_id not in self.mock_data.parents: raise ValueError(f"Parent ID {parent_id} not found") initial_state = { "parent_id": parent_id, "date": datetime.now().strftime("%Y-%m-%d"), "health_metrics": [], "medication_status": [], "concerns": [], "alerts": [], "daily_summary": "", "action_items": [], "family_notifications": [], "emergency_level": "normal" } # Run the workflow result = self.graph.invoke(initial_state) # Add timestamp and processing info result["processed_at"] = datetime.now().isoformat() result["next_check"] = (datetime.now() + timedelta(hours=24)).isoformat() return result # ============== WEB INTERFACE ============== # Initialize FastAPI app app = FastAPI(title="CareLoop - AI-Powered Family Caregiving") # Initialize the care orchestrator care_system = CareLoopOrchestrator() # Store for demo data demo_reports = [] @app.get("/", response_class=HTMLResponse) async def get_dashboard(): """Serve the main dashboard""" html_content = """ CareLoop Dashboard

šŸ  CareLoop

AI-Powered Family Caregiving

Hackathon Demo

Health Conditions:

Family Caregivers:

šŸ“Š Care Status Overview

Health Metrics

Medication Status

Alerts Generated

šŸ“‹ Daily Summary


                        

                    

                    
                    
                    

                        
āœ… Recommended Actions

                        
    
                            
                        

                    

                    
                    
                    

                        
šŸ“± Family Notifications

                        

                            
                        

                    

                    
                    
                    

                        
āš ļø Care Alerts

                        

                            
                        

                    

                

            

        

        
        
    
    
    """
    return HTMLResponse(content=html_content)

@app.post("/api/care-check/{parent_id}")
async def run_care_check(parent_id: str):
    """Run the daily care check for a parent"""
    try:
        result = care_system.run_daily_care_cycle(parent_id)
        demo_reports.append(result)
        return result
    except Exception as e:
        return {"error": str(e)}

@app.get("/api/reports")
async def get_reports():
    """Get all demo reports"""
    return {"reports": demo_reports}

@app.get("/api/parent/{parent_id}")
async def get_parent_info(parent_id: str):
    """Get parent information"""
    mock_data = MockDataGenerator()
    return {
        "parent": asdict(mock_data.get_parent_by_id(parent_id)),
        "family": [asdict(fm) for fm in mock_data.get_family_by_parent_id(parent_id)]
    }

# ============== MAIN EXECUTION ==============

if __name__ == "__main__":
    print("šŸš€ Starting CareLoop Hackathon Demo")
    print("=" * 50)
    
    # Load mock data
    mock_data = MockDataGenerator()
    
    # Display available parents
    print("šŸ  Available Parent Profiles:")
    for parent_id, parent in mock_data.parents.items():
        family_count = len(mock_data.families.get(parent_id, []))
        conditions = ", ".join(parent.conditions[:2]) + ("..." if len(parent.conditions) > 2 else "")
        print(f"  ā€¢ {parent.name} ({parent.age}) - ID: {parent_id}")
        print(f"    Health: {conditions}")
        print(f"    Family caregivers: {family_count}")
    
    # Run a sample care check for demo
    print("\nšŸ”„ Running sample care analysis for Margaret Chen...")
    result = care_system.run_daily_care_cycle("parent_001")
    
    print(f"\nšŸ“‹ Daily Summary:")
    print(result["daily_summary"])
    
    print(f"\nāœ… Action Items ({len(result['action_items'])}):")
    for action in result["action_items"]:
        print(f"  ā€¢ {action}")
    
    print(f"\nšŸ“± Family Notifications ({len(result['family_notifications'])}):")
    for notification in result["family_notifications"]:
        print(f"  ā†’ {notification['recipient']}: {notification['urgency']} priority")
    
    print(f"\nāš ļø Alerts Generated ({len(result['alerts'])}):")
    for alert in result["alerts"]:
        print(f"  ā€¢ {alert['severity'].upper()}: {alert['message']}")
    
    print("\n" + "=" * 50)
    print("šŸŒ Starting web interface...")
    print("Open your browser to: http://localhost:8000")
    print("Select a parent and click 'Run Daily Care Check' to see the AI agents in action!")
    
    # Start the web server
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")