Upload 14 files

prompts/__init__.py

@@ -0,0 +1,18 @@
+"""
+Prompts module for CareFlow Nexus AI Agents
+Contains all prompt templates for Gemini AI interactions
+"""
+
+from .prompt_templates import (
+    BedAllocatorPrompts,
+    CommonPrompts,
+    StateManagerPrompts,
+    TaskCoordinatorPrompts,
+)
+
+__all__ = [
+    "StateManagerPrompts",
+    "BedAllocatorPrompts",
+    "TaskCoordinatorPrompts",
+    "CommonPrompts",
+]

prompts/prompt_templates.py

@@ -0,0 +1,391 @@
+"""
+Prompt Templates for CareFlow Nexus AI Agents
+Contains all prompt templates for State Manager, Bed Allocator, and Task Coordinator agents
+"""
+
+
+class StateManagerPrompts:
+    """Prompt templates for State Manager Agent"""
+
+    STATE_ANALYSIS = """
+You are a State Manager AI for a hospital bed management system.
+
+CURRENT HOSPITAL STATE:
+
+Beds:
+- Total Beds: {total_beds}
+- Available: {available_beds}
+- Occupied: {occupied_beds}
+- In Cleaning: {cleaning_beds}
+- In Maintenance: {maintenance_beds}
+- Utilization Rate: {utilization_rate}%
+
+Patients:
+- Total Patients: {total_patients}
+- Waiting for Admission: {waiting_patients}
+- Currently Admitted: {admitted_patients}
+
+Staff:
+- Nurses On Shift: {nurses_count}
+- Cleaners On Shift: {cleaners_count}
+- Total Staff Available: {total_staff}
+
+Tasks:
+- Active Tasks: {active_tasks}
+- Pending Tasks: {pending_tasks}
+- In Progress Tasks: {in_progress_tasks}
+- Overdue Tasks: {overdue_tasks}
+
+Ward Breakdown:
+{ward_summary}
+
+TASK:
+Analyze this hospital state and identify:
+1. Critical issues requiring immediate attention
+2. Operational bottlenecks
+3. Capacity forecast for next 4-6 hours
+4. Proactive recommendations
+
+Respond ONLY with valid JSON in this exact format:
+{{
+  "critical_alerts": [
+    {{
+      "type": "alert type",
+      "severity": "critical/high/medium",
+      "message": "what's wrong",
+      "action_needed": "what to do"
+    }}
+  ],
+  "bottlenecks": [
+    {{
+      "area": "which area (cleaning, staffing, beds)",
+      "description": "issue description",
+      "impact": "how it affects operations",
+      "recommendation": "suggested solution"
+    }}
+  ],
+  "capacity_forecast": {{
+    "next_4_hours": "forecast description",
+    "bed_availability_trend": "increasing/stable/decreasing",
+    "staffing_adequacy": "sufficient/stretched/insufficient"
+  }},
+  "recommendations": [
+    "proactive action 1",
+    "proactive action 2"
+  ]
+}}
+"""
+
+    BOTTLENECK_DETECTION = """
+Analyze the following hospital operations data for bottlenecks:
+
+{operational_data}
+
+Identify specific bottlenecks and provide actionable recommendations.
+
+Respond with JSON:
+{{
+  "bottlenecks": [
+    {{
+      "type": "cleaning_backlog/staff_overload/bed_shortage",
+      "severity": "low/medium/high/critical",
+      "count": 0,
+      "description": "detailed issue",
+      "recommendation": "action to take"
+    }}
+  ]
+}}
+"""
+
+
+class BedAllocatorPrompts:
+    """Prompt templates for Bed Allocator Agent"""
+
+    BED_ALLOCATION = """
+You are a Bed Allocator AI for a hospital. Your job is to match patients with the most suitable beds.
+
+PATIENT INFORMATION:
+- Name: {patient_name}
+- Age: {age}
+- Gender: {gender}
+- Diagnosis: {diagnosis}
+- Severity: {severity}
+- Mobility Status: {mobility_status}
+
+EXTRACTED REQUIREMENTS:
+- Needs Oxygen: {needs_oxygen}
+- Needs Ventilator: {needs_ventilator}
+- Needs Cardiac Monitor: {needs_cardiac_monitor}
+- Needs Isolation: {needs_isolation}
+- Preferred Ward: {preferred_ward}
+
+AVAILABLE BEDS (Pre-filtered by requirements):
+{beds_json}
+
+CURRENT CONTEXT:
+- Time of Day: {current_time}
+- Day of Week: {day_of_week}
+- Overall Hospital Occupancy: {occupancy_rate}%
+- Staff Availability: {staff_summary}
+
+SCORING CRITERIA:
+1. Medical Appropriateness (40 points):
+   - Equipment match (oxygen, ventilator, monitors)
+   - Isolation capability if needed
+   - Ward specialization
+
+2. Patient Safety & Comfort (25 points):
+   - Proximity to nursing station for monitoring
+   - Appropriate ward environment
+   - Infection control considerations
+
+3. Operational Efficiency (20 points):
+   - Current ward workload distribution
+   - Staff availability in ward
+   - Bed location logistics
+
+4. Resource Optimization (15 points):
+   - Avoid over-specification (don't use ICU for minor cases)
+   - Balance ward occupancy
+   - Equipment availability vs future needs
+
+TASK:
+Rank the available beds and recommend the TOP 3 most suitable options.
+For each bed, provide:
+- Match score (0-100)
+- Detailed reasoning
+- Pros (advantages)
+- Cons (concerns or limitations)
+
+Respond ONLY with valid JSON in this exact format:
+{{
+  "recommendations": [
+    {{
+      "bed_id": "bed_id_here",
+      "bed_number": "bed number",
+      "ward": "ward name",
+      "score": 85,
+      "reasoning": "This bed is ideal because it has oxygen equipment which is required for the pneumonia patient. It's located in the Respiratory ward with specialized staff and has high proximity to nursing station (8/10) for close monitoring.",
+      "pros": [
+        "Has required oxygen equipment",
+        "In specialized Respiratory ward",
+        "Close to nursing station for monitoring",
+        "Currently available and clean"
+      ],
+      "cons": [
+        "Ward is at 75% capacity - relatively busy",
+        "No cardiac monitor (not required but could be useful)"
+      ]
+    }}
+  ],
+  "overall_confidence": 90,
+  "considerations": "Patient has moderate severity pneumonia requiring oxygen support and close monitoring. All recommended beds meet core requirements."
+}}
+"""
+
+    REQUIREMENT_EXTRACTION = """
+You are a medical requirements analyzer for hospital bed allocation.
+
+PATIENT DATA:
+- Age: {age}
+- Gender: {gender}
+- Diagnosis: {diagnosis}
+- Severity: {severity}
+- Admission Type: {admission_type}
+- Mobility Status: {mobility_status}
+
+TASK:
+Extract the medical care requirements needed for this patient.
+
+Consider:
+- What equipment is needed? (oxygen, ventilator, cardiac monitor)
+- Is isolation required? (infectious diseases, immunocompromised)
+- What ward is most appropriate? (ICU, General, Isolation, Cardiac, Respiratory)
+- How much nursing attention is needed? (proximity to nursing station: 1-10)
+- Any special considerations?
+
+Respond ONLY with valid JSON:
+{{
+  "needs_oxygen": true,
+  "needs_ventilator": false,
+  "needs_cardiac_monitor": false,
+  "needs_isolation": true,
+  "preferred_ward": "Respiratory",
+  "proximity_preference": 8,
+  "special_considerations": [
+    "Patient needs close monitoring due to moderate severity",
+    "Infectious precautions required"
+  ],
+  "confidence": 95,
+  "reasoning": "Pneumonia diagnosis requires oxygen support and isolation to prevent spread. Moderate severity indicates need for close monitoring."
+}}
+"""
+
+
+class TaskCoordinatorPrompts:
+    """Prompt templates for Task Coordinator Agent"""
+
+    STAFF_ASSIGNMENT = """
+You are a Task Coordinator AI for hospital operations.
+
+TASK TO ASSIGN:
+- Task ID: {task_id}
+- Type: {task_type}
+- Description: {description}
+- Priority: {priority}
+- Location: Ward {ward}, Bed {bed_number}
+- Estimated Duration: {duration} minutes
+- Patient: {patient_name} (if applicable)
+
+AVAILABLE STAFF CANDIDATES:
+{staff_json}
+
+CURRENT CONTEXT:
+- Current Time: {current_time}
+- Ward Activity Level: {activity_level}
+- Total Pending Tasks: {pending_tasks_count}
+
+SELECTION CRITERIA:
+1. Role Appropriateness (Must Match): {required_role}
+2. Current Workload (Fair Distribution)
+3. Ward Assignment (Preference for same ward)
+4. Location Proximity (Efficiency)
+5. Recent Task History (Avoid overloading)
+
+TASK:
+Select the most appropriate staff member for this task.
+
+Consider:
+- Who has the lowest current workload?
+- Who is already in or near this ward?
+- Who hasn't been assigned a task recently?
+- Balance efficiency with fairness
+
+Respond ONLY with valid JSON:
+{{
+  "recommended_staff_id": "staff_id_here",
+  "staff_name": "Staff Name",
+  "reasoning": "This staff member has the lowest current workload (2 tasks) and is already assigned to the same ward, making them the most efficient choice. They have the appropriate role and capacity.",
+  "workload_impact": "Workload will increase from 2 to 3 tasks, still below maximum of 5.",
+  "concerns": [
+    "Ward is busy - staff may need support"
+  ],
+  "alternatives": [
+    {{
+      "staff_id": "alt_id",
+      "staff_name": "Alt Name",
+      "reason": "Second choice with 3 current tasks"
+    }}
+  ],
+  "confidence": 85
+}}
+"""
+
+    WORKFLOW_ORCHESTRATION = """
+You are orchestrating a multi-step hospital workflow.
+
+WORKFLOW TYPE: {workflow_type}
+
+CONTEXT:
+{context_json}
+
+CURRENT STEP: {current_step}
+PREVIOUS STEPS COMPLETED: {completed_steps}
+
+AVAILABLE STAFF:
+- Nurses: {nurses_available}
+- Cleaners: {cleaners_available}
+
+TASK:
+Determine the next task(s) to create and assign.
+
+Respond with JSON:
+{{
+  "next_tasks": [
+    {{
+      "task_type": "cleaning",
+      "priority": "high",
+      "assigned_role": "cleaner",
+      "description": "Task description",
+      "estimated_duration": 30
+    }}
+  ],
+  "workflow_status": "in_progress/completed",
+  "reasoning": "Why these tasks next"
+}}
+"""
+
+    TASK_ESCALATION = """
+You are handling a delayed hospital task that needs escalation.
+
+TASK DETAILS:
+{task_json}
+
+DELAY INFORMATION:
+- Expected Duration: {expected_duration} minutes
+- Actual Time Elapsed: {actual_elapsed} minutes
+- Delay: {delay_minutes} minutes
+- Current Status: {status}
+
+CONTEXT:
+- Ward: {ward}
+- Priority: {priority}
+- Patient Waiting: {patient_waiting}
+
+AVAILABLE OPTIONS:
+1. Reassign to different staff
+2. Escalate to supervisor
+3. Increase priority
+4. Request additional support
+
+Available staff for reassignment:
+{available_staff_json}
+
+TASK:
+Recommend the best course of action to resolve this delay.
+
+Respond with JSON:
+{{
+  "action": "reassign/escalate/increase_priority/request_support",
+  "reasoning": "Why this action is appropriate",
+  "recommended_staff_id": "staff_id if reassigning",
+  "escalation_message": "Message to supervisor if escalating",
+  "priority_change": "new priority if increasing",
+  "urgency": "low/medium/high/critical"
+}}
+"""
+
+
+class CommonPrompts:
+    """Common prompts used across multiple agents"""
+
+    GENERATE_REASONING = """
+Explain the following decision in simple, clear language:
+
+DECISION: {decision}
+
+FACTORS CONSIDERED:
+{factors_json}
+
+Provide a 2-3 sentence explanation suitable for hospital staff.
+"""
+
+    VALIDATE_DECISION = """
+Validate this decision for potential issues:
+
+{decision_json}
+
+Check for:
+- Safety concerns
+- Resource conflicts
+- Policy violations
+- Logic errors
+
+Respond with JSON:
+{{
+  "is_valid": true/false,
+  "concerns": ["list of issues if any"],
+  "severity": "none/low/medium/high",
+  "recommendation": "proceed/review/reject"
+}}
+"""

