feat(user_prompt): update UserPromptProcessor to integrate Llama3-Med42-70B and enhance query validation; add unit tests for condition extraction and matching mechanisms

.gitignore

@@ -37,6 +37,7 @@ __pycache__/
 *.pem
 credentials.json
 token.json
+*.mdc
 
 # 🚫 Large files - models
 models/cache/

src/__init__.py

@@ -3,6 +3,16 @@ OnCall.ai src package
 
 This package contains the core implementation of the OnCall.ai system.
 """
- 
+
 # Version
-__version__ = '0.1.0' 
+__version__ = '0.1.0'
+
+# import key modules
+from .llm_clients import llm_Med42_70BClient
+from .user_prompt import UserPromptProcessor
+from .retrieval import BasicRetrievalSystem
+from .medical_conditions import (
+    CONDITION_KEYWORD_MAPPING,
+    get_condition_keywords,
+    validate_condition
+) 

src/llm_clients.py

@@ -9,91 +9,71 @@ Date: 2025-07-29
 
 import logging
 import os
-from typing import Dict, Optional
-import torch
-from transformers import AutoTokenizer, AutoModelForCausalLM
+from typing import Dict, Optional, Union
 from huggingface_hub import InferenceClient
 from dotenv import load_dotenv
 
 # Load environment variables from .env file
 load_dotenv()
 
-class MeditronClient:
+class llm_Med42_70BClient:
     def __init__(
         self, 
-        model_name: str = "TheBloke/meditron-7B-GPTQ",
-        local_model_path: Optional[str] = None,
-        use_local: bool = False,
+        model_name: str = "m42-health/Llama3-Med42-70B",
         timeout: float = 30.0
     ):
         """
-        Initialize Meditron client for medical query processing.
+        Initialize Medical LLM client for query processing.
         
         Args:
             model_name: Hugging Face model name
-            local_model_path: Path to local model files
-            use_local: Flag to use local model
             timeout: API call timeout duration
         
         Warning: This model should not be used for professional medical advice.
         """
         self.logger = logging.getLogger(__name__)
         self.timeout = timeout
-        self.use_local = use_local
         
-        if use_local:
-            if not local_model_path:
-                raise ValueError("local_model_path must be provided when use_local is True")
-            
-            try:
-                # Load local model using Hugging Face transformers
-                self.tokenizer = AutoTokenizer.from_pretrained(
-                    model_name, 
-                    local_files_only=True, 
-                    cache_dir=local_model_path
-                )
-                self.model = AutoModelForCausalLM.from_pretrained(
-                    model_name,
-                    local_files_only=True,
-                    cache_dir=local_model_path,
-                    device_map="auto",
-                    torch_dtype=torch.float16
-                )
-                
-                self.logger.info(f"Local Meditron model loaded from: {local_model_path}")
-                self.logger.warning(
-                    "Meditron Model: Research tool only. "
-                    "Not for professional medical diagnosis."
-                )
-            except Exception as e:
-                self.logger.error(f"Failed to load local model: {str(e)}")
-                raise ValueError(f"Failed to initialize local Meditron client: {str(e)}")
-        else:
-            # Existing InferenceClient logic
-            hf_token = os.getenv('HF_TOKEN')
-            if not hf_token:
-                raise ValueError(
-                    "HF_TOKEN not found in environment variables. "
-                    "Please set HF_TOKEN in your .env file or environment."
-                )
+        # Configure logging to show detailed information
+        logging.basicConfig(
+            level=logging.INFO, 
+            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
+        )
+        
+        # Get Hugging Face token from environment
+        hf_token = os.getenv('HF_TOKEN')
+        if not hf_token:
+            self.logger.error("HF_TOKEN is missing from environment variables.")
+            raise ValueError(
+                "HF_TOKEN not found in environment variables. "
+                "Please set HF_TOKEN in your .env file or environment. "
+                "Ensure the token is not empty and is correctly set."
+            )
+        
+        try:
+            # Initialize InferenceClient with the new model
+            self.client = InferenceClient(
+                provider="featherless-ai", 
+                api_key=hf_token
+            )
             
-            try:
-                self.client = InferenceClient(model=model_name, token=hf_token)
-                self.logger.info(f"Meditron client initialized with model: {model_name}")
-                self.logger.warning(
-                    "Meditron Model: Research tool only. "
-                    "Not for professional medical diagnosis."
-                )
-            except Exception as e:
-                self.logger.error(f"Failed to initialize InferenceClient: {str(e)}")
-                raise ValueError(f"Failed to initialize Meditron client: {str(e)}")
+            self.logger.info(f"Medical LLM client initialized with model: {model_name}")
+            self.logger.warning(
+                "Medical LLM Model: Research tool only. "
+                "Not for professional medical diagnosis."
+            )
+        except Exception as e:
+            self.logger.error(f"Failed to initialize InferenceClient: {str(e)}")
+            self.logger.error(f"Error Type: {type(e).__name__}")
+            self.logger.error(f"Detailed Error: {repr(e)}")
+            raise ValueError(f"Failed to initialize Medical LLM client: {str(e)}") from e
 
     def analyze_medical_query(
         self, 
         query: str, 
         max_tokens: int = 100, 
         timeout: Optional[float] = None
-    ) -> Dict[str, str]:
+    ) -> Dict[str, Union[str, float]]:
         """
         Analyze medical query and extract condition.
         
