import re import json import requests import traceback import time import os from typing import Dict, Any, List, Optional, Tuple from datetime import datetime, timedelta # Updated imports for pydantic from pydantic import BaseModel, Field # Updated imports for LangChain from langchain_core.prompts import PromptTemplate, ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from langchain_ollama import OllamaLLM from langchain.chains import LLMChain from langchain.callbacks.manager import CallbackManager from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler from langchain_huggingface.embeddings import HuggingFaceEmbeddings # Enhanced HuggingFace imports for improved functionality from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification import numpy as np import aiohttp import asyncio # Import endpoints documentation from endpoints_documentation import endpoints_documentation # Set environment variables for HuggingFace os.environ["HF_HOME"] = "/tmp/huggingface" os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"] = "1" class ChatMessage(BaseModel): """Data model for chat messages""" message_id: str = Field(..., description="Unique identifier for the message") user_id: str = Field(..., description="User identifier") message: str = Field(..., description="The user's message") timestamp: datetime = Field(default_factory=datetime.now, description="When the message was sent") language: str = Field(default="english", description="Detected language of the message") class ChatResponse(BaseModel): """Data model for chatbot responses""" response_id: str = Field(..., description="Unique identifier for the response") response_type: str = Field(..., description="Type of response: 'conversation' or 'api_action'") message: str = Field(..., description="The chatbot's response message") api_call_made: bool = Field(default=False, description="Whether an API call was made") api_data: Optional[Dict[str, Any]] = Field(default=None, description="API response data if applicable") language: str = Field(default="english", description="Language of the response") timestamp: datetime = Field(default_factory=datetime.now, description="When the response was generated") class RouterResponse(BaseModel): """Data model for router chain response""" intent: str = Field(..., description="Either 'API_ACTION' or 'CONVERSATION'") confidence: float = Field(..., description="Confidence score between 0.0 and 1.0") reasoning: str = Field(..., description="Explanation of the decision") endpoint: Optional[str] = Field(default=None, description="API endpoint if intent is API_ACTION") method: Optional[str] = Field(default=None, description="HTTP method if intent is API_ACTION") params: Dict[str, Any] = Field(default_factory=dict, description="Parameters for API call") missing_required: List[str] = Field(default_factory=list, description="Missing required parameters") class HealthcareChatbot: def __init__(self): self.endpoints_documentation = endpoints_documentation self.cached_endpoints_documentation = json.dumps(self.endpoints_documentation, indent=2) self.ollama_base_url = "http://localhost:11434" self.model_name = "gemma3" self.BASE_URL = 'https://90cb-197-54-60-164.ngrok-free.app' self.headers = {'Content-type': 'application/json'} self.user_id = '8e5720d5-7243-42bd-97aa-10217309be82' self.max_retries = 3 self.retry_delay = 2 # Store conversation history self.conversation_history = [] self.max_history_length = 10 # Keep last 10 exchanges # Initialize components self._initialize_language_tools() self._initialize_llm() self._initialize_parsers_and_chains() self._initialize_date_parser() print("Healthcare Chatbot initialized successfully!") self._print_welcome_message() def _print_welcome_message(self): """Print welcome message in both languages""" print("\n" + "="*60) print("🏥 HEALTHCARE CHATBOT READY") print("="*60) print("English: Hello! I'm your healthcare assistant. I can help you with:") print("• Booking and managing appointments") print("• Finding hospital information") print("• Viewing your medical records") print("• General healthcare questions") print() print("Arabic: مرحباً! أنا مساعدك الطبي. يمكنني مساعدتك في:") print("• حجز وإدارة المواعيد") print("• العثور على معلومات المستشفى") print("• عرض سجلاتك الطبية") print("• الأسئلة الطبية العامة") print("="*60) print("Type 'quit' or 'خروج' to exit\n") def _initialize_language_tools(self): """Initialize language processing tools""" try: self.embeddings = HuggingFaceEmbeddings(model_name="intfloat/multilingual-e5-large") self.language_classifier = pipeline( "text-classification", model="papluca/xlm-roberta-base-language-detection", top_k=1 ) self.sentiment_analyzer = pipeline( "sentiment-analysis", model="cardiffnlp/twitter-xlm-roberta-base-sentiment" ) print("✓ Language processing models loaded successfully") except Exception as e: print(f"⚠ Warning: Some language models failed to load: {e}") self.language_classifier = None self.sentiment_analyzer = None def _initialize_date_parser(self): """Initialize date parsing model""" try: self.date_parser = pipeline( "token-classification", model="Jean-Baptiste/roberta-large-ner-english", aggregation_strategy="simple" ) except Exception as e: print(f"⚠ Warning: Date parsing model failed to load: {e}") self.date_parser = None def _initialize_llm(self): """Initialize the LLM""" callbacks = [StreamingStdOutCallbackHandler()] self.llm = OllamaLLM( model=self.model_name, base_url=self.ollama_base_url, callbacks=callbacks, temperature=0.7, num_ctx=8192, top_p=0.9, request_timeout=60, ) def _initialize_parsers_and_chains(self): """Initialize all prompt templates and chains - REVAMPED to 3 chains only""" self.json_parser = JsonOutputParser(pydantic_object=RouterResponse) # UNIFIED ROUTER CHAIN - Handles both intent classification AND API routing # self.router_prompt_template = PromptTemplate( # template=""" # You are a routing system. Analyze user intent and handle dates precisely. # CONTEXT: # Query: "{user_query}" # Language: {detected_language} # Current: {current_datetime} ({current_day_name}) # Timezone: {timezone} # ENDPOINTS: # {endpoints_documentation} # ANALYSIS STEPS: # 1. **Intent**: What does the user want? (translate Arabic to English first) # 2. **Date/Time**: Calculate precisely using current datetime as base # 3. **Match**: Find endpoint that can fulfill the request # 4. **Decision**: Matching endpoint = API_ACTION, else CONVERSATION # DATE CALCULATIONS (use {current_datetime} as base): # • Today/اليوم = current date # • Tomorrow/غدا = +1 day # • Next week/الأسبوع القادم = +7 days # • Next [weekday]/يوم [weekday] القادم: Calculate days to target weekday # - Weekdays: الأحد=0, الاثنين=1, الثلاثاء=2, الأربعاء=3, الخميس=4, الجمعة=5, السبت=6 # - Formula: If target > current: target-current, else: 7-(current-target) # • Times: صباحًا=09:00, مساءً=18:00, ظهرًا=12:00 # • Format: YYYY-MM-DDTHH:MM:SS # OUTPUT FORMAT: # {{ # "intent": "CONVERSATION|API_ACTION", # "confidence": 0.8, # "reasoning": "User wants: [need]. Date calc: [show work]. Endpoint: [path/reason]", # "endpoint": "/exact/path" or null, # "method": "GET|POST|PUT|DELETE" or null, # "params": {{ # // ALL VALUES IN ENGLISH - translate Arabic names/terms # }}, # "missing_required": [], # "calculated_datetime": "YYYY-MM-DDTHH:MM:SS" or null # }} # CRITICAL RULES: # • Use {current_datetime} for ALL date calculations # • Show calculation work in reasoning # • Translate ALL Arabic parameters to English # • Match endpoints by functionality, not keywords # Analyze and respond with JSON:""", # input_variables=["user_query", "detected_language", "extracted_keywords", # "sentiment_analysis", "endpoints_documentation", "current_datetime", # "timezone", "current_day_name"] # ) self.router_prompt_template = PromptTemplate( template=""" You are a medical appointment router. Determine if user needs API call or conversation. CONTEXT: Query: "{user_query}" | Language: {detected_language} | Current: {current_datetime} ENDPOINTS: {endpoints_documentation} DECISION RULES: API_ACTION only for these medical requests: - View appointments/reservations (my/all) - Book/cancel/reschedule appointments - Show doctors/hospitals/patients - Get doctor details CONVERSATION for everything else: - Greetings: "hello", "hi", "مرحبا" - General questions, help requests, small talk - Unclear/non-medical requests DATE PARSING (if needed): - Today/اليوم = current date - Tomorrow/غدا = +1 day - Times: صباحًا=09:00, مساءً=18:00, ظهرًا=12:00 - Format: YYYY-MM-DDTHH:MM:SS OUTPUT JSON: {{ "intent": "CONVERSATION|API_ACTION", "confidence": 0.9, "reasoning": "Brief explanation of decision", "endpoint": "/path" or null, "method": "GET|POST|PUT" or null, "params": {{ // ALL VALUES IN ENGLISH - translate Arabic names/terms }}, "calculated_datetime": "YYYY-MM-DDTHH:MM:SS" or null }} RULE: When uncertain, choose CONVERSATION. Translate Arabic terms to English in params. Analyze:""", input_variables=["user_query", "detected_language", "extracted_keywords", "sentiment_analysis", "endpoints_documentation", "current_datetime", "timezone", "current_day_name"] ) # self.router_prompt_template = PromptTemplate( # template=""" # You are a routing system. Analyze user intent and handle dates precisely. # CONTEXT: # Query: "{user_query}" # Language: {detected_language} # Current: {current_datetime} ({current_day_name}) # Timezone: {timezone} # ENDPOINTS: # {endpoints_documentation} # ANALYSIS STEPS: # 1. **Intent**: What does the user want? (translate Arabic to English first) # 2. **Translation**: Convert ALL Arabic text to English equivalents # 3. **Date/Time**: Calculate precisely using current datetime as base # 4. **Match**: Find endpoint that can fulfill the request # 5. **Decision**: Matching endpoint = API_ACTION, else CONVERSATION # TRANSLATION REQUIREMENTS: # • ALL parameter values MUST be in English # • Arabic names: محمد→Mohammed, أحمد→Ahmed, فاطمة→Fatima, علي→Ali, etc. # • Arabic terms: طبيب→doctor, مريض→patient, حجز→booking, موعد→appointment # • Arabic reasons: صداع→headache, حمى→fever, فحص→checkup, استشارة→consultation # • NO Arabic characters allowed in final params # DATE CALCULATIONS (use {current_datetime} as base): # • Today/اليوم = current date # • Tomorrow/غدا = +1 day # • Next week/الأسبوع القادم = +7 days # • Next [weekday]/يوم [weekday] القادم: Calculate days to target weekday # - Weekdays: الأحد=0, الاثنين=1, الثلاثاء=2, الأربعاء=3, الخميس=4, الجمعة=5, السبت=6 # - Formula: If target > current: target-current, else: 7-(current-target) # • Times: صباحًا=09:00, مساءً=18:00, ظهرًا=12:00 # • Format: YYYY-MM-DDTHH:MM:SS # PARAMETER VALIDATION: # Before outputting, verify each param value: # - Is it in English? ✓/✗ # - Contains Arabic characters? ✗ (reject if yes) # - Properly translated? ✓/✗ # OUTPUT FORMAT: # {{ # "intent": "CONVERSATION|API_ACTION", # "confidence": 0.8, # "reasoning": "User wants: [need in English]. Translation applied: [Arabic→English]. Date calc: [show work]. Endpoint: [path/reason]", # "endpoint": "/exact/path" or null, # "method": "GET|POST|PUT|DELETE" or null, # "params": {{ # // MANDATORY: ALL VALUES MUST BE IN ENGLISH # // Example: "doctor_name": "Mohammed" NOT "محمد" # }}, # "missing_required": [], # "calculated_datetime": "YYYY-MM-DDTHH:MM:SS" or null # }} # CRITICAL RULES: # • Use {current_datetime} for ALL date calculations # • MANDATORY: Translate ALL Arabic text to English in params # • Show translation work: "محمد→Mohammed" in reasoning # • Reject output if ANY param contains Arabic characters # • Match endpoints by functionality, not keywords # VALIDATION CHECK: # Before final output, ask: "Are ALL param values in English?" If NO, translate them. # Analyze and respond with JSON:""", # input_variables=["user_query", "detected_language", "extracted_keywords", # "sentiment_analysis", "endpoints_documentation", "current_datetime", # "timezone", "current_day_name"] # ) # CONVERSATION CHAIN - Handles conversational responses self.conversation_template = PromptTemplate( template=""" You are a friendly healthcare chatbot assistant. CONTEXT: Message: {user_query} Language: {detected_language} Sentiment: {sentiment_analysis} RESPONSE GUIDELINES: • Respond ONLY in {detected_language} • Be helpful, empathetic, and professional • Keep responses concise but informative • Use caring and supportive tone LANGUAGE SPECIFICS: Arabic: Use Modern Standard Arabic (الفصحى), formal tone, proper medical terms English: Clear professional language, warm and approachable RULES: 1. Address user's question directly 2. Provide helpful information when possible 3. Never give specific medical advice - recommend healthcare professionals 4. Be encouraging and supportive 5. Don't mix languages 6. End naturally without multiple questions Generate a helpful response:""", input_variables=["user_query", "detected_language", "sentiment_analysis", "conversation_history"] ) self.api_response_template = PromptTemplate( template=""" You are a friendly healthcare assistant. Answer using API data in {detected_language}. CONTEXT: Query: {user_query} API Data: {api_response} DATE/TIME PARSING: Extract from API format "YYYY-MM-DDTHH:MM:SS": - Parse: Year(0-3), Month(5-6), Day(8-9), Hour(11-12), Minute(14-15) - Months: 01=Jan/يناير, 02=Feb/فبراير, 03=Mar/مارس, 04=Apr/أبريل, 05=May/مايو, 06=Jun/يونيو, 07=Jul/يوليو, 08=Aug/أغسطس, 09=Sep/سبتمبر, 10=Oct/أكتوبر, 11=Nov/نوفمبر, 12=Dec/ديسمبر - Time: 00-11=AM/صباحاً, 12=12PM/١٢ظهراً, 13-23=subtract 12+PM/مساءً STEP-BY-STEP PROCESS: 1. Find date field in API data 2. Extract the datetime string (YYYY-MM-DDTHH:MM:SS format) 3. Parse each component using the rules above 4. Format according to language requirements RULES: - CRITICAL: Read date field from API data, ignore any example dates - Parse the ACTUAL datetime string from the API response - Use ONLY API data, never make up information - Friendly tone with starters Generate response:""", input_variables=["user_query", "api_response", "detected_language", "sentiment_analysis"] ) # last one # self.api_response_template = PromptTemplate( # template=""" # You are a friendly healthcare assistant. Answer using the API data provided. # CONTEXT: # Query: {user_query} # Language: {detected_language} # API Data: {api_response} # INSTRUCTIONS: # • Use ONLY actual API data - never make up information # • Respond in {detected_language} only # • Sound warm and conversational, like talking to a friend # • Convert technical data to simple language # DATE/TIME FORMAT: # '2025-05-30T10:28:10' → # • English: "May 30, 2025 at 10:28 AM" # • Arabic: "٣٠ مايو ٢٠٢٥ في الساعة ١٠:٢٨ صباحاً" # TONE: # • Friendly starters: "Great!", "Perfect!", "ممتاز!", "رائع!" # • Reassuring: "Everything looks good", "كل شيء جاهز" # • Natural conversation, not robotic # LANGUAGE SPECIFICS: # Arabic: Use Arabic numerals (٠١٢٣٤٥٦٧٨٩) and month names # English: 12-hour format with AM/PM # EXAMPLES: # Appointment confirmed: # • English: "Great! Your appointment is set for May 30, 2025 at 10:28 AM!" # • Arabic: "ممتاز! موعدك محجوز يوم ٣٠ مايو ٢٠٢٥ الساعة ١٠:٢٨ صباحاً!" # Generate a friendly response using the API data:""", # input_variables=["user_query", "api_response", "detected_language", "sentiment_analysis"] # ) # first one to be removed # self.api_response_template = PromptTemplate( # template=""" # You are a professional healthcare assistant. Generate a natural language response to the user's query using ONLY the provided API data. # User Query: {user_query} # User Sentiment: {sentiment_analysis} # Response Language: {detected_language} # API Response Data: # {api_response} # === CORE INSTRUCTIONS === # 1. Analyze the API response structure and extract relevant data points # 2. Cross-reference with the user's query to determine what information to include # 3. Respond in {detected_language} using a warm, conversational tone # 4. Convert technical data into natural language appropriate for healthcare communication # === DATE/TIME HANDLING === # 1. Identify all date/time fields in the API response (look for ISO 8601 format: YYYY-MM-DDTHH:MM:SS) # 2. For English responses: # - Format dates as "Month Day, Year at HH:MM AM/PM" # - Convert times to 12-hour format with proper AM/PM # 3. For Arabic responses: # - Format dates as "Day Month Year الساعة HH:MM صباحاً/مساءً" # - Use Arabic numerals (٠١٢٣٤٥٦٧٨٩) # - Use Arabic month names # 4. Preserve all original date/time values - only change the formatting # === RESPONSE GUIDELINES === # 1. Use ONLY data present in the API response # 2. Maintain a professional yet friendly healthcare tone # 3. Adapt to the user's sentiment: # - Positive: reinforce with encouraging language # - Neutral: provide clear, factual information # - Negative: show empathy and offer assistance # 4. Structure the response to directly answer the user's query # 5. Include relevant details from the API response that address the user's needs # === CRITICAL RULES === # 1. Never invent or hallucinate information not present in the API response # 2. If the API response doesn't contain requested information, say so politely # 3. All dates/times must exactly match the API data # 4. Maintain strict language consistency (respond only in {detected_language}) # 5. Format all technical data (IDs, codes, etc.) for easy understanding # Generate a helpful response that addresses the user's query using the API data. # """, # input_variables=["user_query", "api_response", "detected_language", "sentiment_analysis"] # ) # Create the 3 chains self.router_chain = LLMChain(llm=self.llm, prompt=self.router_prompt_template) self.conversation_chain = LLMChain(llm=self.llm, prompt=self.conversation_template) self.api_response_chain = LLMChain(llm=self.llm, prompt=self.api_response_template) def detect_language(self, text): """Detect language of the input text""" if self.language_classifier and len(text.strip()) > 3: try: result = self.language_classifier(text) detected_lang = result[0][0]['label'] confidence = result[0][0]['score'] if detected_lang in ['ar', 'arabic']: return "arabic" elif detected_lang in ['en', 'english']: return "english" elif confidence > 0.8: return "english" # Default to English for unsupported languages except: pass # Fallback: Basic Arabic detection arabic_pattern = re.compile(r'[\u0600-\u06FF\u0750-\u077F\u08A0-\u08FF]+') if arabic_pattern.search(text): return "arabic" return "english" def analyze_sentiment(self, text): """Analyze sentiment of the text""" if self.sentiment_analyzer and len(text.strip()) > 3: try: result = self.sentiment_analyzer(text) return { "sentiment": result[0]['label'], "score": result[0]['score'] } except: pass return {"sentiment": "NEUTRAL", "score": 0.5} def extract_keywords(self, text): """Extract keywords from text""" # Simple keyword extraction words = re.findall(r'\b\w+\b', text.lower()) # Filter out common words and keep meaningful ones stopwords = {'the', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', 'of', 'with', 'by', 'is', 'are', 'was', 'were'} keywords = [w for w in words if len(w) > 3 and w not in stopwords] return list(set(keywords))[:5] # Return top 5 unique keywords def get_conversation_context(self): """Get recent conversation history as context""" if not self.conversation_history: return "No previous conversation" context = [] for item in self.conversation_history[-3:]: # Last 3 exchanges context.append(f"User: {item['user_message']}") context.append(f"Bot: {item['bot_response'][:100]}...") # Truncate long responses return " | ".join(context) def add_to_history(self, user_message, bot_response, response_type): """Add exchange to conversation history""" self.conversation_history.append({ 'timestamp': datetime.now(), 'user_message': user_message, 'bot_response': bot_response, 'response_type': response_type }) # Keep only recent history if len(self.conversation_history) > self.max_history_length: self.conversation_history = self.conversation_history[-self.max_history_length:] def parse_relative_date(self, text, detected_language): """Parse relative dates from text using a combination of methods""" today = datetime.now() # Handle common relative date patterns in English and Arabic tomorrow_patterns = { 'english': [r'\btomorrow\b', r'\bnext day\b'], 'arabic': [r'\bغدا\b', r'\bبكرة\b', r'\bغدًا\b', r'\bالغد\b'] } next_week_patterns = { 'english': [r'\bnext week\b'], 'arabic': [r'\bالأسبوع القادم\b', r'\bالأسبوع المقبل\b', r'\bالاسبوع الجاي\b'] } # Check for "tomorrow" patterns for pattern in tomorrow_patterns.get(detected_language, []) + tomorrow_patterns.get('english', []): if re.search(pattern, text, re.IGNORECASE): return (today + timedelta(days=1)).strftime('%Y-%m-%dT%H:%M:%S') # Check for "next week" patterns for pattern in next_week_patterns.get(detected_language, []) + next_week_patterns.get('english', []): if re.search(pattern, text, re.IGNORECASE): return (today + timedelta(days=7)).strftime('%Y-%m-%dT%H:%M:%S') # If NER model is available, use it to extract date entities if self.date_parser and detected_language == 'english': try: date_entities = self.date_parser(text) for entity in date_entities: if entity['entity_group'] == 'DATE': print(f"Found date entity: {entity['word']}") # Default to tomorrow if we detect any date return (today + timedelta(days=1)).strftime('%Y-%m-%dT%H:%M:%S') except Exception as e: print(f"Error in date parsing: {e}") # Default return None if no date pattern is recognized return None def parse_router_response(self, router_text): """Parse the router chain response into structured data""" try: # Clean the response text cleaned_response = router_text # Remove any comments (both single-line and multi-line) cleaned_response = re.sub(r'//.*?$', '', cleaned_response, flags=re.MULTILINE) cleaned_response = re.sub(r'/\*.*?\*/', '', cleaned_response, flags=re.DOTALL) # Remove any trailing commas cleaned_response = re.sub(r',(\s*[}\]])', r'\1', cleaned_response) # Try different methods to parse the JSON response try: # First attempt: direct JSON parsing of cleaned response parsed_response = json.loads(cleaned_response) except json.JSONDecodeError: try: # Second attempt: extract JSON from markdown code block json_match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', cleaned_response, re.DOTALL) if json_match: parsed_response = json.loads(json_match.group(1)) else: raise ValueError("No JSON found in code block") except (json.JSONDecodeError, ValueError): try: # Third attempt: find JSON-like content using regex json_pattern = r'\{\s*"intent"\s*:.*?\}' json_match = re.search(json_pattern, cleaned_response, re.DOTALL) if json_match: json_str = json_match.group(0) # Additional cleaning for the extracted JSON json_str = re.sub(r'//.*?$', '', json_str, flags=re.MULTILINE) json_str = re.sub(r',(\s*[}\]])', r'\1', json_str) parsed_response = json.loads(json_str) else: raise ValueError("Could not extract JSON using regex") except (json.JSONDecodeError, ValueError): print(f"Failed to parse JSON. Raw response: {router_text}") print(f"Cleaned response: {cleaned_response}") # Return default conversation response on parse failure return { "intent": "CONVERSATION", "confidence": 0.5, "reasoning": "Failed to parse router response - defaulting to conversation", "endpoint": None, "method": None, "params": {}, "missing_required": [] } # Validate required fields and set defaults validated_response = { "intent": parsed_response.get("intent", "CONVERSATION"), "confidence": parsed_response.get("confidence", 0.5), "reasoning": parsed_response.get("reasoning", "Router decision"), "endpoint": parsed_response.get("endpoint"), "method": parsed_response.get("method"), "params": parsed_response.get("params", {}), "missing_required": parsed_response.get("missing_required", []) } return validated_response except Exception as e: print(f"Error parsing router response: {e}") return { "intent": "CONVERSATION", "confidence": 0.5, "reasoning": f"Parse error: {str(e)}", "endpoint": None, "method": None, "params": {}, "missing_required": [] } def handle_conversation(self, user_query, detected_language, sentiment_result): """Handle conversational responses""" try: result = self.conversation_chain.invoke({ "user_query": user_query, "detected_language": detected_language, "sentiment_analysis": json.dumps(sentiment_result), "conversation_history": self.get_conversation_context() }) return result["text"].strip() except Exception as e: # Fallback response if detected_language == "arabic": return "أعتذر، واجهت مشكلة في المعالجة. كيف يمكنني مساعدتك؟" else: return "I apologize, I encountered a processing issue. How can I help you?" def backend_call(self, data: Dict[str, Any]) -> Dict[str, Any]: """Make API call to backend with retry logic""" endpoint_url = data.get('endpoint') endpoint_method = data.get('method') endpoint_params = data.get('params', {}).copy() print(f"🔗 Making API call to {endpoint_method} {self.BASE_URL + endpoint_url} with params: {endpoint_params}") # Inject patient_id if needed if 'patient_id' in endpoint_params: endpoint_params['patient_id'] = self.user_id retries = 0 response = None while retries < self.max_retries: try: if endpoint_method.upper() == 'GET': response = requests.get( self.BASE_URL + endpoint_url, params=endpoint_params, headers=self.headers, timeout=10 ) elif endpoint_method.upper() in ['POST', 'PUT', 'DELETE']: response = requests.request( endpoint_method.upper(), self.BASE_URL + endpoint_url, json=endpoint_params, headers=self.headers, timeout=10 ) response.raise_for_status() print('Backend Response:', response.json()) return response.json() except requests.exceptions.RequestException as e: retries += 1 if retries >= self.max_retries: return { "error": "Backend API call failed after multiple retries", "details": str(e), "status_code": getattr(e.response, 'status_code', None) if hasattr(e, 'response') else None } time.sleep(self.retry_delay) # async with aiohttp.ClientSession() as session: # while retries < self.max_retries: # try: # if endpoint_method.upper() == 'GET': # response = await session.get( # self.BASE_URL + endpoint_url, # params=endpoint_params, # headers=self.headers, # timeout=aiohttp.ClientTimeout(total=10) # ) # return await response.json() # elif endpoint_method.upper() in ['POST', 'PUT', 'DELETE']: # response = await session.request( # endpoint_method.upper(), # self.BASE_URL + endpoint_url, # json=endpoint_params, # headers=self.headers, # timeout=aiohttp.ClientTimeout(total=10) # ) # return await response.json() # except aiohttp.ClientResponseError as e: # retries += 1 # if retries >= self.max_retries: # return { # "error": "Backend API call failed after multiple retries", # "details": str(e), # "status_code": e.status # } # await asyncio.sleep(self.retry_delay) # except (aiohttp.ClientError, asyncio.TimeoutError) as e: # retries += 1 # if retries >= self.max_retries: # return { # "error": "Backend API call failed after multiple retries", # "details": str(e), # "status_code": None # } # await asyncio.sleep(self.retry_delay) def handle_api_action(self, user_query, detected_language, sentiment_result, keywords, router_data): """Handle API-based actions using router data""" try: # Parse relative dates and inject into parameters # parsed_date = self.parse_relative_date(user_query, detected_language) # if parsed_date: # print(f"Parsed relative date: {parsed_date}") # # Inject parsed date if available and a date parameter exists # date_params = ['appointment_date', 'date', 'schedule_date', 'date_time', 'new_date_time'] # for param in date_params: # if param in router_data['params']: # router_data['params'][param] = parsed_date # Inject patient_id if needed if 'patient_id' in router_data['params']: router_data['params']['patient_id'] = self.user_id else: router_data['params']['patient_id'] = self.user_id print(f"🔍 Final API call data: {router_data}") # Make backend API call try: api_response = self.backend_call(router_data) # api_response = asyncio.run(self.backend_call(router_data)) # loop = asyncio.get_event_loop() # api_response = loop.run_until_complete(self.backend_call(router_data)) # loop.close() except: print(traceback.format_exc()) # try: # def run_async(): # new_loop = asyncio.new_event_loop() # asyncio.set_event_loop(new_loop) # try: # return new_loop.run_until_complete(self.backend_call(router_data)) # finally: # new_loop.close() # import concurrent.futures # with concurrent.futures.ThreadPoolExecutor() as executor: # future = executor.submit(run_async) # api_response = future.result(timeout=30) # 30 second timeout # except: # print(traceback.format_exc()) print("🔗 API response received:", api_response) # Generate user-friendly response user_response_result = self.api_response_chain.invoke({ "user_query": user_query, "api_response": json.dumps(api_response, indent=2), "detected_language": detected_language, "sentiment_analysis": json.dumps(sentiment_result), }) print("🔗 Final user response:", user_response_result["text"].strip()) return { "response": user_response_result["text"].strip(), "api_data": api_response, "routing_info": router_data } except Exception as e: # Fallback error response if detected_language == "arabic": error_msg = "أعتذر، لم أتمكن من معالجة طلبك. يرجى المحاولة مرة أخرى أو صياغة السؤال بطريقة مختلفة." else: error_msg = "I apologize, I couldn't process your request. Please try again or rephrase your question." return { "response": error_msg, "api_data": {"error": str(e)}, "routing_info": None } def chat(self, user_message: str) -> ChatResponse: """Main chat method that handles user messages - REVAMPED to use 3 chains""" start_time = time.time() # Check for exit commands if user_message.lower().strip() in ['quit', 'exit', 'خروج', 'bye', 'goodbye']: if self.detect_language(user_message) == "arabic": return ChatResponse( response_id=str(time.time()), response_type="conversation", message="مع السلامة! أتمنى لك يوماً سعيداً. 👋", language="arabic" ) else: return ChatResponse( response_id=str(time.time()), response_type="conversation", message="Goodbye! Have a great day! 👋", language="english" ) try: print(f"\n{'='*50}") print(f"🔍 Processing: '{user_message}'") print(f"{'='*50}") # Step 1: Language and sentiment analysis detected_language = self.detect_language(user_message) sentiment_result = self.analyze_sentiment(user_message) keywords = self.extract_keywords(user_message) print(f"🌐 Detected Language: {detected_language}") print(f"😊 Sentiment: {sentiment_result}") print(f"🔑 Keywords: {keywords}") # Step 2: Router Chain - Determine intent and route appropriately print(f"\n🤖 Running Router Chain...") router_result = self.router_chain.invoke({ "user_query": user_message, "detected_language": detected_language, "extracted_keywords": json.dumps(keywords), "sentiment_analysis": json.dumps(sentiment_result), "conversation_history": self.get_conversation_context(), "endpoints_documentation": self.cached_endpoints_documentation, "current_datetime": datetime.now().strftime('%Y-%m-%dT%H:%M:%S'), "timezone": "UTC", "current_day_name": datetime.now().strftime('%A'), }) # Parse router response router_data = self.parse_router_response(router_result["text"]) print(f"🎯 Router Decision: {router_data}") # Step 3: Handle based on intent if router_data["intent"] == "CONVERSATION": print(f"\n💬 Handling as CONVERSATION") response_text = self.handle_conversation(user_message, detected_language, sentiment_result) # Add to conversation history self.add_to_history(user_message, response_text, "conversation") return ChatResponse( response_id=str(time.time()), response_type="conversation", message=response_text, api_call_made=False, language=detected_language, api_data=None ) else: print(f"\n🔗 Handling as API_ACTION") # Check for missing required parameters # if router_data.get("missing_required"): # missing_params = router_data["missing_required"] # if detected_language == "arabic": # response_text = f"أحتاج إلى مزيد من المعلومات: {', '.join(missing_params)}" # else: # response_text = f"I need more information: {', '.join(missing_params)}" # return ChatResponse( # response_id=str(time.time()), # response_type="conversation", # message=response_text, # api_call_made=False, # language=detected_language # ) # Handle API action api_result = self.handle_api_action( user_message, detected_language, sentiment_result, keywords, router_data ) # Add to conversation history self.add_to_history(user_message, api_result["response"], "api_action") return ChatResponse( response_id=str(time.time()), response_type="api_action", message=api_result["response"], api_call_made=True, # api_data=api_result["api_data"] # api_data=json.dumps(api_result["api_data"]) if 'action_result' in api_result else None, language=detected_language ) # else: # # Fallback for unknown intent # print(f"⚠️ Unknown intent: {router_data['intent']}") # fallback_response = self.handle_conversation(user_message, detected_language, sentiment_result) # return ChatResponse( # response_id=str(time.time()), # response_type="conversation", # message=fallback_response, # api_call_made=False, # language=detected_language # ) except Exception as e: print(f"❌ Error in chat method: {str(e)}") print(f"❌ Traceback: {traceback.format_exc()}") # Fallback error response if self.detect_language(user_message) == "arabic": error_message = "أعتذر، حدث خطأ في معالجة رسالتك. يرجى المحاولة مرة أخرى." else: error_message = "I apologize, there was an error processing your message. Please try again." return ChatResponse( response_id=str(time.time()), response_type="conversation", message=error_message, api_call_made=False, language=self.detect_language(user_message) ) finally: end_time = time.time() print(f"⏱️ Processing time: {end_time - start_time:.2f} seconds") def run_interactive_chat(self): """Run the interactive chat interface""" try: while True: try: # Get user input user_input = input("\n👤 You: ").strip() if not user_input: continue # Process the message response = self.chat(user_input) # Display the response print(f"\n🤖 Bot: {response.message}") # Check for exit if user_input.lower() in ['quit', 'exit', 'خروج', 'bye', 'goodbye']: break except KeyboardInterrupt: print("\n\n👋 Chat interrupted. Goodbye!") break except EOFError: print("\n\n👋 Chat ended. Goodbye!") break except Exception as e: print(f"\n❌ Error: {e}") continue except Exception as e: print(f"❌ Fatal error in chat interface: {e}") def clear_history(self): """Clear conversation history""" self.conversation_history = [] print("🗑️ Conversation history cleared.") # def main(): # """Main function to run the healthcare chatbot""" # try: # print("🚀 Starting Healthcare Chatbot...") # chatbot = HealthcareChatbot() # chatbot.run_interactive_chat() # except KeyboardInterrupt: # print("\n\n👋 Shutting down gracefully...") # except Exception as e: # print(f"❌ Fatal error: {e}") # print(f"❌ Traceback: {traceback.format_exc()}") # if __name__ == "__main__": # main() from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Dict, Any, Optional app = FastAPI( title="Healthcare AI Assistant", description="An AI-powered healthcare assistant that handles appointment booking and queries", version="1.0.0" ) # Initialize the AI agent agent = HealthcareChatbot() class QueryRequest(BaseModel): query: str @app.post("/query") async def process_query(request: QueryRequest): """ Process a user query and return a response """ try: response = agent.chat(request.query).message return response except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/health") async def health_check(): """ Health check endpoint """ return {"status": "healthy", "service": "healthcare-ai-assistant"} @app.get("/") async def root(): return {"message": "Hello World"} # if __name__ == "__main__": # import uvicorn # uvicorn.run(app, host="0.0.0.0", port=8000)