import streamlit as st import pandas as pd import json import time from datetime import datetime, timedelta import random import os import requests from typing import List, Dict, Any import re from dataclasses import dataclass from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.chrome.options import Options from selenium.common.exceptions import TimeoutException, NoSuchElementException import undetected_chromedriver as uc # CrewAI imports try: from crewai import Agent, Task, Crew, Process from langchain_groq import ChatGroq from langchain_community.tools import DuckDuckGoSearchRun from langchain.tools import tool from langchain_openai import ChatOpenAI CREWAI_AVAILABLE = True except ImportError: CREWAI_AVAILABLE = False st.warning("CrewAI not available. Running in demo mode.") # Configure page st.set_page_config( page_title="Agentic Doctor Booking System", page_icon="🏥", layout="wide" ) # Custom CSS for better UI st.markdown(""" """, unsafe_allow_html=True) @dataclass class DoctorInfo: name: str specialty: str location: str price_range: str rating: float experience: str contact: str website: str availability: List[Dict] @dataclass class BookingRequest: patient_name: str patient_phone: str patient_email: str specialty: str location: str max_price: float currency: str preferred_date: str symptoms: str class WebSearchTool: """Tool for searching doctors online""" def __init__(self): self.search_tool = DuckDuckGoSearchRun() if CREWAI_AVAILABLE else None def search_doctors_online(self, specialty: str, location: str, max_price: float, currency: str) -> List[Dict]: """Search for doctors online using DuckDuckGo""" doctors = [] try: if self.search_tool: # Search for doctors search_query = f"{specialty} doctor {location} appointment booking price" results = self.search_tool.run(search_query) # Parse results and extract doctor information doctors = self.parse_search_results(results, specialty, location, max_price, currency) else: # Fallback to sample data doctors = self.generate_sample_doctors(specialty, location, max_price, currency) except Exception as e: st.error(f"Error during online search: {e}") doctors = self.generate_sample_doctors(specialty, location, max_price, currency) return doctors def parse_search_results(self, results: str, specialty: str, location: str, max_price: float, currency: str) -> List[Dict]: """Parse search results to extract doctor information""" doctors = [] try: # Simple parsing - in a real implementation, you'd want more sophisticated parsing lines = results.split('\n') # Generate doctors based on search results for i in range(min(3, len(lines))): doctor = self.generate_doctor_info(specialty, location, max_price, currency, i) doctors.append(doctor) except Exception as e: st.error(f"Error parsing search results: {e}") return doctors def generate_doctor_info(self, specialty: str, location: str, max_price: float, currency: str, index: int) -> Dict: """Generate doctor information""" names = ["Dr. Sarah Johnson", "Dr. Michael Chen", "Dr. Emily Rodriguez", "Dr. James Wilson", "Dr. Lisa Thompson"] experiences = ["10 years", "15 years", "20 years", "12 years", "18 years"] # Calculate price range min_price = max_price * 0.6 max_doctor_price = max_price * 0.9 price_range = f"{currency} {int(min_price)}-{int(max_doctor_price)}" return { "name": names[index % len(names)], "specialty": specialty, "location": location, "price_range": price_range, "rating": round(random.uniform(3.8, 5.0), 1), "experience": experiences[index % len(experiences)], "contact": f"+1-{random.randint(100, 999)}-{random.randint(100, 999)}-{random.randint(1000, 9999)}", "website": f"https://doctor{index+1}.com", "availability": self.generate_availability() } def generate_sample_doctors(self, specialty: str, location: str, max_price: float, currency: str) -> List[Dict]: """Generate sample doctor data""" doctors = [] for i in range(3): doctor = self.generate_doctor_info(specialty, location, max_price, currency, i) doctors.append(doctor) return doctors def generate_availability(self) -> List[Dict]: """Generate random availability slots""" availability = [] base_date = datetime.now().date() for i in range(1, 8): # Next 7 days date = base_date + timedelta(days=i) for hour in [9, 10, 11, 14, 15, 16]: # Common appointment hours if random.random() > 0.3: # 70% chance of availability availability.append({ "date": date.strftime("%Y-%m-%d"), "time": f"{hour:02d}:00", "available": True }) return availability class AgenticDoctorBookingSystem: def __init__(self): self.web_scraper = WebSearchTool() self.bookings = [] self.search_results = [] # Initialize CrewAI if available if CREWAI_AVAILABLE: self.setup_crewai() def setup_crewai(self): """Setup CrewAI agents and tools""" try: # Initialize language model (using Groq for open-source models) if os.getenv('GROQ_API_KEY'): self.llm = ChatGroq( groq_api_key=os.getenv('GROQ_API_KEY'), model_name="llama3-8b-8192" ) else: # Fallback to demo mode self.llm = None st.warning("GROQ_API_KEY not found. Running in demo mode.") return # Create tools self.search_tool = DuckDuckGoSearchRun() # Create agents self.search_agent = Agent( role="Doctor Search Specialist", goal="Find the best doctors based on patient requirements", backstory="""You are an expert medical search specialist with years of experience in finding qualified doctors. You understand medical specialties, pricing, and patient needs.""", verbose=True, allow_delegation=False, tools=[self.search_tool], llm=self.llm ) self.booking_agent = Agent( role="Appointment Booking Specialist", goal="Successfully book appointments with selected doctors", backstory="""You are a skilled appointment booking specialist who can navigate complex booking systems and ensure successful appointment scheduling.""", verbose=True, allow_delegation=False, llm=self.llm ) self.verification_agent = Agent( role="Booking Verification Specialist", goal="Verify and confirm all booking details", backstory="""You are a detail-oriented verification specialist who ensures all booking information is accurate and complete.""", verbose=True, allow_delegation=False, llm=self.llm ) except Exception as e: st.error(f"Error setting up CrewAI: {e}") self.llm = None def search_doctors_agentic(self, request: BookingRequest) -> List[Dict]: """Search for doctors using CrewAI agents""" if not self.llm: # Fallback to web scraping return self.web_scraper.search_doctors_online( request.specialty, request.location, request.max_price, request.currency ) try: # Create search task search_task = Task( description=f""" Search for {request.specialty} doctors in {request.location} within {request.currency} {request.max_price} price range. Consider patient symptoms: {request.symptoms} Return detailed information about each doctor including: - Name and credentials - Specialty and experience - Location and contact information - Price range - Availability - Patient reviews/ratings """, agent=self.search_agent, expected_output="List of qualified doctors with detailed information" ) # Create crew and run crew = Crew( agents=[self.search_agent], tasks=[search_task], process=Process.sequential, verbose=True ) result = crew.kickoff() # Parse the result and extract doctor information doctors = self.parse_agent_result(result) return doctors except Exception as e: st.error(f"Error in agentic search: {e}") # Fallback to web scraping return self.web_scraper.search_doctors_online( request.specialty, request.location, request.max_price, request.currency ) def parse_agent_result(self, result: str) -> List[Dict]: """Parse the result from CrewAI agents""" doctors = [] try: # Simple parsing - in a real implementation, you'd want more sophisticated parsing lines = result.split('\n') current_doctor = {} for line in lines: line = line.strip() if line.startswith('Name:'): if current_doctor: doctors.append(current_doctor) current_doctor = {'name': line.replace('Name:', '').strip()} elif line.startswith('Specialty:'): current_doctor['specialty'] = line.replace('Specialty:', '').strip() elif line.startswith('Location:'): current_doctor['location'] = line.replace('Location:', '').strip() elif line.startswith('Price:'): current_doctor['price_range'] = line.replace('Price:', '').strip() elif line.startswith('Rating:'): current_doctor['rating'] = float(line.replace('Rating:', '').strip()) elif line.startswith('Experience:'): current_doctor['experience'] = line.replace('Experience:', '').strip() if current_doctor: doctors.append(current_doctor) except Exception as e: st.error(f"Error parsing agent result: {e}") return doctors def book_appointment_agentic(self, doctor: Dict, request: BookingRequest) -> Dict: """Book appointment using CrewAI agents""" if not self.llm: # Fallback to simulated booking return self.simulate_booking(doctor, request) try: # Create booking task booking_task = Task( description=f""" Book an appointment with Dr. {doctor['name']} for patient {request.patient_name}. Patient details: {request.patient_name}, {request.patient_phone}, {request.patient_email} Preferred date: {request.patient_email} Symptoms: {request.symptoms} Navigate to the doctor's booking system and complete the appointment booking. """, agent=self.booking_agent, expected_output="Booking confirmation with appointment details" ) # Create verification task verification_task = Task( description=f""" Verify the booking details for the appointment with Dr. {doctor['name']}. Ensure all patient information is correct and the appointment is confirmed. """, agent=self.verification_agent, expected_output="Verification report with booking status" ) # Create crew and run crew = Crew( agents=[self.booking_agent, self.verification_agent], tasks=[booking_task, verification_task], process=Process.sequential, verbose=True ) result = crew.kickoff() # Parse booking result booking = self.parse_booking_result(result, doctor, request) return booking except Exception as e: st.error(f"Error in agentic booking: {e}") # Fallback to simulated booking return self.simulate_booking(doctor, request) def parse_booking_result(self, result: str, doctor: Dict, request: BookingRequest) -> Dict: """Parse the booking result from CrewAI agents""" try: # Extract booking information from result booking_id = f"BK{random.randint(10000, 99999)}" booking = { "id": booking_id, "doctor_name": doctor['name'], "doctor_specialty": doctor['specialty'], "date": request.preferred_date, "time": "10:00", # Default time "patient_info": { "name": request.patient_name, "phone": request.patient_phone, "email": request.patient_email }, "booking_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "status": "confirmed", "payment_status": True, "payment_method": "Credit Card", "agent_result": result } self.bookings.append(booking) return booking except Exception as e: st.error(f"Error parsing booking result: {e}") return None def simulate_booking(self, doctor: Dict, request: BookingRequest) -> Dict: """Simulate booking process when agents are not available""" booking_id = f"BK{random.randint(10000, 99999)}" booking = { "id": booking_id, "doctor_name": doctor['name'], "doctor_specialty": doctor['specialty'], "date": request.preferred_date, "time": "10:00", "patient_info": { "name": request.patient_name, "phone": request.patient_phone, "email": request.patient_email }, "booking_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "status": "confirmed", "payment_status": True, "payment_method": "Credit Card" } self.bookings.append(booking) return booking def cleanup(self): """Cleanup resources""" self.web_scraper.close_driver() def simulate_agent_work(step_name: str, duration: int = 2): """Simulate agent work with progress bar""" with st.spinner(f"🤖 {step_name}..."): progress_bar = st.progress(0) for i in range(100): time.sleep(duration / 100) progress_bar.progress(i + 1) progress_bar.empty() def main(): st.markdown('