@@ -103,82 +83,74 @@ class MeditronClient:
             timeout: Specific API call timeout
         
         Returns:
-            Extracted medical condition information
+            Extracted medical condition information with latency
         """
+        import time
+        
+        # Start timing
+        start_time = time.time()
+        
         try:
-            # ChatML style prompt for Meditron
-            prompt = f"""<|im_start|>system
-You are a professional medical assistant trained to extract medical conditions. 
-Provide only the most representative condition name.
-DO NOT provide medical advice.
-<|im_end|>
-<|im_start|>user
-{query}
-<|im_end|>
-<|im_start|>assistant
-"""
-            
-            self.logger.info(f"Calling Meditron with query: {query}")
+            self.logger.info(f"Calling Medical LLM with query: {query}")
             
-            if self.use_local:
-                # Local model inference
-                input_ids = self.tokenizer(prompt, return_tensors='pt').input_ids.to(self.model.device)
-                
-                response = self.model.generate(
-                    input_ids, 
-                    max_new_tokens=max_tokens, 
-                    temperature=0.7, 
-                    do_sample=True, 
-                    top_k=50
-                )
-                
-                response_text = self.tokenizer.decode(response[0], skip_special_tokens=True)
-                self.logger.info(f"Local model response: {response_text}")
-            else:
-                # InferenceClient inference
-                self.logger.info(f"Using model: {self.client.model}")
-                
-                # Test API connection first
-                try:
-                    test_response = self.client.text_generation(
-                        "Hello",
-                        max_new_tokens=5,
-                        temperature=0.7,
-                        top_k=50
-                    )
-                    self.logger.info("API connection test successful")
-                except Exception as test_error:
-                    self.logger.error(f"API connection test failed: {str(test_error)}")
-                    return {
-                        'extracted_condition': '',
-                        'confidence': 0,
-                        'error': f"API connection failed: {str(test_error)}"
+            # Prepare chat completion request
+            response = self.client.chat.completions.create(
+                model="m42-health/Llama3-Med42-70B",
+                messages=[
+                    {
+                        "role": "system", 
+                        "content": "You are a professional medical assistant trained to extract medical conditions. Provide only the most representative condition name. DO NOT provide medical advice."
+                    },
+                    {
+                        "role": "user", 
+                        "content": query
                     }
-                
-                response_text = self.client.text_generation(
-                    prompt,
-                    max_new_tokens=max_tokens,
-                    temperature=0.7,
-                    top_k=50
-                )
+                ],
+                max_tokens=max_tokens
+            )
+            
+            # Calculate latency
+            end_time = time.time()
+            latency = end_time - start_time
+            
+            # Extract the response text
+            response_text = response.choices[0].message.content or ""
+            
+            # Log raw response and latency
+            self.logger.info(f"Raw LLM Response: {response_text}")
+            self.logger.info(f"Query Latency: {latency:.4f} seconds")
             
             # Extract condition from response
             extracted_condition = self._extract_condition(response_text)
             
+            # Log the extracted condition
+            self.logger.info(f"Extracted Condition: {extracted_condition}")
+            
             return {
                 'extracted_condition': extracted_condition,
-                'confidence': 0.8,
-                'raw_response': response_text
+                'confidence': '0.8',
+                'raw_response': response_text,
+                'latency': latency  # Add latency to the return dictionary
             }
         
         except Exception as e:
-            self.logger.error(f"Meditron query error: {str(e)}")
-            self.logger.error(f"Error type: {type(e).__name__}")
-            self.logger.error(f"Error details: {repr(e)}")
+            # Calculate latency even for failed requests
+            end_time = time.time()
+            latency = end_time - start_time
+            
+            self.logger.error(f"Medical LLM query error: {str(e)}")
+            self.logger.error(f"Error Type: {type(e).__name__}")
+            self.logger.error(f"Detailed Error: {repr(e)}")
+            self.logger.error(f"Query Latency (on error): {latency:.4f} seconds")
+            
+            # Additional context logging
+            self.logger.error(f"Query that caused error: {query}")
+            
             return {
                 'extracted_condition': '',
-                'confidence': 0,
-                'error': f"{type(e).__name__}: {str(e)}"
+                'confidence': '0',
+                'error': str(e),
+                'latency': latency  # Include latency even for error cases
             }
 
     def _extract_condition(self, response: str) -> str:
@@ -193,26 +165,29 @@ DO NOT provide medical advice.
         """
         from medical_conditions import CONDITION_KEYWORD_MAPPING
         