services/__init__.py

@@ -0,0 +1,9 @@
+"""
+Services module for CareFlow Nexus AI Agents
+Provides Firebase and Gemini AI service integrations
+"""
+
+from .firebase_service import FirebaseService
+from .gemini_service import GeminiService
+
+__all__ = ["FirebaseService", "GeminiService"]

services/firebase_service.py

@@ -0,0 +1,515 @@
+"""
+Firebase Service for CareFlow Nexus
+Handles all Firebase Firestore operations for beds, patients, staff, and tasks
+"""
+
+import logging
+import os
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+import firebase_admin
+from firebase_admin import credentials, firestore
+from google.cloud.firestore_v1.base_query import FieldFilter
+
+logger = logging.getLogger(__name__)
+
+
+class FirebaseService:
+    """Service class for all Firebase Firestore operations"""
+
+    def __init__(self, service_account_path: str = None):
+        """
+        Initialize Firebase service
+
+        Args:
+            service_account_path: Path to Firebase service account JSON file
+        """
+        try:
+            if not firebase_admin._apps:
+                if service_account_path and os.path.exists(service_account_path):
+                    cred = credentials.Certificate(service_account_path)
+                    firebase_admin.initialize_app(cred)
+                else:
+                    # Use default credentials
+                    firebase_admin.initialize_app()
+
+            self.db = firestore.client()
+            logger.info("Firebase service initialized successfully")
+        except Exception as e:
+            logger.error(f"Failed to initialize Firebase: {e}")
+            raise
+
+    # ==================== BED OPERATIONS ====================
+
+    async def get_bed(self, bed_id: str) -> Optional[Dict]:
+        """Get a specific bed by ID"""
+        try:
+            doc = self.db.collection("beds").document(bed_id).get()
+            if doc.exists:
+                data = doc.to_dict()
+                data["id"] = doc.id
+                return data
+            return None
+        except Exception as e:
+            logger.error(f"Error getting bed {bed_id}: {e}")
+            return None
+
+    async def get_all_beds(self, filters: Optional[Dict] = None) -> List[Dict]:
+        """
+        Get all beds with optional filters
+
+        Args:
+            filters: Dict with keys like 'status', 'ward', 'has_oxygen', etc.
+        """
+        try:
+            query = self.db.collection("beds")
+
+            if filters:
+                if "status" in filters:
+                    query = query.where(
+                        filter=FieldFilter("status", "==", filters["status"])
+                    )
+                if "ward" in filters:
+                    query = query.where(
+                        filter=FieldFilter("ward", "==", filters["ward"])
+                    )
+
+            docs = query.stream()
+            beds = []
+            for doc in docs:
+                data = doc.to_dict()
+                data["id"] = doc.id
+
+                # Apply equipment filters (not indexed in Firestore)
+                if filters:
+                    equipment = data.get("equipment", {})
+                    if (
+                        "has_oxygen" in filters
+                        and equipment.get("has_oxygen") != filters["has_oxygen"]
+                    ):
+                        continue
+                    if (
+                        "has_ventilator" in filters
+                        and equipment.get("has_ventilator") != filters["has_ventilator"]
+                    ):
+                        continue
+                    if (
+                        "is_isolation" in filters
+                        and equipment.get("is_isolation") != filters["is_isolation"]
+                    ):
+                        continue
+
+                beds.append(data)
+
+            logger.info(f"Retrieved {len(beds)} beds")
+            return beds
+        except Exception as e:
+            logger.error(f"Error getting beds: {e}")
+            return []
+
+    async def get_available_beds(self, filters: Optional[Dict] = None) -> List[Dict]:
+        """Get all beds with status 'ready'"""
+        filter_dict = filters or {}
+        filter_dict["status"] = "ready"
+        return await self.get_all_beds(filter_dict)
+
+    async def update_bed_status(
+        self, bed_id: str, status: str, notes: Optional[str] = None
+    ) -> bool:
+        """
+        Update bed status
+
+        Args:
+            bed_id: Bed document ID
+            status: New status (ready, reserved, occupied, cleaning, maintenance)
+            notes: Optional notes
+        """
+        try:
+            update_data = {"status": status, "last_updated": firestore.SERVER_TIMESTAMP}
+            if notes:
+                update_data["notes"] = notes
+
+            self.db.collection("beds").document(bed_id).update(update_data)
+            logger.info(f"Updated bed {bed_id} status to {status}")
+            return True
+        except Exception as e:
+            logger.error(f"Error updating bed status: {e}")
+            return False
+
+    async def assign_bed_to_patient(self, bed_id: str, patient_id: str) -> bool:
+        """Assign a bed to a patient"""
+        try:
+            self.db.collection("beds").document(bed_id).update(
+                {
+                    "assigned_patient_id": patient_id,
+                    "status": "reserved",
+                    "last_updated": firestore.SERVER_TIMESTAMP,
+                }
+            )
+
+            self.db.collection("patients").document(patient_id).update(
+                {
+                    "assigned_bed_id": bed_id,
+                    "status": "admitted",
+                    "admission_time": firestore.SERVER_TIMESTAMP,
+                }
+            )
+
+            logger.info(f"Assigned bed {bed_id} to patient {patient_id}")
+            return True
+        except Exception as e:
+            logger.error(f"Error assigning bed: {e}")
+            return False
+
+    # ==================== PATIENT OPERATIONS ====================
+
+    async def get_patient(self, patient_id: str) -> Optional[Dict]:
+        """Get a specific patient by ID"""
+        try:
+            doc = self.db.collection("patients").document(patient_id).get()
+            if doc.exists:
+                data = doc.to_dict()
+                data["id"] = doc.id
+                return data
+            return None
+        except Exception as e:
+            logger.error(f"Error getting patient {patient_id}: {e}")
+            return None
+
+    async def get_all_patients(self, filters: Optional[Dict] = None) -> List[Dict]:
+        """Get all patients with optional filters"""
+        try:
+            query = self.db.collection("patients")
+
+            if filters and "status" in filters:
+                query = query.where(
+                    filter=FieldFilter("status", "==", filters["status"])
+                )
+
+            docs = query.stream()
+            patients = []
+            for doc in docs:
+                data = doc.to_dict()
+                data["id"] = doc.id
+                patients.append(data)
+
+            logger.info(f"Retrieved {len(patients)} patients")
+            return patients
+        except Exception as e:
+            logger.error(f"Error getting patients: {e}")
+            return []
+
+    async def create_patient(self, patient_data: Dict) -> Optional[str]:
+        """Create a new patient record"""
+        try:
+            patient_data["created_at"] = firestore.SERVER_TIMESTAMP
+            patient_data["status"] = patient_data.get("status", "waiting")
+
+            doc_ref = self.db.collection("patients").add(patient_data)
+            patient_id = doc_ref[1].id
+
+            logger.info(f"Created patient with ID: {patient_id}")
+            return patient_id
+        except Exception as e:
+            logger.error(f"Error creating patient: {e}")
+            return None
+
+    async def update_patient(self, patient_id: str, updates: Dict) -> bool:
+        """Update patient information"""
+        try:
+            updates["updated_at"] = firestore.SERVER_TIMESTAMP
+            self.db.collection("patients").document(patient_id).update(updates)
+            logger.info(f"Updated patient {patient_id}")
+            return True
+        except Exception as e:
+            logger.error(f"Error updating patient: {e}")
+            return False
+
+    # ==================== STAFF OPERATIONS ====================
+
+    async def get_staff(self, staff_id: str) -> Optional[Dict]:
+        """Get a specific staff member by ID"""
+        try:
+            doc = self.db.collection("staff").document(staff_id).get()
+            if doc.exists:
+                data = doc.to_dict()
+                data["id"] = doc.id
+                return data
+            return None
+        except Exception as e:
+            logger.error(f"Error getting staff {staff_id}: {e}")
+            return None
+
+    async def get_all_staff(self, filters: Optional[Dict] = None) -> List[Dict]:
+        """Get all staff with optional filters"""
+        try:
+            query = self.db.collection("staff")
+
+            if filters:
+                if "role" in filters:
+                    query = query.where(
+                        filter=FieldFilter("role", "==", filters["role"])
+                    )
+                if "is_on_shift" in filters:
+                    query = query.where(
+                        filter=FieldFilter("is_on_shift", "==", filters["is_on_shift"])
+                    )
+
+            docs = query.stream()
+            staff_list = []
+            for doc in docs:
+                data = doc.to_dict()
+                data["id"] = doc.id
+
+                # Apply ward filter if specified
+                if filters and "assigned_ward" in filters:
+                    if data.get("assigned_ward") != filters["assigned_ward"]:
+                        continue
+
+                staff_list.append(data)
+
+            logger.info(f"Retrieved {len(staff_list)} staff members")
+            return staff_list
+        except Exception as e:
+            logger.error(f"Error getting staff: {e}")
+            return []
+
+    async def get_available_staff(
+        self, role: str, ward: Optional[str] = None, max_workload: int = 5
+    ) -> List[Dict]:
+        """Get available staff by role with workload filtering"""
+        filters = {"role": role, "is_on_shift": True}
+        if ward:
+            filters["assigned_ward"] = ward
+
+        staff_list = await self.get_all_staff(filters)
+
+        # Filter by workload
+        available = [s for s in staff_list if s.get("current_load", 0) < max_workload]
+
+        # Sort by current load (least busy first)
+        available.sort(key=lambda x: x.get("current_load", 0))
+
+        return available
+
+    async def get_staff_workload(self, staff_id: str) -> int:
+        """Get current workload count for a staff member"""
+        try:
+            doc = self.db.collection("staff").document(staff_id).get()
+            if doc.exists:
+                return doc.to_dict().get("current_load", 0)
+            return 0
+        except Exception as e:
+            logger.error(f"Error getting staff workload: {e}")
+            return 0
+
+    async def update_staff_workload(self, staff_id: str, increment: int) -> bool:
+        """
+        Update staff workload
+
+        Args:
+            staff_id: Staff document ID
+            increment: Amount to increment (positive or negative)
+        """
+        try:
+            staff_ref = self.db.collection("staff").document(staff_id)
+            staff_doc = staff_ref.get()
+
+            if staff_doc.exists:
+                current_load = staff_doc.to_dict().get("current_load", 0)
+                new_load = max(0, current_load + increment)
+
+                staff_ref.update(
+                    {"current_load": new_load, "updated_at": firestore.SERVER_TIMESTAMP}
+                )
+                logger.info(
+                    f"Updated staff {staff_id} workload: {current_load} -> {new_load}"
+                )
+                return True
+            return False
+        except Exception as e:
+            logger.error(f"Error updating staff workload: {e}")
+            return False
+
+    # ==================== TASK OPERATIONS ====================
+
+    async def create_task(self, task_data: Dict) -> Optional[str]:
+        """Create a new task"""
+        try:
+            task_data["created_at"] = firestore.SERVER_TIMESTAMP
+            task_data["status"] = task_data.get("status", "pending")
+
+            doc_ref = self.db.collection("tasks").add(task_data)
+            task_id = doc_ref[1].id
+
+            # Increment staff workload if assigned
+            if "assigned_to" in task_data:
+                await self.update_staff_workload(task_data["assigned_to"], 1)
+
+            logger.info(f"Created task with ID: {task_id}")
+            return task_id
+        except Exception as e:
+            logger.error(f"Error creating task: {e}")
+            return None
+
+    async def get_task(self, task_id: str) -> Optional[Dict]:
+        """Get a specific task by ID"""
+        try:
+            doc = self.db.collection("tasks").document(task_id).get()
+            if doc.exists:
+                data = doc.to_dict()
+                data["id"] = doc.id
+                return data
+            return None
+        except Exception as e:
+            logger.error(f"Error getting task {task_id}: {e}")
+            return None
+
+    async def get_tasks(self, filters: Optional[Dict] = None) -> List[Dict]:
+        """Get tasks with optional filters"""
+        try:
+            query = self.db.collection("tasks")
+
+            if filters:
+                if "status" in filters:
+                    if isinstance(filters["status"], list):
+                        # Multiple statuses - filter in memory
+                        pass
+                    else:
+                        query = query.where(
+                            filter=FieldFilter("status", "==", filters["status"])
+                        )
+
+                if "assigned_to" in filters:
+                    query = query.where(
+                        filter=FieldFilter("assigned_to", "==", filters["assigned_to"])
+                    )
+
+                if "priority" in filters:
+                    query = query.where(
+                        filter=FieldFilter("priority", "==", filters["priority"])
+                    )
+
+            # Order by creation time
+            query = query.order_by("created_at", direction=firestore.Query.DESCENDING)
+
+            docs = query.stream()
+            tasks = []
+            for doc in docs:
+                data = doc.to_dict()
+                data["id"] = doc.id
+
+                # Apply multi-status filter if needed
+                if (
+                    filters
+                    and "status" in filters
+                    and isinstance(filters["status"], list)
+                ):
+                    if data.get("status") not in filters["status"]:
+                        continue
+
+                tasks.append(data)
+
+            logger.info(f"Retrieved {len(tasks)} tasks")
+            return tasks
+        except Exception as e:
+            logger.error(f"Error getting tasks: {e}")
+            return []
+
+    async def update_task_status(
+        self, task_id: str, status: str, notes: Optional[str] = None
+    ) -> bool:
+        """Update task status"""
+        try:
+            update_data = {"status": status, "updated_at": firestore.SERVER_TIMESTAMP}
+
+            if status == "in_progress" and notes is None:
+                update_data["started_at"] = firestore.SERVER_TIMESTAMP
+            elif status == "completed":
+                update_data["completed_at"] = firestore.SERVER_TIMESTAMP
+
+                # Decrease staff workload
+                task = await self.get_task(task_id)
+                if task and "assigned_to" in task:
+                    await self.update_staff_workload(task["assigned_to"], -1)
+
+            if notes:
+                update_data["notes"] = notes
+
+            self.db.collection("tasks").document(task_id).update(update_data)
+            logger.info(f"Updated task {task_id} status to {status}")
+            return True
+        except Exception as e:
+            logger.error(f"Error updating task status: {e}")
+            return False
+
+    # ==================== EVENT LOGGING ====================
+
+    async def log_event(self, event_data: Dict) -> Optional[str]:
+        """Log an event to the event_logs collection"""
+        try:
+            event_data["timestamp"] = firestore.SERVER_TIMESTAMP
+
+            doc_ref = self.db.collection("event_logs").add(event_data)
+            event_id = doc_ref[1].id
+
+            logger.debug(f"Logged event with ID: {event_id}")
+            return event_id
+        except Exception as e:
+            logger.error(f"Error logging event: {e}")
+            return None
+
+    # ==================== ANALYTICS ====================
+
+    async def get_metrics(self) -> Dict[str, Any]:
+        """Get current system metrics"""
+        try:
+            beds = await self.get_all_beds()
+            patients = await self.get_all_patients()
+            staff = await self.get_all_staff({"is_on_shift": True})
+            tasks = await self.get_tasks({"status": ["pending", "in_progress"]})
+
+            total_beds = len(beds)
+            available_beds = len([b for b in beds if b["status"] == "ready"])
+            occupied_beds = len([b for b in beds if b["status"] == "occupied"])
+            cleaning_beds = len([b for b in beds if b["status"] == "cleaning"])
+
+            metrics = {
+                "beds": {
+                    "total": total_beds,
+                    "available": available_beds,
+                    "occupied": occupied_beds,
+                    "cleaning": cleaning_beds,
+                    "utilization_rate": (occupied_beds / total_beds * 100)
+                    if total_beds > 0
+                    else 0,
+                },
+                "patients": {
+                    "total": len(patients),
+                    "waiting": len(
+                        [p for p in patients if p.get("status") == "waiting"]
+                    ),
+                    "admitted": len(
+                        [p for p in patients if p.get("status") == "admitted"]
+                    ),
+                },
+                "staff": {
+                    "on_shift": len(staff),
+                    "nurses": len([s for s in staff if s["role"] == "nurse"]),
+                    "cleaners": len([s for s in staff if s["role"] == "cleaner"]),
+                },
+                "tasks": {
+                    "active": len(tasks),
+                    "pending": len([t for t in tasks if t["status"] == "pending"]),
+                    "in_progress": len(
+                        [t for t in tasks if t["status"] == "in_progress"]
+                    ),
+                },
+                "timestamp": datetime.now().isoformat(),
+            }
+
+            return metrics
+        except Exception as e:
+            logger.error(f"Error getting metrics: {e}")
+            return {}