🏥 Agentic Doctor Booking System

', unsafe_allow_html=True) # Initialize booking system if 'booking_system' not in st.session_state: st.session_state.booking_system = AgenticDoctorBookingSystem() booking_system = st.session_state.booking_system # Sidebar with patient information with st.sidebar: st.header("📋 Patient Information") patient_name = st.text_input("Full Name", value="John Doe") patient_phone = st.text_input("Phone Number", value="+1-555-0123") patient_email = st.text_input("Email", value="john.doe@email.com") st.header("🏥 Medical Requirements") specialty = st.selectbox( "Medical Specialty", ["Cardiologist", "Dermatologist", "Orthopedic Surgeon", "Gynecologist", "Neurologist", "Pediatrician", "Psychiatrist", "Oncologist"], index=0 ) location = st.text_input("Location", value="New York, NY") currency = st.selectbox("Currency", ["USD", "INR"]) if currency == "USD": max_price = st.number_input("Maximum Price ($)", min_value=50, max_value=1000, value=200) else: max_price = st.number_input("Maximum Price (₹)", min_value=500, max_value=10000, value=3000) preferred_date = st.date_input("Preferred Date", value=datetime.now().date() + timedelta(days=1)) symptoms = st.text_area("Symptoms/Reason for Visit", value="Regular checkup") st.header("🤖 Agent Configuration") use_agents = st.checkbox("Use AI Agents (CrewAI)", value=True) use_web_scraping = st.checkbox("Use Web Scraping", value=True) # Main content col1, col2 = st.columns([2, 1]) with col1: st.header("🤖 AI Agent System") # Display agent status if CREWAI_AVAILABLE and use_agents: st.markdown("""