-        # Remove prompt parts, keep only generated content
-        generated_text = response.split('<|im_start|>assistant\n')[-1].strip()
-        
         # Search in known medical conditions
         for condition in CONDITION_KEYWORD_MAPPING.keys():
-            if condition.lower() in generated_text.lower():
+            if condition.lower() in response.lower():
                 return condition
         
-        return generated_text.split('\n')[0].strip()
+        return response.split('\n')[0].strip() or ""
 
 def main():
     """
-    Test Meditron client functionality
+    Test Medical LLM client functionality
     """
+    import time
+    from datetime import datetime
+
+    # Record total execution start time
+    total_start_time = time.time()
+    execution_start_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    
     try:
-        # Test local model loading
-        client = MeditronClient(
-            local_model_path="/Users/yanbochen/Documents/Life in Canada/CS study related/*Student Course, Guide/CS7180 GenAI/FinalProject_git_copy/models/cache/meditron-7B-GPTQ", 
-            use_local=True
-        )
+        print(f"Execution Started at: {execution_start_timestamp}")
+        
+        # Test client initialization
+        client = llm_Med42_70BClient()
         
         test_queries = [
             "patient experiencing chest pain",
@@ -220,24 +195,66 @@ def main():
             "severe headache with neurological symptoms"
         ]
         
+        # Store individual query results
+        query_results = []
+        
         for query in test_queries:
             print(f"\nTesting query: {query}")
             result = client.analyze_medical_query(query)
-            print("Extracted Condition:", result['extracted_condition'])
-            print("Confidence:", result['confidence'])
+            
+            # Store query result
+            query_result = {
+                'query': query,
+                'extracted_condition': result.get('extracted_condition', 'N/A'),
+                'confidence': result.get('confidence', 'N/A'),
+                'latency': result.get('latency', 'N/A')
+            }
+            query_results.append(query_result)
+            
+            # Print individual query results
+            print("Extracted Condition:", query_result['extracted_condition'])
+            print("Confidence:", query_result['confidence'])
+            print(f"Latency: {query_result['latency']:.4f} seconds")
+            
             if 'error' in result:
                 print("Error:", result['error'])
             print("---")