services/gemini_service.py

@@ -0,0 +1,401 @@
+"""
+Gemini Service for CareFlow Nexus
+Handles all Gemini AI API interactions using gemini-2.0-flash-exp model
+"""
+
+import json
+import logging
+import re
+from typing import Any, Dict, List, Optional
+
+import google.generativeai as genai
+from tenacity import retry, stop_after_attempt, wait_exponential
+
+logger = logging.getLogger(__name__)
+
+
+class GeminiService:
+    """Service class for Gemini AI API operations"""
+
+    def __init__(self, api_key: str, model_name: str = "gemini-2.0-flash-exp"):
+        """
+        Initialize Gemini service
+
+        Args:
+            api_key: Google API key for Gemini
+            model_name: Model name (default: gemini-2.0-flash-exp)
+        """
+        try:
+            genai.configure(api_key=api_key)
+            self.model_name = model_name
+            self.model = genai.GenerativeModel(model_name)
+            logger.info(f"Gemini service initialized with model: {model_name}")
+        except Exception as e:
+            logger.error(f"Failed to initialize Gemini service: {e}")
+            raise
+
+    @retry(
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=1, min=2, max=10),
+        reraise=True,
+    )
+    async def generate_response(
+        self, prompt: str, temperature: float = 0.7, max_tokens: int = 2048
+    ) -> str:
+        """
+        Generate a text response from Gemini
+
+        Args:
+            prompt: Input prompt
+            temperature: Sampling temperature (0.0-1.0)
+            max_tokens: Maximum tokens in response
+
+        Returns:
+            Generated text response
+        """
+        try:
+            generation_config = genai.GenerationConfig(
+                temperature=temperature,
+                max_output_tokens=max_tokens,
+                candidate_count=1,
+            )
+
+            response = self.model.generate_content(
+                prompt, generation_config=generation_config
+            )
+
+            if response.text:
+                logger.debug(f"Generated response: {response.text[:100]}...")
+                return response.text
+            else:
+                logger.warning("Empty response from Gemini")
+                return ""
+
+        except Exception as e:
+            logger.error(f"Error generating response: {e}")
+            raise
+
+    @retry(
+        stop=stop_after_attempt(3),
+        wait=wait_exponential(multiplier=1, min=2, max=10),
+        reraise=True,
+    )
+    async def generate_json_response(
+        self, prompt: str, temperature: float = 0.5, max_tokens: int = 2048
+    ) -> Dict[str, Any]:
+        """
+        Generate a JSON response from Gemini
+
+        Args:
+            prompt: Input prompt (should instruct to return JSON)
+            temperature: Sampling temperature
+            max_tokens: Maximum tokens in response
+
+        Returns:
+            Parsed JSON response as dictionary
+        """
+        try:
+            # Add JSON instruction to prompt if not present
+            if "json" not in prompt.lower():
+                prompt = f"{prompt}\n\nRespond ONLY with valid JSON."
+
+            text_response = await self.generate_response(
+                prompt, temperature=temperature, max_tokens=max_tokens
+            )
+
+            # Extract JSON from response
+            json_data = self._extract_json(text_response)
+
+            if json_data:
+                logger.debug("Successfully extracted JSON from response")
+                return json_data
+            else:
+                logger.warning("Failed to extract JSON, returning empty dict")
+                return {}
+
+        except Exception as e:
+            logger.error(f"Error generating JSON response: {e}")
+            return {}
+
+    def _extract_json(self, text: str) -> Optional[Dict[str, Any]]:
+        """
+        Extract JSON from text response
+
+        Args:
+            text: Text containing JSON
+
+        Returns:
+            Parsed JSON dictionary or None
+        """
+        try:
+            # Try direct JSON parse first
+            return json.loads(text)
+        except json.JSONDecodeError:
+            pass
+
+        # Try to find JSON in markdown code blocks
+        json_pattern = r"```(?:json)?\s*(\{.*?\})\s*```"
+        matches = re.findall(json_pattern, text, re.DOTALL)
+
+        if matches:
+            try:
+                return json.loads(matches[0])
+            except json.JSONDecodeError:
+                pass
+
+        # Try to find any JSON object in the text
+        json_pattern = r"\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}"
+        matches = re.findall(json_pattern, text, re.DOTALL)
+
+        for match in matches:
+            try:
+                return json.loads(match)
+            except json.JSONDecodeError:
+                continue
+
+        logger.warning("Could not extract valid JSON from response")
+        return None
+
+    async def generate_structured(
+        self,
+        system_instruction: str,
+        user_prompt: str,
+        temperature: float = 0.5,
+        max_tokens: int = 2048,
+    ) -> Dict[str, Any]:
+        """
+        Generate structured response with system instruction
+
+        Args:
+            system_instruction: System-level instruction/context
+            user_prompt: User query/request
+            temperature: Sampling temperature
+            max_tokens: Maximum tokens
+
+        Returns:
+            Parsed JSON response
+        """
+        try:
+            # Create model with system instruction
+            model_with_system = genai.GenerativeModel(
+                self.model_name, system_instruction=system_instruction
+            )
+
+            generation_config = genai.GenerationConfig(
+                temperature=temperature,
+                max_output_tokens=max_tokens,
+                candidate_count=1,
+            )
+
+            response = model_with_system.generate_content(
+                user_prompt, generation_config=generation_config
+            )
+
+            if response.text:
+                return self._extract_json(response.text) or {}
+            return {}
+
+        except Exception as e:
+            logger.error(f"Error generating structured response: {e}")
+            return {}
+
+    async def analyze_text(
+        self, text: str, analysis_type: str, context: Optional[Dict] = None
+    ) -> str:
+        """
+        Analyze text with specific analysis type
+
+        Args:
+            text: Text to analyze
+            analysis_type: Type of analysis (e.g., "diagnosis", "requirements", "sentiment")
+            context: Optional additional context
+
+        Returns:
+            Analysis result as text
+        """
+        context_str = ""
+        if context:
+            context_str = f"\n\nContext: {json.dumps(context, indent=2)}"
+
+        prompt = f"""
+Analyze the following text for {analysis_type}:
+
+{text}
+{context_str}
+
+Provide a clear, concise analysis.
+"""
+
+        return await self.generate_response(prompt, temperature=0.3)
+
+    async def score_and_rank(
+        self, items: List[Dict], criteria: str, context: Dict
+    ) -> Dict[str, Any]:
+        """
+        Score and rank items based on criteria
+
+        Args:
+            items: List of items to rank
+            criteria: Ranking criteria description
+            context: Context information
+
+        Returns:
+            Dictionary with ranked items and scores
+        """
+        prompt = f"""
+You are an expert ranking system.
+
+CRITERIA: {criteria}
+
+CONTEXT:
+{json.dumps(context, indent=2)}
+
+ITEMS TO RANK:
+{json.dumps(items, indent=2)}
+
+Score each item from 0-100 and rank them. Provide reasoning for each score.
+
+Respond with JSON in this format:
+{{
+  "rankings": [
+    {{
+      "item_id": "id",
+      "score": 0-100,
+      "reasoning": "detailed explanation",
+      "pros": ["advantage 1", "advantage 2"],
+      "cons": ["concern 1", "concern 2"]
+    }}
+  ],
+  "confidence": 0-100,
+  "overall_recommendation": "summary"
+}}
+"""
+
+        return await self.generate_json_response(prompt, temperature=0.5)
+
+    async def extract_requirements(
+        self, diagnosis: str, patient_info: Dict
+    ) -> Dict[str, Any]:
+        """
+        Extract medical requirements from diagnosis
+
+        Args:
+            diagnosis: Patient diagnosis text
+            patient_info: Patient information dictionary
+
+        Returns:
+            Dictionary of extracted requirements
+        """
+        prompt = f"""
+You are a medical requirements analyzer.
+
+PATIENT INFORMATION:
+- Age: {patient_info.get("age", "Unknown")}
+- Gender: {patient_info.get("gender", "Unknown")}
+- Diagnosis: {diagnosis}
+- Severity: {patient_info.get("severity", "moderate")}
+- Mobility: {patient_info.get("mobility_status", "ambulatory")}
+
+Extract the medical care requirements and respond with JSON:
+{{
+  "needs_oxygen": true/false,
+  "needs_ventilator": true/false,
+  "needs_cardiac_monitor": true/false,
+  "needs_isolation": true/false,
+  "preferred_ward": "ward name or null",
+  "proximity_preference": 1-10,
+  "special_considerations": ["list of special needs"],
+  "reasoning": "brief explanation"
+}}
+"""
+
+        return await self.generate_json_response(prompt, temperature=0.3)
+
+    async def generate_task_assignment_reasoning(
+        self, task: Dict, staff: Dict, context: Dict
+    ) -> str:
+        """
+        Generate reasoning for task assignment
+
+        Args:
+            task: Task details
+            staff: Staff member details
+            context: Additional context
+
+        Returns:
+            Reasoning text
+        """
+        prompt = f"""
+Explain why {staff.get("name")} is the best choice for this task:
+
+TASK:
+{json.dumps(task, indent=2)}
+
+STAFF MEMBER:
+{json.dumps(staff, indent=2)}
+
+CONTEXT:
+{json.dumps(context, indent=2)}
+
+Provide a clear, concise explanation in 2-3 sentences.
+"""
+
+        return await self.generate_response(prompt, temperature=0.5)
+
+    async def detect_bottlenecks(self, system_state: Dict) -> Dict[str, Any]:
+        """
+        Analyze system state for bottlenecks
+
+        Args:
+            system_state: Current system state metrics
+
+        Returns:
+            Bottleneck analysis
+        """
+        prompt = f"""
+You are a hospital operations analyst.
+
+CURRENT SYSTEM STATE:
+{json.dumps(system_state, indent=2)}
+
+Analyze for bottlenecks, inefficiencies, and potential issues.
+
+Respond with JSON:
+{{
+  "bottlenecks": [
+    {{
+      "type": "bottleneck type",
+      "severity": "low/medium/high/critical",
+      "description": "what's wrong",
+      "impact": "how it affects operations",
+      "recommendation": "suggested action"
+    }}
+  ],
+  "alerts": ["urgent issues requiring immediate attention"],
+  "recommendations": ["proactive suggestions"],
+  "capacity_forecast": "prediction for next 4-6 hours"
+}}
+"""
+
+        return await self.generate_json_response(prompt, temperature=0.3)
+
+    def validate_json_schema(self, data: Dict, required_keys: List[str]) -> bool:
+        """
+        Validate JSON data has required keys
+
+        Args:
+            data: JSON data to validate
+            required_keys: List of required keys
+
+        Returns:
+            True if valid, False otherwise
+        """
+        return all(key in data for key in required_keys)
+
+    async def get_model_info(self) -> Dict[str, Any]:
+        """Get information about the current model"""
+        return {
+            "model_name": self.model_name,
+            "provider": "Google Gemini",
+            "version": "2.0-flash-exp",
+        }

