| """ |
| Maria NHS GP Clinical Reasoning & Synthesis Engine |
| Integrates NICE guideline retrieval with Maria-1B for structured clinical consultation guidance. |
| Includes direct PyTorch GPU inference from Maria-1B to demonstrate LLM output generation. |
| """ |
|
|
| import os |
| import sys |
| import re |
| import torch |
| from typing import Dict, Any, List |
|
|
| |
| current_dir = os.path.dirname(os.path.abspath(__file__)) |
| if current_dir not in sys.path: |
| sys.path.insert(0, current_dir) |
|
|
| from retriever import ClinicalRetriever |
|
|
| class MariaNICEAgent: |
| def __init__(self, model_id_or_path: str = "Zahid2005/Maria-1B-Instruct", use_llm_synthesis: bool = True): |
| self.retriever = ClinicalRetriever() |
| self.use_llm_synthesis = use_llm_synthesis |
| self.model_id_or_path = model_id_or_path |
| self.model = None |
| self.tokenizer = None |
| self.device = "cuda:0" if torch.cuda.is_available() else "cpu" |
| self.model_loaded = False |
| |
| if self.use_llm_synthesis: |
| self._load_maria_model() |
|
|
| def _load_maria_model(self): |
| """Loads Maria-1B-Instruct model on GPU/CPU for real clinical generation.""" |
| try: |
| from transformers import AutoModelForCausalLM, AutoTokenizer |
| local_fallback = r"C:\rep1\huggingface_maria_instruct_deploy" |
| target_path = local_fallback if os.path.exists(local_fallback) else self.model_id_or_path |
| print(f"[*] Loading Maria-1B from '{target_path}' onto {self.device}...") |
| self.tokenizer = AutoTokenizer.from_pretrained(target_path) |
| self.model = AutoModelForCausalLM.from_pretrained( |
| target_path, |
| trust_remote_code=True, |
| torch_dtype=torch.float16 if self.device.startswith("cuda") else torch.float32, |
| device_map="auto" if self.device.startswith("cuda") else None |
| ) |
| self.model.eval() |
| self.model_loaded = True |
| print(f"[OK] Maria-1B successfully loaded on {self.device}!") |
| except Exception as e: |
| print(f"[!] Warning: Could not load Maria-1B GPU model ({e}). Falling back to deterministic mode.") |
| self.model_loaded = False |
|
|
| def generate_maria_synthesis(self, query: str, guideline: Dict[str, Any]) -> str: |
| """ |
| Runs real inference with Maria-1B using RAG-prompted NICE guidance context. |
| """ |
| if not self.model_loaded: |
| return "Maria-1B model is not active on GPU." |
|
|
| rag_snippet = f"NICE {guideline['guideline_code']}: {guideline['title']}. Management: {guideline['management_pathway'][:300]}" |
| prompt = ( |
| f"<|begin_of_text|><|start_header_id|>system<|end_header_id|>\n" |
| f"You are Maria, an NHS General Practitioner clinical assistant. Use this guidance:\n{rag_snippet}<|eot_id|>" |
| f"<|start_header_id|>user<|end_header_id|>\n" |
| f"{query}<|eot_id|>" |
| f"<|start_header_id|>assistant<|end_header_id|>\n" |
| ) |
| try: |
| inputs = self.tokenizer(prompt, return_tensors="pt").to(self.device) |
| with torch.no_grad(): |
| out = self.model.generate( |
| **inputs, |
| max_new_tokens=45, |
| do_sample=True, |
| temperature=0.7, |
| top_p=0.9, |
| repetition_penalty=1.2, |
| pad_token_id=self.tokenizer.eos_token_id |
| ) |
| generated_text = self.tokenizer.decode(out[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True).strip() |
| return generated_text if generated_text else "Generated token stream completed." |
| except Exception as e: |
| return f"Inference exception: {e}" |
|
|
| def analyze_consultation(self, query: str, patient_age: int = None, gender: str = None) -> Dict[str, Any]: |
| top_guidelines = self.retriever.retrieve(query, top_k=2) |
| primary_guideline = top_guidelines[0] if top_guidelines else None |
|
|
| if not primary_guideline: |
| return { |
| "status": "not_found", |
| "message": "No specific NICE clinical guideline matched. Please refine your clinical query." |
| } |
|
|
| |
| detected_red_flags = [] |
| q_lower = query.lower() |
| red_flag_triggers = [ |
| "weight loss", "dysphagia", "swallowing", "haematuria", "blood in urine", |
| "haemoptysis", "coughing up blood", "postmenopausal", "vaginal bleeding", |
| "rectal bleeding", "looser stools", "iron deficiency", "anaemia", |
| "jaundice", "yellow sclera", "lump", "mass", "breast", "prostate", |
| "psa", "mole", "melanoma", "2ww", "2-week wait", "crushing chest pain", |
| "chest pain", "papilloedema", "retinal haemorrhage", "accelerated hypertension", |
| "silent chest", "cyanosis", "exhaustion", "short phrases", "severe asthma", |
| "hypoglycaem", "glucose 3.", "rapid breathing", "deep rapid", "ketoacidosis", |
| "dka", "pyelonephritis", "rigors", "flank tenderness", "haematemesis", |
| "melaena", "vomiting blood", "tarry stools", "confusion", "suicide" |
| ] |
| |
| has_trigger = any(trigger in q_lower for trigger in red_flag_triggers) |
| for rf in primary_guideline.get("red_flags", []): |
| rf_words = set(re.findall(r'\b\w{4,}\b', rf.lower())) |
| q_words = set(re.findall(r'\b\w{4,}\b', q_lower)) |
| overlap = rf_words.intersection(q_words) |
| if has_trigger or len(overlap) >= 2: |
| detected_red_flags.append(rf) |
|
|
| |
| maria_raw_generation = "" |
| if self.model_loaded: |
| maria_raw_generation = self.generate_maria_synthesis(query, primary_guideline) |
|
|
| |
| clinical_report = { |
| "query": query, |
| "patient_context": { |
| "age": patient_age or "Not specified", |
| "gender": gender or "Not specified" |
| }, |
| "primary_guideline": { |
| "code": primary_guideline["guideline_code"], |
| "title": primary_guideline["title"], |
| "category": primary_guideline["category"] |
| }, |
| "maria_llm_output": maria_raw_generation, |
| "red_flags": primary_guideline.get("red_flags", []), |
| "detected_urgencies": detected_red_flags, |
| "diagnostic_criteria": primary_guideline.get("diagnostic_criteria", ""), |
| "management_pathway": primary_guideline.get("management_pathway", ""), |
| "monitoring_and_safetynetting": primary_guideline.get("monitoring", ""), |
| "reference_url": f"https://www.nice.org.uk/guidance/{primary_guideline['guideline_code'].lower().replace(' ', '')}" |
| } |
| return clinical_report |
|
|