+        
+        # Calculate total execution time
+        total_end_time = time.time()
+        total_execution_time = total_end_time - total_start_time
+        execution_end_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        
+        # Print summary
+        print("\n--- Execution Summary ---")
+        print(f"Execution Started at: {execution_start_timestamp}")
+        print(f"Execution Ended at: {execution_end_timestamp}")
+        print(f"Total Execution Time: {total_execution_time:.4f} seconds")
+        
+        # Optional: Return results for potential further processing
+        return {
+            'start_time': execution_start_timestamp,
+            'end_time': execution_end_timestamp,
+            'total_execution_time': total_execution_time,
+            'query_results': query_results
+        }
             
     except Exception as e:
         print(f"Client initialization error: {str(e)}")
-        print("This might be due to:")
-        print("1. Incorrect local model path")
-        print("2. Missing dependencies")
-        print("3. Hardware limitations")
-        print("\nTo fix:")
-        print("1. Verify local model path")
-        print("2. Install required dependencies")
+        print("Possible issues:")
+        print("1. Invalid or missing Hugging Face token")
+        print("2. Network connectivity problems")
+        print("3. Model access restrictions")
+        print("\nPlease check your .env file and Hugging Face token.")
+        
+        # Calculate total execution time even in case of error
+        total_end_time = time.time()
+        total_execution_time = total_end_time - total_start_time
+        
+        return {
+            'error': str(e),
+            'total_execution_time': total_execution_time
+        }
 
 if __name__ == "__main__":
     main()

src/medical_conditions.py

@@ -26,7 +26,7 @@ CONDITION_KEYWORD_MAPPING: Dict[str, Dict[str, str]] = {
         "emergency": "chest pain|shortness of breath|sudden dyspnea",
         "treatment": "anticoagulation|heparin|embolectomy"
     },