utils/__init__.py

@@ -0,0 +1,8 @@
+"""
+Utils module for CareFlow Nexus AI Agents
+Provides utility functions for response parsing and validation
+"""
+
+from .response_parser import ResponseParser
+
+__all__ = ["ResponseParser"]

utils/response_parser.py

@@ -0,0 +1,375 @@
+"""
+Response Parser Utility for CareFlow Nexus
+Handles parsing and validation of Gemini AI responses
+"""
+
+import json
+import logging
+import re
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class ResponseParser:
+    """Utility class for parsing and validating AI responses"""
+
+    @staticmethod
+    def extract_json(text: str) -> Optional[Dict[str, Any]]:
+        """
+        Extract JSON from text response (handles various formats)
+
+        Args:
+            text: Text containing JSON
+
+        Returns:
+            Parsed JSON dictionary or None
+        """
+        if not text:
+            return None
+
+        # Try direct JSON parse first
+        try:
+            return json.loads(text.strip())
+        except json.JSONDecodeError:
+            pass
+
+        # Try to find JSON in markdown code blocks
+        patterns = [
+            r"```json\s*(\{.*?\})\s*```",  # ```json {...} ```
+            r"```\s*(\{.*?\})\s*```",  # ``` {...} ```
+            r"```json\s*(\[.*?\])\s*```",  # ```json [...] ```
+            r"```\s*(\[.*?\])\s*```",  # ``` [...] ```
+        ]
+
+        for pattern in patterns:
+            matches = re.findall(pattern, text, re.DOTALL)
+            if matches:
+                try:
+                    return json.loads(matches[0])
+                except json.JSONDecodeError:
+                    continue
+
+        # Try to find any JSON object or array in the text
+        json_object_pattern = r"\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}"
+        json_array_pattern = r"\[[^\[\]]*(?:\[[^\[\]]*\][^\[\]]*)*\]"
+
+        for pattern in [json_object_pattern, json_array_pattern]:
+            matches = re.findall(pattern, text, re.DOTALL)
+            for match in matches:
+                try:
+                    parsed = json.loads(match)
+                    # Verify it's a meaningful JSON (not just empty)
+                    if parsed:
+                        return parsed
+                except json.JSONDecodeError:
+                    continue
+
+        logger.warning("Could not extract valid JSON from response")
+        return None
+
+    @staticmethod
+    def validate_required_fields(
+        data: Dict[str, Any], required_fields: List[str]
+    ) -> tuple[bool, List[str]]:
+        """
+        Validate that dictionary contains required fields
+
+        Args:
+            data: Dictionary to validate
+            required_fields: List of required field names
+
+        Returns:
+            Tuple of (is_valid, missing_fields)
+        """
+        if not isinstance(data, dict):
+            return False, required_fields
+
+        missing = [field for field in required_fields if field not in data]
+        return len(missing) == 0, missing
+
+    @staticmethod
+    def sanitize_response(data: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Clean and normalize response data
+
+        Args:
+            data: Raw response data
+
+        Returns:
+            Sanitized dictionary
+        """
+        if not isinstance(data, dict):
+            return {}
+
+        sanitized = {}
+        for key, value in data.items():
+            # Clean key (remove special chars, lowercase)
+            clean_key = key.strip().lower().replace(" ", "_")
+
+            # Clean value based on type
+            if isinstance(value, str):
+                sanitized[clean_key] = value.strip()
+            elif isinstance(value, dict):
+                sanitized[clean_key] = ResponseParser.sanitize_response(value)
+            elif isinstance(value, list):
+                sanitized[clean_key] = [
+                    ResponseParser.sanitize_response(item)
+                    if isinstance(item, dict)
+                    else item
+                    for item in value
+                ]
+            else:
+                sanitized[clean_key] = value
+
+        return sanitized
+
+    @staticmethod
+    def validate_score(score: Any, min_val: int = 0, max_val: int = 100) -> int:
+        """
+        Validate and normalize score to range
+
+        Args:
+            score: Score value (any type)
+            min_val: Minimum valid score
+            max_val: Maximum valid score
+
+        Returns:
+            Validated score within range
+        """
+        try:
+            score_int = int(float(score))
+            return max(min_val, min(max_val, score_int))
+        except (ValueError, TypeError):
+            logger.warning(f"Invalid score value: {score}, returning 0")
+            return 0
+
+    @staticmethod
+    def parse_bed_allocation_response(response: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Parse and validate bed allocation response
+
+        Args:
+            response: Raw response from AI
+
+        Returns:
+            Validated and structured response
+        """
+        try:
+            recommendations = response.get("recommendations", [])
+            if not isinstance(recommendations, list):
+                recommendations = []
+
+            parsed_recs = []
+            for rec in recommendations[:3]:  # Top 3 only
+                if not isinstance(rec, dict):
+                    continue
+
+                parsed_rec = {
+                    "bed_id": rec.get("bed_id", ""),
+                    "bed_number": rec.get("bed_number", ""),
+                    "ward": rec.get("ward", ""),
+                    "score": ResponseParser.validate_score(rec.get("score", 0)),
+                    "reasoning": rec.get("reasoning", "No reasoning provided"),
+                    "pros": rec.get("pros", [])
+                    if isinstance(rec.get("pros"), list)
+                    else [],
+                    "cons": rec.get("cons", [])
+                    if isinstance(rec.get("cons"), list)
+                    else [],
+                }
+
+                parsed_recs.append(parsed_rec)
+
+            return {
+                "recommendations": parsed_recs,
+                "overall_confidence": ResponseParser.validate_score(
+                    response.get("overall_confidence", 50)
+                ),
+                "considerations": response.get("considerations", ""),
+            }
+        except Exception as e:
+            logger.error(f"Error parsing bed allocation response: {e}")
+            return {
+                "recommendations": [],
+                "overall_confidence": 0,
+                "considerations": "",
+            }
+
+    @staticmethod
+    def parse_requirement_extraction_response(
+        response: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        """
+        Parse and validate requirement extraction response
+
+        Args:
+            response: Raw response from AI
+
+        Returns:
+            Validated requirements dictionary
+        """
+        try:
+            return {
+                "needs_oxygen": bool(response.get("needs_oxygen", False)),
+                "needs_ventilator": bool(response.get("needs_ventilator", False)),
+                "needs_cardiac_monitor": bool(
+                    response.get("needs_cardiac_monitor", False)
+                ),
+                "needs_isolation": bool(response.get("needs_isolation", False)),
+                "preferred_ward": response.get("preferred_ward"),
+                "proximity_preference": ResponseParser.validate_score(
+                    response.get("proximity_preference", 5), 1, 10
+                ),
+                "special_considerations": response.get("special_considerations", [])
+                if isinstance(response.get("special_considerations"), list)
+                else [],
+                "confidence": ResponseParser.validate_score(
+                    response.get("confidence", 50)
+                ),
+                "reasoning": response.get("reasoning", ""),
+            }
+        except Exception as e:
+            logger.error(f"Error parsing requirement extraction response: {e}")
+            return {
+                "needs_oxygen": False,
+                "needs_ventilator": False,
+                "needs_cardiac_monitor": False,
+                "needs_isolation": False,
+                "preferred_ward": None,
+                "proximity_preference": 5,
+                "special_considerations": [],
+                "confidence": 0,
+                "reasoning": "Error parsing response",
+            }
+
+    @staticmethod
+    def parse_staff_assignment_response(response: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Parse and validate staff assignment response
+
+        Args:
+            response: Raw response from AI
+
+        Returns:
+            Validated assignment dictionary
+        """
+        try:
+            alternatives = response.get("alternatives", [])
+            if not isinstance(alternatives, list):
+                alternatives = []
+
+            return {
+                "recommended_staff_id": response.get("recommended_staff_id", ""),
+                "staff_name": response.get("staff_name", ""),
+                "reasoning": response.get("reasoning", "No reasoning provided"),
+                "workload_impact": response.get("workload_impact", ""),
+                "concerns": response.get("concerns", [])
+                if isinstance(response.get("concerns"), list)
+                else [],
+                "alternatives": alternatives[:2],  # Top 2 alternatives
+                "confidence": ResponseParser.validate_score(
+                    response.get("confidence", 50)
+                ),
+            }
+        except Exception as e:
+            logger.error(f"Error parsing staff assignment response: {e}")
+            return {
+                "recommended_staff_id": "",
+                "staff_name": "",
+                "reasoning": "Error parsing response",
+                "workload_impact": "",
+                "concerns": [],
+                "alternatives": [],
+                "confidence": 0,
+            }
+
+    @staticmethod
+    def parse_state_analysis_response(response: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Parse and validate state analysis response
+
+        Args:
+            response: Raw response from AI
+
+        Returns:
+            Validated analysis dictionary
+        """
+        try:
+            return {
+                "critical_alerts": response.get("critical_alerts", [])
+                if isinstance(response.get("critical_alerts"), list)
+                else [],
+                "bottlenecks": response.get("bottlenecks", [])
+                if isinstance(response.get("bottlenecks"), list)
+                else [],
+                "capacity_forecast": response.get("capacity_forecast", {})
+                if isinstance(response.get("capacity_forecast"), dict)
+                else {},
+                "recommendations": response.get("recommendations", [])
+                if isinstance(response.get("recommendations"), list)
+                else [],
+            }
+        except Exception as e:
+            logger.error(f"Error parsing state analysis response: {e}")
+            return {
+                "critical_alerts": [],
+                "bottlenecks": [],
+                "capacity_forecast": {},
+                "recommendations": [],
+            }
+
+    @staticmethod
+    def combine_scores(
+        rule_score: float, ai_score: float, rule_weight: float = 0.5
+    ) -> float:
+        """
+        Combine rule-based and AI scores with weights
+
+        Args:
+            rule_score: Rule-based score (0-100)
+            ai_score: AI-generated score (0-100)
+            rule_weight: Weight for rule score (0-1), AI gets (1-rule_weight)
+
+        Returns:
+            Combined score
+        """
+        ai_weight = 1.0 - rule_weight
+        combined = (rule_score * rule_weight) + (ai_score * ai_weight)
+        return round(combined, 2)
+
+    @staticmethod
+    def format_error_response(
+        error_message: str, error_type: str = "general"
+    ) -> Dict[str, Any]:
+        """
+        Format error into standard response structure
+
+        Args:
+            error_message: Error message
+            error_type: Type of error
+
+        Returns:
+            Error response dictionary
+        """
+        return {
+            "success": False,
+            "error": True,
+            "error_type": error_type,
+            "message": error_message,
+            "data": None,
+        }
+
+    @staticmethod
+    def format_success_response(data: Any, message: str = "Success") -> Dict[str, Any]:
+        """
+        Format success response
+
+        Args:
+            data: Response data
+            message: Success message
+
+        Returns:
+            Success response dictionary
+        """
+        return {"success": True, "error": False, "message": message, "data": data}