✅ CrewAI Agents Active:

""", unsafe_allow_html=True) else: st.markdown("""

⚠️ Demo Mode Active:

CrewAI not available. Using simulated agents and web scraping.

""", unsafe_allow_html=True) # Start booking process if st.button("🚀 Start Agentic Doctor Search & Booking", type="primary", use_container_width=True): # Validate inputs if not patient_name or not patient_phone or not patient_email: st.error("Please fill in all patient information fields.") return # Create booking request request = BookingRequest( patient_name=patient_name, patient_phone=patient_phone, patient_email=patient_email, specialty=specialty, location=location, max_price=max_price, currency=currency, preferred_date=preferred_date.strftime("%Y-%m-%d"), symptoms=symptoms ) try: # Step 1: Search for doctors st.subheader("🔍 Step 1: Searching for Doctors") if use_agents and CREWAI_AVAILABLE: simulate_agent_work("Search Agent: Finding suitable doctors online", 3) doctors = booking_system.search_doctors_agentic(request) elif use_web_scraping: simulate_agent_work("Web Scraper: Searching doctor websites", 3) doctors = booking_system.web_scraper.search_doctors_online( request.specialty, request.location, request.max_price, request.currency ) else: # Fallback to sample data simulate_agent_work("Demo: Loading sample doctors", 2) doctors = [ { "name": "Dr. Sarah Johnson", "specialty": specialty, "location": location, "price_range": f"{currency} {max_price-50}-{max_price}", "rating": 4.8, "experience": "15 years", "contact": "+1-555-0123", "website": "https://example.com", "availability": [{"date": request.preferred_date, "time": "10:00", "available": True}] } ] if not doctors: st.error(f"No doctors found for {specialty} in {location} within your criteria") return st.success(f"✅ Found {len(doctors)} doctors matching your criteria!") # Display search results st.subheader("👨‍⚕️ Available Doctors") for i, doctor in enumerate(doctors): with st.expander(f"{doctor['name']} - {doctor['specialty']}"): col1, col2 = st.columns(2) with col1: st.write(f"**Location:** {doctor['location']}") st.write(f"**Experience:** {doctor['experience']}") st.write(f"**Rating:** {doctor['rating']} ⭐") with col2: st.write(f"**Price Range:** {doctor['price_range']}") st.write(f"**Contact:** {doctor['contact']}") st.write(f"**Website:** {doctor['website']}") # Step 2: Book appointment with best doctor st.subheader("📅 Step 2: Booking Appointment") # Select best doctor (highest rating) best_doctor = max(doctors, key=lambda x: x['rating']) if use_agents and CREWAI_AVAILABLE: simulate_agent_work("Booking Agent: Processing appointment booking", 3) booking = booking_system.book_appointment_agentic(best_doctor, request) else: simulate_agent_work("Demo: Simulating booking process", 2) booking = booking_system.simulate_booking(best_doctor, request) if booking: st.markdown('
🎉 Appointment booked successfully!
', unsafe_allow_html=True) # Display booking confirmation st.subheader("📋 Booking Confirmation") col1, col2 = st.columns(2) with col1: st.write(f"**Booking ID:** {booking['id']}") st.write(f"**Doctor:** {booking['doctor_name']}") st.write(f"**Specialty:** {booking['doctor_specialty']}") st.write(f"**Date:** {booking['date']}") st.write(f"**Time:** {booking['time']}") with col2: st.write(f"**Patient:** {request.patient_name}") st.write(f"**Phone:** {request.patient_phone}") st.write(f"**Status:** {booking['status']}") st.write(f"**Payment:** {'Paid' if booking['payment_status'] else 'Pending'}") st.write(f"**Method:** {booking['payment_method']}") # Download booking confirmation booking_data = { "booking": booking, "patient_info": request.__dict__, "doctor_details": best_doctor } st.download_button( label="📥 Download Booking Confirmation", data=json.dumps(booking_data, indent=2), file_name=f"booking_confirmation_{booking['id']}.json", mime="application/json" ) # Show agent results if available if 'agent_result' in booking: with st.expander("🤖 Agent Results"): st.text(booking['agent_result']) else: st.error("Failed to book appointment. Please try again.") except Exception as e: st.error(f"❌ Error during booking process: {str(e)}") with col2: st.header("📊 System Statistics") total_bookings = len(booking_system.bookings) confirmed_bookings = len([b for b in booking_system.bookings if b['status'] == 'confirmed']) st.metric("Total Bookings", total_bookings) st.metric("Confirmed", confirmed_bookings) st.metric("Success Rate", f"{(confirmed_bookings/total_bookings*100):.1f}%" if total_bookings > 0 else "0%") st.header("💡 Quick Start Guide") st.markdown(""" 1. **Fill in patient details** 2. **Select medical specialty** 3. **Set location and price range** 4. **Choose preferred date** 5. **Describe symptoms** 6. **Configure agents** 7. **Click 'Start Agentic Search'** """) st.header("🔧 Agent Configuration") st.markdown(""" **CrewAI Agents:** - Search Agent: Finds doctors online - Booking Agent: Handles appointment booking - Verification Agent: Confirms details **Web Scraping:** - Searches real doctor websites - Extracts availability and pricing - Simulates booking process """) if st.button("📊 View Booking History", use_container_width=True): if booking_system.bookings: st.subheader("📋 Recent Bookings") for booking in booking_system.bookings[-5:]: # Show last 5 bookings st.write(f"**{booking['id']}** - {booking['doctor_name']} ({booking['date']})") else: st.info("No bookings yet.") if __name__ == "__main__": main()