-    # 從 @20250729Test_Retrieval.md 擴展的條件
+    # extended from @20250729Test_Retrieval.md 
     "acute_ischemic_stroke": {
         "emergency": "ischemic stroke|neurological deficit",
         "treatment": "tPA|stroke unit management"

src/retrieval.py

@@ -368,4 +368,22 @@ class BasicRetrievalSystem:
         
         except Exception as e:
             logger.error(f"Sliding window search failed: {e}")
+            return [] 
+
+    def search_generic_medical_content(self, query: str, top_k: int = 5) -> List[Dict]:
+        """
+        Perform generic medical content search
+        
+        Args:
+            query: Search query
+            top_k: Number of top results to return
+        
+        Returns:
+            List of search results
+        """
+        try:
+            # re-use search_sliding_window_chunks method
+            return self.search_sliding_window_chunks(query, top_k=top_k)
+        except Exception as e:
+            logger.error(f"Generic medical content search error: {e}")
             return [] 

src/user_prompt.py

@@ -34,15 +34,15 @@ logging.basicConfig(
 logger = logging.getLogger(__name__)
 
 class UserPromptProcessor:
-    def __init__(self, meditron_client=None, retrieval_system=None):
+    def __init__(self, llm_client=None, retrieval_system=None):
         """
-        Initialize UserPromptProcessor with optional Meditron and retrieval system
+        Initialize UserPromptProcessor with optional LLM and retrieval system
         
         Args:
-            meditron_client: Optional Meditron client for advanced condition extraction
+            llm_client: Optional Llama3-Med42-70B client for advanced condition extraction
             retrieval_system: Optional retrieval system for semantic search
         """
-        self.meditron_client = meditron_client
+        self.llm_client = llm_client
         self.retrieval_system = retrieval_system
         self.embedding_model = SentenceTransformer("NeuML/pubmedbert-base-embeddings")
         
@@ -66,11 +66,11 @@ class UserPromptProcessor:
         if predefined_result:
             return predefined_result
         
-        # Level 2: Meditron Extraction (if available)
-        if self.meditron_client:
-            meditron_result = self._extract_with_meditron(user_query)
-            if meditron_result:
-                return meditron_result
+        # Level 2: Llama3-Med42-70B Extraction (if available)
+        if self.llm_client:
+            llm_result = self._extract_with_llm(user_query)
+            if llm_result:
+                return llm_result
         
         # Level 3: Semantic Search Fallback
         semantic_result = self._semantic_search_fallback(user_query)
@@ -112,9 +112,9 @@ class UserPromptProcessor:
         
         return None
 
-    def _extract_with_meditron(self, user_query: str) -> Optional[Dict[str, str]]:
+    def _extract_with_llm(self, user_query: str) -> Optional[Dict[str, str]]:
         """
-        Use Meditron for advanced condition extraction
+        Use Llama3-Med42-70B for advanced condition extraction
         
         Args:
             user_query: User's medical query
@@ -122,17 +122,17 @@ class UserPromptProcessor:
         Returns:
             Dict with condition and keywords, or None
         """
-        if not self.meditron_client:
+        if not self.llm_client:
             return None
         
         try:
-            meditron_response = self.meditron_client.analyze_medical_query(
+            llama_response = self.llm_client.analyze_medical_query(
                 query=user_query,
                 max_tokens=100,
                 timeout=2.0
             )
             
-            extracted_condition = meditron_response.get('extracted_condition', '')
+            extracted_condition = llama_response.get('extracted_condition', '')
             
             if extracted_condition and validate_condition(extracted_condition):
                 condition_details = get_condition_keywords(extracted_condition)
@@ -145,12 +145,12 @@ class UserPromptProcessor:
             return None
         
         except Exception as e:
-            logger.error(f"Meditron condition extraction error: {e}")
+            logger.error(f"Llama3-Med42-70B condition extraction error: {e}")
             return None
 
     def _semantic_search_fallback(self, user_query: str) -> Optional[Dict[str, str]]:
         """
-        Perform semantic search for condition extraction
+        Perform semantic search for condition extraction using sliding window chunks
         
         Args:
             user_query: User's medical query
@@ -158,31 +158,45 @@ class UserPromptProcessor:
         Returns:
             Dict with condition and keywords, or None
         """
+        logger.info(f"Starting semantic search fallback for query: '{user_query}'")
+        
         if not self.retrieval_system:
+            logger.warning("No retrieval system available for semantic search")
             return None
         
         try:
             # Perform semantic search on sliding window chunks
             semantic_results = self.retrieval_system.search_sliding_window_chunks(user_query)
             
+            logger.info(f"Semantic search returned {len(semantic_results)} results")
+            
             if semantic_results:
                 # Extract condition from top semantic result
                 top_result = semantic_results[0]
                 condition = self._infer_condition_from_text(top_result['text'])
                 
+                logger.info(f"Inferred condition: {condition}")
+                
                 if condition and validate_condition(condition):
                     condition_details = get_condition_keywords(condition)
-                    return {
+                    result = {
                         'condition': condition,
                         'emergency_keywords': condition_details.get('emergency', ''),
                         'treatment_keywords': condition_details.get('treatment', ''),
                         'semantic_confidence': top_result.get('distance', 0)
                     }
+                    
+                    logger.info(f"Semantic search successful. Condition: {condition}, "
+                                f"Confidence: {result['semantic_confidence']}")
+                    return result
+                else:
+                    logger.warning(f"Condition validation failed for: {condition}")
             
+            logger.info("No suitable condition found in semantic search")
             return None
         
         except Exception as e:
-            logger.error(f"Semantic search fallback error: {e}")
+            logger.error(f"Semantic search fallback error: {e}", exc_info=True)
             return None
 
     def _generic_medical_search(self, user_query: str) -> Optional[Dict[str, str]]:
@@ -369,97 +383,162 @@ Please confirm:
             'extracted_info': extracted_info
         }
 
-def validate_medical_query(self, user_query: str) -> Dict[str, Any]:
-    """
-    Validate if the query is a medical-related query using multi-layer verification
-    
-    Args:
-        user_query: User's input query
-    
-    Returns:
-        Dict with validation result or None if medical query
-    """
-    # Expanded medical keywords covering comprehensive medical terminology
-    predefined_medical_keywords = {
-        # Symptoms and signs
-        'pain', 'symptom', 'ache', 'fever', 'inflammation', 
-        'bleeding', 'swelling', 'rash', 'bruise', 'wound',
-        
-        # Medical professional terms
-        'disease', 'condition', 'syndrome', 'disorder', 
-        'medical', 'health', 'diagnosis', 'treatment', 
-        'therapy', 'medication', 'prescription',
-        
-        # Body systems and organs
-        'heart', 'lung', 'brain', 'kidney', 'liver', 
-        'blood', 'nerve', 'muscle', 'bone', 'joint',
-        
-        # Medical actions
-        'examine', 'check', 'test', 'scan', 'surgery', 
-        'operation', 'emergency', 'urgent', 'critical',
-        
-        # Specific medical fields
-        'cardiology', 'neurology', 'oncology', 'pediatrics', 
-        'psychiatry', 'dermatology', 'orthopedics'
-    }
-    
-    # Check if query contains predefined medical keywords
-    query_lower = user_query.lower()
-    if any(kw in query_lower for kw in predefined_medical_keywords):
-        return None  # Validated by predefined keywords
-    
-    # Step 2: Use Meditron for final determination
-    try:
-        # Ensure Meditron client is properly initialized
-        if not hasattr(self, 'meditron_client') or self.meditron_client is None:
-            self.logger.warning("Meditron client not initialized")
-            return self._generate_invalid_query_response()
-        
-        meditron_result = self.meditron_client.analyze_medical_query(
-            query=user_query,
-            max_tokens=100  # Limit tokens for efficiency
-        )
-        
-        # If Meditron successfully extracts a medical condition
-        if meditron_result.get('extracted_condition'):
-            return None  # Validated by Meditron
-        
-    except Exception as e:
-        # Log Meditron analysis failure without blocking the process
-        self.logger.warning(f"Meditron query validation failed: {e}")
-    
-    # If no medical relevance is found
-    return self._generate_invalid_query_response()
+    def _handle_matching_failure_level1(self, condition: str) -> Optional[Dict[str, Any]]:
+        """
+        Level 1 Fallback: Loose keyword matching for medical conditions
+        
+        Args:
+            condition: The condition to match loosely
+        
+        Returns:
+            Dict with matched keywords or None
+        """
+        # Predefined loose matching keywords for different medical domains
+        loose_medical_keywords = {
+            'emergency': [
+                'urgent', 'critical', 'severe', 'acute', 
+                'immediate', 'life-threatening', 'emergency'
+            ],
+            'treatment': [
+                'manage', 'cure', 'heal', 'recover', 
+                'therapy', 'medication', 'intervention'
+            ]
+        }
+        
+        # Normalize condition
+        condition_lower = condition.lower().strip()
+        
+        # Check emergency keywords
+        emergency_matches = [
+            kw for kw in loose_medical_keywords['emergency'] 
+            if kw in condition_lower
+        ]
+        
+        # Check treatment keywords
+        treatment_matches = [
+            kw for kw in loose_medical_keywords['treatment'] 
+            if kw in condition_lower
+        ]
+        
+        # If matches found, return result
+        if emergency_matches or treatment_matches:
+            logger.info(f"Loose keyword match for condition: {condition}")
+            return {
+                'type': 'loose_keyword_match',
+                'condition': condition,
+                'emergency_keywords': '|'.join(emergency_matches),
+                'treatment_keywords': '|'.join(treatment_matches),
+                'confidence': 0.5  # Lower confidence due to loose matching
+            }
+        
+        # No loose matches found
+        logger.info(f"No loose keyword match for condition: {condition}")
+        return None
 
-def _generate_invalid_query_response(self) -> Dict[str, Any]:
-    """
-    Generate response for non-medical queries
-    
-    Returns:
-        Dict with invalid query guidance
-    """
-    return {
-        'type': 'invalid_query',
-        'message': "This is OnCall.AI, a clinical medical assistance platform. "
-                   "Please input a medical problem you need help resolving. "
-                   "\n\nExamples:\n"
-                   "- 'I'm experiencing chest pain'\n"
-                   "- 'What are symptoms of stroke?'\n"
-                   "- 'How to manage acute asthma?'\n"
-                   "- 'I have a persistent headache'"
-    }
+    def validate_medical_query(self, user_query: str) -> Dict[str, Any]:
+        """
+        Validate if the query is a medical-related query using Llama3-Med42-70B multi-layer verification
+        
+        Args:
+            user_query: User's input query
+        
+        Returns:
+            Dict with validation result or None if medical query
+        """
+        # Expanded medical keywords covering comprehensive medical terminology
+        predefined_medical_keywords = {
+            # Symptoms and signs
+            'pain', 'symptom', 'ache', 'fever', 'inflammation', 
+            'bleeding', 'swelling', 'rash', 'bruise', 'wound',
+            
+            # Medical professional terms
+            'disease', 'condition', 'syndrome', 'disorder', 
+            'medical', 'health', 'diagnosis', 'treatment', 
+            'therapy', 'medication', 'prescription',
+            
+            # Body systems and organs
+            'heart', 'lung', 'brain', 'kidney', 'liver', 
+            'blood', 'nerve', 'muscle', 'bone', 'joint',
+            
+            # Medical actions
+            'examine', 'check', 'test', 'scan', 'surgery', 
+            'operation', 'emergency', 'urgent', 'critical',
+            
+            # Specific medical fields
+            'cardiology', 'neurology', 'oncology', 'pediatrics', 
+            'psychiatry', 'dermatology', 'orthopedics'
+        }
+        
+        # Check if query contains predefined medical keywords
+        query_lower = user_query.lower()
+        if any(kw in query_lower for kw in predefined_medical_keywords):
+            return None  # Validated by predefined keywords
+        
+        try:
+            # Ensure Llama3-Med42-70B client is properly initialized
+            if not hasattr(self, 'llm_client') or self.llm_client is None:
+                self.logger.warning("Llama3-Med42-70B client not initialized")
+                return self._generate_invalid_query_response()
+            
+            # Use Llama3-Med42-70B for final medical query determination
+            llama_result = self.llm_client.analyze_medical_query(
+                query=user_query,
+                max_tokens=100  # Limit tokens for efficiency
+            )
+            
+            # If Llama3-Med42-70B successfully extracts a medical condition
+            if llama_result.get('extracted_condition'):
+                return None  # Validated by Llama3-Med42-70B
+            
+        except Exception as e:
+            # Log Llama3-Med42-70B analysis failure without blocking the process
+            self.logger.warning(f"Llama3-Med42-70B query validation failed: {e}")
+        
+        # If no medical relevance is found
+        return self._generate_invalid_query_response()
+
+    def _generate_invalid_query_response(self) -> Dict[str, Any]:
+        """
+        Generate response for non-medical queries
+        
+        Returns:
+            Dict with invalid query guidance
+        """
+        return {
+            'type': 'invalid_query',
+            'message': "This is OnCall.AI, a clinical medical assistance platform. "
+                       "Please input a medical problem you need help resolving. "
+                       "\n\nExamples:\n"
+                       "- 'I'm experiencing chest pain'\n"
+                       "- 'What are symptoms of stroke?'\n"
+                       "- 'How to manage acute asthma?'\n"
+                       "- 'I have a persistent headache'"
+        }
 
 def main():
     """
-    Example usage and testing of UserPromptProcessor
+    Example usage and testing of UserPromptProcessor with Llama3-Med42-70B
+    Demonstrates condition extraction and query validation
     """
-    processor = UserPromptProcessor()
+    from .retrieval import BasicRetrievalSystem
+
+    # use relative import to avoid circular import
+    from .llm_clients import llm_Med42_70BClient
+    
+    # Initialize LLM client
+    llm_client = llm_Med42_70BClient()
+    retrieval_system = BasicRetrievalSystem()
+    
+    # Initialize UserPromptProcessor with the LLM client
+    processor = UserPromptProcessor(
+        llm_client=llm_client, retrieval_system=retrieval_system
+    )
     
-    # Test cases
+    # Update test cases with more representative medical queries
     test_queries = [
-        "how to treat acute MI?",
-        "patient with stroke symptoms",
-        "chest pain and breathing difficulty"
+        "patient with severe chest pain and shortness of breath",
+        "sudden neurological symptoms suggesting stroke",
+        "persistent headache with vision changes"
     ]
     
     for query in test_queries:

tests/test_user_prompt.py

@@ -0,0 +1,92 @@
+"""
+User Prompt Processor Test Suite
+
+Comprehensive unit tests for UserPromptProcessor class
+Ensures robust functionality across medical query scenarios.
+"""
+
+import pytest
+import sys
+from pathlib import Path
+
+# Dynamically add project root to Python path
+project_root = Path(__file__).parent.parent
+sys.path.insert(0, str(project_root / "src"))
+
+from user_prompt import UserPromptProcessor
+
+class TestUserPromptProcessor:
+    """Test suite for UserPromptProcessor functionality"""
+
+    def setup_method(self):
+        """Initialize test environment before each test method"""
+        self.processor = UserPromptProcessor()
+
+    def test_extract_condition_keywords_predefined(self):
+        """Test predefined condition extraction"""
+        query = "heart attack symptoms"
+        result = self.processor.extract_condition_keywords(query)
+        
+        assert result is not None
+        assert 'condition' in result
+        assert 'emergency_keywords' in result
+        assert 'treatment_keywords' in result
+
+    def test_handle_matching_failure_level1(self):
+        """Test loose keyword matching mechanism"""
+        test_queries = [
+            "urgent medical help",
+            "critical condition",
+            "severe symptoms"
+        ]
+        
+        for query in test_queries:
+            result = self.processor._handle_matching_failure_level1(query)
+            
+            assert result is not None
+            assert result['type'] == 'loose_keyword_match'
+            assert result['confidence'] == 0.5
+
+    def test_semantic_search_fallback(self):
+        """Verify semantic search fallback mechanism"""
+        test_queries = [
+            "how to manage chest pain",
+            "treatment for acute stroke",
+            "emergency cardiac care"
+        ]
+        
+        for query in test_queries:
+            result = self.processor._semantic_search_fallback(query)
+            
+            # Result can be None if no match found
+            if result is not None:
+                assert 'condition' in result
+                assert 'emergency_keywords' in result
+                assert 'treatment_keywords' in result
+
+    def test_validate_keywords(self):
+        """Test keyword validation functionality"""
+        valid_keywords = {
+            'emergency_keywords': 'urgent|critical',
+            'treatment_keywords': 'medication|therapy'
+        }
+        
+        invalid_keywords = {
+            'emergency_keywords': '',
+            'treatment_keywords': ''
+        }
+        
+        assert self.processor.validate_keywords(valid_keywords) is True
+        assert self.processor.validate_keywords(invalid_keywords) is False
+
+def main():
+    """Run comprehensive test suite with detailed reporting"""
+    print("\n" + "="*60)
+    print("OnCall.ai: User Prompt Processor Test Suite")
+    print("="*60)
+    
+    # Run pytest with verbose output
+    pytest.main([__file__, '-v', '--tb=short'])
+
+if __name__ == "__main__":
+    main()