import json import os import re from html import escape from typing import Dict, List, Literal, Optional, Tuple import gradio as gr import pandas as pd # --------------------------------------------------------------------- # SentinelPlus — In-Class Presentation Build # Workflow: 1) Patient Check-in 2) Severity Score 3) ADR Triage 4) Doctor's View # --------------------------------------------------------------------- # Models used: # Severity — local DeBERTa v2 fine-tuned classifier (SAFE / INJECTION) # Extraction — Clinical-AI-Apollo/Medical-NER (HuggingFace Hub) # --------------------------------------------------------------------- DEMO_CASES: Dict[str, Dict] = { "DEMO-001 - Maria Thompson (Osteopenia / Alendronate)": { "patient_id": "DEMO-001", "patient_name": "Maria Thompson", "suspected_medication": "Alendronate", "expected_severity_label": "Severe", "expected_triage_summary": "Severe muscle pain, weakness, chills, difficulty walking, chest/rib tightness, emergency department evaluation after taking alendronate.", "profile": { "age": "67", "sex": "Female", "dob": "Not on file", "height": "5 ft 4 in", "weight": "142 lb", "race": "Not on file", "condition": "Osteopenia", "allergies": "Sulfa antibiotics", "relevant_history": "Hypertension, osteopenia, mild chronic kidney disease", "contact_status": "Pre-filled from SentinelPlus profile", }, "medications": { "rx": [ {"Medication": "Alendronate", "Dose": "70 mg weekly", "Start Date": "2026-04-28", "Status": "New / suspected"}, {"Medication": "Lisinopril", "Dose": "10 mg daily", "Start Date": "Ongoing", "Status": "Ongoing"}, ], "otc": [ {"Product": "Calcium carbonate", "Dose": "As directed", "Start Date": "Ongoing"}, {"Product": "Vitamin D3", "Dose": "As directed", "Start Date": "Ongoing"}, ], }, "biometrics": { "summary": "Elevated heart rate from pain and stress. Sleep disrupted by chills and deep muscle pain. Mobility sharply reduced because walking required assistance.", "flags": ["Difficulty walking", "Chills and weakness", "Chest/rib tightness"], "metrics": [ {"type": "heart_rate", "value": "108 bpm", "subtitle": "Elevated during pain episode", "flag": "Chest/rib tightness"}, {"type": "sleep", "value": "Poor", "subtitle": "Disrupted by chills and pain", "flag": "Chills and weakness"}, {"type": "mobility", "value": "Very Low", "subtitle": "Needed help walking to bathroom", "flag": "Difficulty walking"}, {"type": "weight", "value": "142 lb", "subtitle": "Patient weight on file", "flag": ""}, ], }, "checkins": [ {"Date": "2026-04-28", "Check-in": "Took weekly alendronate dose in the morning."}, {"Date": "2026-04-29", "Check-in": "Severe muscle pain, chills, weakness, and trouble walking; evaluated in emergency department."}, ], "testimony": ( "I took my weekly bone medication in the morning like usual. Later that night I started having deep muscle pain " "in my legs and back. By the next day the pain was much worse and I had chills, weakness, and trouble walking " "to the bathroom without help. My chest and ribs felt tight when I tried to take a deep breath. My daughter took " "me to urgent care and they sent me to the emergency department because I could barely stand." ), }, "DEMO-002 - James Carter (Skin Infection / TMP-SMX)": { "patient_id": "DEMO-002", "patient_name": "James Carter", "suspected_medication": "Trimethoprim-sulfamethoxazole", "expected_severity_label": "Severe", "expected_triage_summary": "Rapidly spreading rash, lip/facial swelling, lightheadedness, breathing concern, emergency response after starting trimethoprim-sulfamethoxazole.", "profile": { "age": "54", "sex": "Male", "dob": "Not on file", "height": "5 ft 11 in", "weight": "204 lb", "race": "Not on file", "condition": "Skin infection", "allergies": "None reported", "relevant_history": "Type 2 diabetes, high cholesterol", "contact_status": "Pre-filled from SentinelPlus profile", }, "medications": { "rx": [ {"Medication": "Trimethoprim-sulfamethoxazole", "Dose": "800/160 mg twice daily", "Start Date": "2026-05-01", "Status": "New / suspected"}, {"Medication": "Metformin", "Dose": "500 mg twice daily", "Start Date": "Ongoing", "Status": "Ongoing"}, {"Medication": "Atorvastatin", "Dose": "40 mg daily", "Start Date": "Ongoing", "Status": "Ongoing"}, ], "otc": [{"Product": "Ibuprofen", "Dose": "400 mg as needed", "Start Date": "Ongoing"}], }, "biometrics": { "summary": "Heart rate and breathing rate increased during rash and swelling episode. Lightheadedness reported when standing. Emergency services were contacted.", "flags": ["Rapidly spreading rash", "Lip/facial swelling", "Emergency response"], "metrics": [ {"type": "heart_rate", "value": "116 bpm", "subtitle": "Elevated during acute reaction", "flag": "Emergency response"}, {"type": "sleep", "value": "Disrupted", "subtitle": "Symptoms worsened overnight", "flag": "Rapidly spreading rash"}, {"type": "mobility", "value": "Reduced", "subtitle": "Lightheaded when standing", "flag": "Lip/facial swelling"}, {"type": "weight", "value": "204 lb", "subtitle": "Patient weight on file", "flag": ""}, ], }, "checkins": [ {"Date": "2026-05-03", "Check-in": "Rash appeared on chest and arms after two days of antibiotic use."}, {"Date": "2026-05-04", "Check-in": "Rash spread with lip/facial swelling, lightheadedness, and faster breathing; treated in emergency room."}, ], "testimony": ( "I was prescribed an antibiotic for a skin infection. After two days, I noticed a rash on my chest and arms. " "By the next morning the rash had spread, my lips felt swollen, and I felt lightheaded when I stood up. My wife " "said my face looked puffy and I was breathing faster than normal. We called 911 and I was treated in the emergency room." ), }, "DEMO-003 - Priya Shah (Anxiety / Sertraline)": { "patient_id": "DEMO-003", "patient_name": "Priya Shah", "suspected_medication": "Sertraline", "expected_severity_label": "Non-severe", "expected_triage_summary": "Mild nausea and dry mouth after starting sertraline, no emergency symptoms, patient remained functional and contacted doctor for guidance.", "profile": { "age": "31", "sex": "Female", "dob": "Not on file", "height": "5 ft 6 in", "weight": "129 lb", "race": "Not on file", "condition": "Anxiety symptoms", "allergies": "None reported", "relevant_history": "Seasonal allergies, migraines", "contact_status": "Pre-filled from SentinelPlus profile", }, "medications": { "rx": [ {"Medication": "Sertraline", "Dose": "25 mg daily", "Start Date": "2026-05-03", "Status": "New / suspected"}, {"Medication": "Sumatriptan", "Dose": "As needed", "Start Date": "Ongoing", "Status": "Ongoing"}, ], "otc": [{"Product": "Cetirizine", "Dose": "As needed", "Start Date": "Ongoing"}], }, "biometrics": { "summary": "No significant biometric changes. Patient remained functional, went to work, and tolerated small meals.", "flags": ["Mild nausea", "Dry mouth"], "metrics": [ {"type": "heart_rate", "value": "72 bpm", "subtitle": "Within usual range", "flag": ""}, {"type": "sleep", "value": "Normal", "subtitle": "No sleep disruption reported", "flag": ""}, {"type": "mobility", "value": "Normal", "subtitle": "Went to work and remained functional", "flag": ""}, {"type": "weight", "value": "129 lb", "subtitle": "Patient weight on file", "flag": "Mild nausea"}, ], }, "checkins": [ {"Date": "2026-05-05", "Check-in": "Mild morning nausea and dry mouth after starting sertraline."}, {"Date": "2026-05-06", "Check-in": "No swelling, breathing issues, chest pain, rash, or loss of function; contacted doctor's office."}, ], "testimony": ( "I started sertraline a few days ago. Since then I have had mild nausea in the morning and a dry mouth. I was " "still able to go to work and eat small meals. I did not have trouble breathing, swelling, chest pain, or a rash. " "I called my doctor's office to ask whether these side effects are expected." ), }, "DEMO-004 - Robert Nguyen (Seasonal Allergies / Loratadine)": { "patient_id": "DEMO-004", "patient_name": "Robert Nguyen", "suspected_medication": "Loratadine", "expected_severity_label": "Non-severe", "expected_triage_summary": "Fatigue and mild headache after loratadine, symptoms improved with rest, no signs of serious allergic reaction or emergency-level ADR.", "profile": { "age": "46", "sex": "Male", "dob": "Not on file", "height": "5 ft 8 in", "weight": "176 lb", "race": "Not on file", "condition": "Seasonal allergies", "allergies": "Penicillin causes mild rash", "relevant_history": "Acid reflux, seasonal allergies", "contact_status": "Pre-filled from SentinelPlus profile", }, "medications": { "rx": [{"Medication": "Omeprazole", "Dose": "20 mg daily", "Start Date": "Ongoing", "Status": "Ongoing"}], "otc": [ {"Product": "Loratadine", "Dose": "10 mg daily as needed", "Start Date": "2026-05-06"}, {"Product": "Acetaminophen", "Dose": "500 mg occasional", "Start Date": "Ongoing"}, ], }, "biometrics": { "summary": "Mild fatigue and headache with normal heart rate, normal breathing, and preserved mobility. Symptoms improved after hydration and rest.", "flags": ["Fatigue", "Mild headache"], "metrics": [ {"type": "heart_rate", "value": "68 bpm", "subtitle": "Within usual range", "flag": ""}, {"type": "sleep", "value": "Normal", "subtitle": "No sleep disruption reported", "flag": ""}, {"type": "mobility", "value": "Normal", "subtitle": "Symptoms improved with rest", "flag": "Fatigue"}, {"type": "weight", "value": "176 lb", "subtitle": "Patient weight on file", "flag": "Mild headache"}, ], }, "checkins": [ {"Date": "2026-05-06", "Check-in": "Felt more tired than usual after loratadine and had a slight headache."}, {"Date": "2026-05-07", "Check-in": "Headache improved with water and rest; no hives, swelling, breathing problems, fever, or severe pain."}, ], "testimony": ( "I took loratadine for allergies and noticed I felt more tired than usual that afternoon. I also had a slight " "headache, but it improved after drinking water and resting. I did not have hives, swelling, breathing problems, " "fever, or severe pain. I skipped the next dose and planned to ask the pharmacist if I should try a different allergy medicine." ), }, } EHR_EXTRAS: Dict[str, Dict] = { "DEMO-001": { "profile": { "dob": "1959-02-14", "race": "White", "ethnicity": "Not Hispanic or Latino", "address": "1846 Maple Ridge Lane", "city_state": "Columbus, OH", "zip": "43214", "country": "United States", "phone": "614-555-0184", "email": "maria.thompson@example.com", "initials": "M. T.", "signature": "Maria Thompson", }, "product_details": { "product_available": "Yes", "product_picture_available": "Yes", "manufacturer": "Generic alendronate manufacturer on pharmacy label", "product_type": "Drug or Biologic - Generic", "expiration_date": "2027-11-30", "lot_number": "ALN042826A", "ndc_number": "60505-2578-1", "quantity": "1 tablet", "frequency": "Once weekly", "route": "By mouth", "duration": "1 dose before event", "therapy_stop_date": "2026-04-29", "therapy_reduced_date": "Not applicable", "problem_stopped_after_reduced_or_stopped": "Not yet known", "problem_returned_after_restart": "Did not restart", }, }, "DEMO-002": { "profile": { "dob": "1972-07-09", "race": "Black or African American", "ethnicity": "Not Hispanic or Latino", "address": "7720 Brookstone Court", "city_state": "Charlotte, NC", "zip": "28210", "country": "United States", "phone": "704-555-0197", "email": "james.carter@example.com", "initials": "J. C.", "signature": "James Carter", }, "product_details": { "product_available": "Yes", "product_picture_available": "No", "manufacturer": "Dispensed generic trimethoprim-sulfamethoxazole", "product_type": "Drug or Biologic - Generic", "expiration_date": "2027-08-31", "lot_number": "TS050126B", "ndc_number": "65862-420-05", "quantity": "1 tablet", "frequency": "Twice daily", "route": "By mouth", "duration": "2 days", "therapy_stop_date": "2026-05-04", "therapy_reduced_date": "Not applicable", "problem_stopped_after_reduced_or_stopped": "Improved after emergency treatment", "problem_returned_after_restart": "Did not restart", }, }, "DEMO-003": { "profile": { "dob": "1995-11-21", "race": "Asian", "ethnicity": "Not Hispanic or Latino", "address": "4218 Cedar Park Drive", "city_state": "Austin, TX", "zip": "78731", "country": "United States", "phone": "512-555-0131", "email": "priya.shah@example.com", "initials": "P. S.", "signature": "Priya Shah", }, "product_details": { "product_available": "Yes", "product_picture_available": "Yes", "manufacturer": "Generic sertraline manufacturer on pharmacy label", "product_type": "Drug or Biologic - Generic", "expiration_date": "2027-10-31", "lot_number": "SER050326C", "ndc_number": "31722-214-30", "quantity": "1 tablet", "frequency": "Once daily", "route": "By mouth", "duration": "3 days at time of report", "therapy_stop_date": "Ongoing at time of report", "therapy_reduced_date": "Not applicable", "problem_stopped_after_reduced_or_stopped": "Not applicable", "problem_returned_after_restart": "Did not restart", }, }, "DEMO-004": { "profile": { "dob": "1980-03-18", "race": "Asian", "ethnicity": "Not Hispanic or Latino", "address": "936 Lakeview Avenue", "city_state": "Portland, OR", "zip": "97213", "country": "United States", "phone": "503-555-0168", "email": "robert.nguyen@example.com", "initials": "R. N.", "signature": "Robert Nguyen", }, "product_details": { "product_available": "Yes", "product_picture_available": "Yes", "manufacturer": "OTC loratadine private-label package", "product_type": "Over-the-Counter (OTC)", "expiration_date": "2027-06-30", "lot_number": "LOR050626D", "ndc_number": "68196-281-10", "quantity": "1 tablet", "frequency": "Once daily as needed", "route": "By mouth", "duration": "1 dose before event", "therapy_stop_date": "2026-05-07", "therapy_reduced_date": "Not applicable", "problem_stopped_after_reduced_or_stopped": "Improved with rest and hydration", "problem_returned_after_restart": "Did not restart", }, }, } for _case in DEMO_CASES.values(): _extra = EHR_EXTRAS.get(_case.get("patient_id"), {}) _case["profile"].update(_extra.get("profile", {})) _case["product_details"] = _extra.get("product_details", {}) _case["reporter"] = { "last_name": "SentinelPlus", "first_name": "Automated System", "address": "1200 Health System Way", "city_state": "Cleveland, OH", "zip": "44114", "country": "United States", "phone": "216-555-0100", "email": "DrOzOffice@Stayhealthy.com", "today_date": "2026-05-13", "reported_to_manufacturer": False, "do_not_disclose_identity": False, } # Terms used only for the driver-extraction helper. SYMPTOM_TERMS = [ "muscle pain", "deep muscle pain", "leg pain", "back pain", "chills", "weakness", "trouble walking", "difficulty walking", "chest tightness", "rib tightness", "rash", "spreading rash", "swollen lips", "lip swelling", "facial swelling", "lightheaded", "breathing faster", "breathing concern", "nausea", "mild nausea", "dry mouth", "tired", "fatigue", "headache", "mild headache", "hives", "swelling", "breathing problems", "chest pain", "fever", "severe pain", ] MEDICATION_TERMS = [ "alendronate", "lisinopril", "trimethoprim-sulfamethoxazole", "trimethoprim sulfamethoxazole", "tmp-smx", "antibiotic", "metformin", "atorvastatin", "ibuprofen", "sertraline", "sumatriptan", "cetirizine", "loratadine", "omeprazole", "acetaminophen", "calcium carbonate", "vitamin d3", ] TIMING_PATTERNS = [ r"\b\d+\s*hours?\s+later\b", r"\b\d+\s*days?\s+ago\b", r"\byesterday\b", r"\btwo\s+hours\s+after\b", r"\bafter\s+first\s+dose\b", r"\bover\s+several\s+days\b", r"\bthe\s+next\s+day\b", r"\b\d+\s*weeks?\b", r"\bwithin\s+\d+\s*hours?\b", r"\ba few days ago\b", r"\bafter two days\b", r"\bby the next morning\b", r"\blater that night\b", ] # HTML injected when "Submit to MedWatch" is clicked. MODAL_HTML = """
✓

Form 3500B Submitted

Your MedWatch report has been submitted to the FDA.
A confirmation has been sent to the reporter on file.

""" # --------------------------------------------------------------------- # SCENARIO LOADING HELPERS # --------------------------------------------------------------------- def load_scenario(case_name: str) -> Tuple[str, str, str, str, List[Dict]]: case = DEMO_CASES[case_name] return ( case["testimony"], format_profile(case), format_medications(case), format_biometrics(case), [], ) def format_profile(case: Dict) -> str: p = case["profile"] def row(label: str, value: str) -> str: return ( f'
' f'
{label}
' f'
' f'{value}
' f'
' ) fields = [ ("Patient ID", case.get("patient_id", "Not on file")), ("Patient", case["patient_name"]), ("Age", p["age"]), ("Sex", p["sex"]), ("Date of Birth", p["dob"]), ("Height", p.get("height", "Not on file")), ("Weight", p["weight"]), ("Race / Ethnicity", p["race"]), ("Ethnicity Detail", p.get("ethnicity", "Not on file")), ("Primary Condition", p["condition"]), ("Suspected Product", case.get("suspected_medication", "Not on file")), ("Allergies", p["allergies"]), ("Relevant History", p["relevant_history"]), ("Home Address", p.get("address", "Not on file")), ("City / State", p.get("city_state", "Not on file")), ("ZIP", p.get("zip", "Not on file")), ("Phone", p.get("phone", "Not on file")), ("Email", p.get("email", "Not on file")), ("Initials", p.get("initials", "Not on file")), ("Contact Status", p["contact_status"]), ] rows_html = "".join(row(lbl, val) for lbl, val in fields) return ( '
' '
SentinelPlus Patient Profile
' '
' + rows_html + '
' ) def format_medications(case: Dict) -> str: rx = case["medications"]["rx"] otc = case["medications"]["otc"] rx_df = pd.DataFrame(rx) rx_part = "### Current RX Medications\n\n" + rx_df.to_markdown(index=False) if otc: otc_df = pd.DataFrame(otc) otc_part = "\n\n### Current OTC / Supplement Products\n\n" + otc_df.to_markdown(index=False) else: otc_part = "\n\n### Current OTC / Supplement Products\n\nNone listed." return rx_part + otc_part _METRIC_ICONS = { "heart_rate": "❤️", "sleep": "😴", "mobility": "🚶", "weight": "⚖️", } def format_biometrics(case: Dict) -> str: metrics = case["biometrics"].get("metrics", []) cards_html = "" for m in metrics: icon = _METRIC_ICONS.get(m["type"], "📊") if m["type"] == "heart_rate": animated = ' class="hb-icon"' elif m["type"] == "weight": animated = ' class="scale-icon"' else: animated = "" cards_html += f"""
{icon}
{m['value']}
""" checkins_html = "".join( f'
  • {row["Date"]}: {row["Check-in"]}
    • ' for row in case["checkins"] ) return f"""

    Wearable Biometric Summary

    {cards_html}

    Recent Check-ins

    """ # --------------------------------------------------------------------- # MODEL LOADING (LAZY) # --------------------------------------------------------------------- _severity_pipe = None _ner_pipe = None def _get_severity_pipe(): global _severity_pipe if _severity_pipe is None: from transformers import ( AutoTokenizer, AutoModelForSequenceClassification, pipeline as hf_pipeline, ) MODEL_NAME = "Alanouette/sentinelplus-adr-severity" tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME) model = AutoModelForSequenceClassification.from_pretrained(MODEL_NAME) _severity_pipe = hf_pipeline( "text-classification", model=model, tokenizer=tokenizer, top_k=None, truncation=True, max_length=512, ) return _severity_pipe def _get_ner_pipe(): global _ner_pipe if _ner_pipe is None: from transformers import pipeline as hf_pipeline _ner_pipe = hf_pipeline( "token-classification", model="Clinical-AI-Apollo/Medical-NER", aggregation_strategy="simple", ) return _ner_pipe # --------------------------------------------------------------------- # MODEL FUNCTIONS # --------------------------------------------------------------------- _DRUG_KEYS = {"drug", "medication", "chemical", "medicine", "substance", "pharma"} _DISEASE_KEYS = {"disease", "disorder", "condition", "diagnosis", "pathology"} _TIMING_KEYS = {"time", "date", "duration", "temporal", "frequency", "period"} LIVE_AGENT_MODEL = os.getenv("SENTINELPLUS_AGENT_MODEL", "gpt-4.1-mini") ENABLE_LIVE_AGENT = os.getenv("SENTINELPLUS_LIVE_AGENT", "1").lower() not in {"0", "false", "no"} TRIAGE_AGENT_SYSTEM_PROMPT = """ You are SentinelPlus ADR Triage Agent, a narrow adverse drug reaction documentation assistant. Allowed scope: - Summarize possible adverse drug reaction documentation from the provided case context. - Return priority, next step, explanation, missing information, and mock FDA Form 3500B fields. - Ask or suggest clarifying documentation questions about symptoms, timing, medications, product details, allergies, medical history, and reporter fields. - Recommend human clinician/pharmacist review, urgent provider review, emergency care for red-flag symptoms, or completion of missing report fields. Hard prohibitions: - Do not diagnose. - Do not state that a medication caused the event. - Do not tell the user to start, stop, restart, skip, increase, decrease, or replace any medication. - Do not prescribe, dose, recommend, compare, or select prescription medications. - Do not answer unrelated requests such as restaurants, travel, entertainment, shopping, coding, homework, finance, or legal advice. - Do not invent facts that are not in the case context. Use "Not on file" or add the field to missing_information. Tone: - Plain language. - Human-in-the-loop. - Documentation and triage support only. """.strip() TRIAGE_QA_SYSTEM_PROMPT = """ You are SentinelPlus ADR Triage Agent in live Q&A mode. Answer only questions about the user's adverse drug reaction documentation, symptom clarification, medication/report context, missing information, or what to discuss with a human clinician. You must refuse off-topic requests. You must not diagnose or give prescription advice. For medication changes, advise the user to contact their prescribing clinician or pharmacist. For severe or emergency symptoms, advise urgent medical care. """.strip() OFF_TOPIC_KEYWORDS = { "restaurant", "restaurants", "pizza", "burger", "hotel", "flight", "vacation", "movie", "music", "sports", "stock", "crypto", "homework", "essay", "code", "python", "javascript", "recipe", "dating", "shopping", "weather", } PRESCRIPTION_ADVICE_PATTERNS = [ r"\bshould\s+i\s+(stop|start|take|skip|restart|continue|increase|decrease|double|halve)\b", r"\bcan\s+i\s+(stop|start|take|skip|restart|continue|increase|decrease|double|halve)\b", r"\bwhat\s+(dose|dosage)\b", r"\bprescribe\b", r"\brecommend\s+(a|another|different)?\s*(medication|drug|dose|dosage)\b", ] ADR_ALLOWED_KEYWORDS = { "adr", "adverse", "reaction", "side effect", "symptom", "symptoms", "medication", "medicine", "drug", "dose", "timing", "rash", "hives", "nausea", "vomit", "diarrhea", "pain", "swelling", "breathing", "dizzy", "doctor", "clinician", "pharmacist", "report", "medwatch", "fda", "3500b", "allergy", "allergies", "hospital", "urgent", "severe", "mild", "moderate", "biometric", "heart rate", "sleep", "mobility", "missing", "clarify", "form", } def _openai_client(): if not ENABLE_LIVE_AGENT or not os.getenv("OPENAI_API_KEY"): return None try: from openai import OpenAI return OpenAI() except Exception: return None def _friendly_live_ai_error(exc: Exception) -> str: text = str(exc).lower() if "insufficient_quota" in text or "exceeded your current quota" in text or "error code: 429" in text: return ( "Live AI Q&A is connected, but the OpenAI API key has no available quota right now. " "The local SentinelPlus triage workflow is still available. To enable live responses, " "update billing/quota for the OpenAI project behind the Hugging Face `OPENAI_API_KEY` secret " "or replace that secret with a key that has active quota." ) if "authentication" in text or "api key" in text or "401" in text: return ( "Live AI Q&A could not authenticate with OpenAI. Check the Hugging Face `OPENAI_API_KEY` " "secret and make sure it is current." ) return ( "Live AI Q&A is temporarily unavailable. The local SentinelPlus triage workflow is still available, " "and all outputs continue to require human clinical review." ) def _list_text(items: List[Dict], name_key: str) -> str: return "\n".join( f"- {item.get(name_key, 'Not listed')}, {item.get('Dose', 'dose not listed')}, " f"started {item.get('Start Date', 'date not listed')} ({item.get('Status', 'status not listed')})" for item in items ) or "None listed" def fallback_form_3500b_fields(case: Dict, severity: Dict) -> Dict: profile = case["profile"] rx = case["medications"]["rx"] otc = case["medications"]["otc"] testimony = case.get("testimony", "") suspected = case.get("suspected_medication", "") suspected_l = suspected.lower() primary_rx = next( (m for m in rx if suspected_l and suspected_l in m.get("Medication", "").lower()), rx[0] if rx else {}, ) primary_otc = next( (m for m in otc if suspected_l and suspected_l in m.get("Product", "").lower()), {}, ) primary_product = primary_rx or primary_otc product = case.get("product_details", {}) reporter = case.get("reporter", {}) is_severe = severity["label"] == "Severe ADR" hospitalized = "hospital" in testimony.lower() and is_severe name_parts = case["patient_name"].strip().split() return { "problem_type_hurt_or_side_effect": True, "problem_type_product_use_error": False, "problem_type_product_quality": False, "problem_type_switching_maker": False, "outcome_hospitalization": hospitalized, "outcome_required_help_prevent_harm": not is_severe, "outcome_disability": False, "outcome_birth_defect": False, "outcome_life_threatening": is_severe, "outcome_death": False, "outcome_other_serious_event": False, "date_problem_occurred": case["checkins"][0]["Date"] if case.get("checkins") else "Not on file", "event_narrative": testimony[:4000], "relevant_tests": "Vitals, medication list, allergy list, and encounter notes available in EHR.", "product_available": product.get("product_available", "Yes"), "product_picture_available": product.get("product_picture_available", "No"), "product_name": suspected or primary_rx.get("Medication") or primary_otc.get("Product", "Not listed"), "place_and_date_of_purchase": product.get("place_and_date_of_purchase", "Patient pharmacy record on file"), "therapy_ongoing": primary_rx.get("Status", "").lower() in ("ongoing", "continuing"), "manufacturer": product.get("manufacturer", "Manufacturer from product label on file"), "product_type": product.get("product_type", "Drug or Biologic"), "expiration_date": product.get("expiration_date", "On file"), "lot_number": product.get("lot_number", "On file"), "ndc_number": product.get("ndc_number", "On file"), "strength": primary_product.get("Dose", "Not listed"), "quantity": product.get("quantity", "1 dose"), "frequency": product.get("frequency", primary_product.get("Dose", "As directed")), "route": product.get("route", "By mouth"), "therapy_start_date": primary_product.get("Start Date", "Not listed"), "therapy_stop_date": product.get("therapy_stop_date", "Ongoing at time of report"), "therapy_reduced_date": product.get("therapy_reduced_date", "Not applicable"), "duration": product.get("duration", "On file"), "reason_for_use": profile.get("condition", "Not on file"), "problem_stopped_after_reduced_or_stopped": product.get("problem_stopped_after_reduced_or_stopped", "On file"), "problem_returned_after_restart": product.get("problem_returned_after_restart", "Did not restart"), "patient_initials": profile.get("initials") or ". ".join(p[0].upper() for p in name_parts if p and p[0].isalpha()) + ".", "patient_sex": profile.get("sex", "Not on file"), "patient_age": profile.get("age", "Not on file"), "patient_dob": profile.get("dob", "Not on file"), "patient_weight": profile.get("weight", "Not on file"), "patient_race_ethnicity": profile.get("race", "Not on file"), "known_medical_conditions": profile.get("relevant_history", "Not on file"), "allergies": profile.get("allergies", "Not on file"), "other_important_information": case["biometrics"].get("summary", "Not on file"), "otc_medications": "\n".join(f"- {m['Product']}, {m['Dose']}" for m in otc) or "None listed", "current_prescriptions": _list_text(rx, "Medication"), "reporter_last_name": reporter.get("last_name", "SentinelPlus"), "reporter_first_name": reporter.get("first_name", "Automated System"), "reporter_address": reporter.get("address", profile.get("address", "On file")), "reporter_city_state": reporter.get("city_state", profile.get("city_state", "On file")), "reporter_zip": reporter.get("zip", profile.get("zip", "On file")), "reporter_country": reporter.get("country", profile.get("country", "United States")), "reporter_phone": reporter.get("phone", profile.get("phone", "On file")), "reporter_email": reporter.get("email", profile.get("email", "On file")), "reporter_today_date": reporter.get("today_date", "2026-05-13"), "reported_to_manufacturer": reporter.get("reported_to_manufacturer", False), "do_not_disclose_identity": reporter.get("do_not_disclose_identity", False), } def build_agent_context(severity: Dict, extraction: Dict, case: Dict) -> Dict: profile = case["profile"] return { "purpose": "ADR documentation triage and mock FDA Form 3500B preparation only", "severity_model_output": severity, "ner_evidence": extraction, "patient_profile": { "age": profile.get("age"), "sex": profile.get("sex"), "height": profile.get("height"), "weight": profile.get("weight"), "race_ethnicity": profile.get("race"), "condition": profile.get("condition"), "allergies": profile.get("allergies"), "relevant_history": profile.get("relevant_history"), }, "suspected_medication": case.get("suspected_medication"), "medications": case["medications"], "product_details": case.get("product_details", {}), "biometrics": case["biometrics"], "checkins": case.get("checkins", []), "reporter": case.get("reporter", {}), "form_3500b_prefill": fallback_form_3500b_fields(case, severity), "patient_testimony": case.get("testimony", ""), "allowed_form": "FDA Form 3500B consumer voluntary report fields from Sections A, B, C, E, and F", } def is_question_in_scope(question: str) -> Tuple[bool, str]: q = question.lower().strip() if not q: return False, "Ask a question about symptoms, medication context, missing report details, or the ADR situation." if any(word in q for word in OFF_TOPIC_KEYWORDS) and not any(word in q for word in ADR_ALLOWED_KEYWORDS): return False, "I can only help with ADR documentation, symptom clarification, medication-context questions, and MedWatch report preparation." if any(re.search(pattern, q) for pattern in PRESCRIPTION_ADVICE_PATTERNS): return False, ( "I cannot recommend starting, stopping, changing, or dosing a medication. " "For medication decisions, contact the prescribing clinician or pharmacist. " "If symptoms feel severe or life-threatening, seek urgent medical care." ) if not any(word in q for word in ADR_ALLOWED_KEYWORDS): return False, "I can only answer questions related to this ADR case and report preparation." return True, "" def heuristic_severity_from_text(text: str) -> Dict: t = text.lower() severe_cues = [ "911", "emergency", "emergency room", "emergency department", "urgent care", "could barely stand", "trouble walking", "difficulty walking", "swollen", "swelling", "breathing faster", "breathing problems", "chest", "lightheaded", "face looked puffy", "lips felt swollen", ] score = 0.82 if any(cue in t for cue in severe_cues) else 0.18 label = "Severe ADR" if score >= 0.5 else "No Severe ADR Detected" return { "label": label, "probability": score, "drivers": [], "note": "Fallback severity estimate from testimony cues.", } def _terms_found(text: str, terms: List[str]) -> List[str]: t = text.lower() found = [term for term in terms if term.lower() in t] return sorted(set(found), key=lambda item: (t.find(item.lower()), item)) def run_severity_model(text: str) -> Dict: """ Binary ADR severity classifier. SAFE (0) = no severe ADR, INJECTION (1) = severe ADR detected. """ try: pipe = _get_severity_pipe() raw = pipe(text) except Exception: return heuristic_severity_from_text(text) # top_k=None returns [[{...}, {...}]]; normalize to a flat list of score dicts result = raw[0] if isinstance(raw[0], list) else raw inj_entry = next( (s for s in result if "INJECTION" in s["label"].upper() or s["label"] in ("LABEL_1", "1")), None, ) inj_score = inj_entry["score"] if inj_entry else max(s["score"] for s in result) label = "Severe ADR" if inj_score >= 0.5 else "No Severe ADR Detected" return { "label": label, "probability": round(inj_score, 3), "drivers": [], # populated from NER output in analyze() "note": "Output from trained DeBERTa v2 ADR severity classifier (SAFE / INJECTION).", } def run_medical_extraction_model(text: str) -> Dict: """Extract medical entities using Clinical-AI-Apollo/Medical-NER.""" try: pipe = _get_ner_pipe() entities = pipe(text) except Exception: entities = [] medications, symptoms, timing, patient_context = [], [], [], [] for ent in entities: grp = ent["entity_group"].lower() word = ent["word"].strip().strip("##") if not word: continue if any(k in grp for k in _DRUG_KEYS): medications.append(word) elif any(k in grp for k in _DISEASE_KEYS): patient_context.append(word) elif any(k in grp for k in _TIMING_KEYS): timing.append(word) else: symptoms.append(word) t = text.lower() for pattern in TIMING_PATTERNS: timing.extend(re.findall(pattern, t, flags=re.IGNORECASE)) medications.extend(_terms_found(text, MEDICATION_TERMS)) symptoms.extend(_terms_found(text, SYMPTOM_TERMS)) return { "medications": sorted(set(medications)) or ["Not detected"], "symptoms": sorted(set(symptoms)) or ["Not detected"], "timing": sorted(set(timing)) or ["Not detected"], "patient_context": sorted(set(patient_context)) or ["Not detected"], } # --------------------------------------------------------------------- # AGENTIC TRIAGE LAYER # --------------------------------------------------------------------- def run_live_triage_agent(severity: Dict, extraction: Dict, case: Dict) -> Optional[Dict]: client = _openai_client() if client is None: return None try: from pydantic import BaseModel, Field class Form3500BFields(BaseModel): problem_type_hurt_or_side_effect: bool problem_type_product_use_error: bool problem_type_product_quality: bool problem_type_switching_maker: bool outcome_hospitalization: bool outcome_required_help_prevent_harm: bool outcome_disability: bool outcome_birth_defect: bool outcome_life_threatening: bool outcome_death: bool outcome_other_serious_event: bool date_problem_occurred: str event_narrative: str = Field(max_length=4000) relevant_tests: str product_available: str product_picture_available: str product_name: str place_and_date_of_purchase: str therapy_ongoing: bool manufacturer: str product_type: str expiration_date: str lot_number: str ndc_number: str strength: str quantity: str frequency: str route: str therapy_start_date: str therapy_stop_date: str therapy_reduced_date: str duration: str reason_for_use: str problem_stopped_after_reduced_or_stopped: str problem_returned_after_restart: str patient_initials: str patient_sex: str patient_age: str patient_dob: str patient_weight: str patient_race_ethnicity: str known_medical_conditions: str allergies: str other_important_information: str otc_medications: str current_prescriptions: str reporter_last_name: str reporter_first_name: str reporter_address: str reporter_city_state: str reporter_zip: str reporter_country: str reporter_phone: str reporter_email: str reporter_today_date: str reported_to_manufacturer: bool do_not_disclose_identity: bool class AgentTriageOutput(BaseModel): priority: Literal[ "Urgent provider review", "Provider review - non-urgent", "Continue monitoring / routine review", ] next_step: str explanation: str missing: List[str] form_3500b: Form3500BFields response = client.responses.parse( model=LIVE_AGENT_MODEL, input=[ {"role": "system", "content": TRIAGE_AGENT_SYSTEM_PROMPT}, { "role": "user", "content": json.dumps( build_agent_context(severity, extraction, case), ensure_ascii=True, ), }, ], text_format=AgentTriageOutput, max_output_tokens=2500, safety_identifier=f"sentinelplus_demo_{case.get('patient_name', 'unknown').replace(' ', '_')}", ) parsed = response.output_parsed if parsed is None: return None result = parsed.model_dump() result["source"] = "live_ai_agent" return result except Exception: return None def triage_agent(severity: Dict, extraction: Dict, case: Dict) -> Dict: live = run_live_triage_agent(severity, extraction, case) if live: return live is_severe = severity["label"] == "Severe ADR" missing = [] if extraction["medications"] == ["Not detected"]: missing.append("Confirm suspected medication name") if extraction["timing"] == ["Not detected"]: missing.append("Confirm symptom onset date or timing after medication use") if extraction["symptoms"] == ["Not detected"]: missing.append("Confirm symptom description") if not case["medications"]["otc"]: missing.append("Confirm OTC medication and supplement use") if is_severe: priority = "Urgent provider review" next_step = "Notify care team immediately and prepare high-priority ADR report for human review." else: priority = "Provider review — non-urgent" next_step = "Continue monitoring. Prepare ADR report and confirm any missing details before submission." return { "priority": priority, "next_step": next_step, "missing": missing or ["No major missing fields detected"], "explanation": build_agent_explanation(severity, extraction, case), "form_3500b": fallback_form_3500b_fields(case, severity), "source": "local_guardrailed_fallback", } def build_agent_explanation(severity: Dict, extraction: Dict, case: Dict) -> str: label = severity["label"] driver_text = ", ".join(severity["drivers"]) bio_flags = ", ".join(case["biometrics"]["flags"]) or "none flagged" meds = ", ".join(extraction["medications"]) symptoms = ", ".join(extraction["symptoms"]) timing = ", ".join(extraction["timing"]) return ( f"Severity was classified as **{label}** because the testimony included: {driver_text}. " f"SentinelPlus also considered medication context ({meds}), symptoms ({symptoms}), " f"timing ({timing}), and biometric flags ({bio_flags})." ) # --------------------------------------------------------------------- # DISPLAY BUILDERS # --------------------------------------------------------------------- def build_severity_html(severity: Dict) -> str: label = severity["label"] is_severe = label == "Severe ADR" color = "#c0392b" if is_severe else "#27ae60" bg = "#fdf0ef" if is_severe else "#edfaf1" icon = "🚨" if is_severe else "✅" pulse = "animation:pulse 1.5s ease-in-out infinite;" if is_severe else "" drivers_html = "".join( f'
  • {d}
    • ' for d in severity.get("drivers", []) ) return f"""
    {icon}
    {label}

    Evidence Cues Detected

    Model: DeBERTa v2 fine-tuned ADR severity classifier
    """ def build_severity_alert_html(severity: Dict) -> str: if severity["label"] != "Severe ADR": return "" return """
    ⚠ RECOMMENDED: Send to Doctor
    """ def build_fda_3500b_html(case: Dict, severity: Dict, form_fields: Optional[Dict] = None) -> str: profile = case["profile"] rx = case["medications"]["rx"] otc = case["medications"]["otc"] testimony = case.get("testimony", "") form = form_fields or fallback_form_3500b_fields(case, severity) name_parts = case["patient_name"].strip().split() initials = ". ".join(p[0].upper() for p in name_parts if p and p[0].isalpha()) + "." is_severe = severity["label"] == "Severe ADR" hospitalized = "hospital" in testimony.lower() and is_severe def chk(val: bool) -> str: return "☑" if val else "☐" def filled(val: str) -> str: return ( f'{escape(str(val))}' ) def nof() -> str: return 'Not on file' def field(name: str, fallback: str = "Not on file") -> str: value = form.get(name, fallback) if value is None or value == "": return fallback return str(value) def bool_field(name: str, fallback: bool = False) -> bool: return bool(form.get(name, fallback)) date_occurred = field("date_problem_occurred", case["checkins"][0]["Date"] if case.get("checkins") else "Not on file") narrative = escape(field("event_narrative", testimony[:4000])[:4000]) if testimony or form else "" primary_rx = rx[0] if rx else {} is_ongoing = bool_field("therapy_ongoing", primary_rx.get("Status", "").lower() in ("ongoing", "continuing")) otc_text = "\n".join(f"• {m['Product']}, {m['Dose']}" for m in otc) or "None listed" rx_text = "\n".join( f"• {m['Medication']}, {m['Dose']}, started {m['Start Date']} ({m['Status']})" for m in rx ) or "None listed" css = """""" return css + f"""
    Form FDA 3500B (09/2025)  |  OMB Control No. 0910-0291  |  Expiration Date: 09/30/2027  |  Pre-filled by SentinelPlus - Demonstration only
    FDA
    MedWatch FORM 3500B
    U.S. Dept. of Health and Human Services - Food and Drug Administration
    Consumer Voluntary Reporting of Adverse Events, Product Problems and Medication Errors
    Section A – About the Problem
    1. What kind of problem was it?
    {chk(bool_field("problem_type_hurt_or_side_effect", True))} Were hurt or had a bad side effect
    {chk(bool_field("problem_type_product_use_error"))} Used a product incorrectly
    {chk(bool_field("problem_type_product_quality"))} Noticed a problem with quality
    {chk(bool_field("problem_type_switching_maker"))} Problems after switching product maker
    2. Did any of the following happen?
    {chk(bool_field("outcome_hospitalization", hospitalized))} Hospitalization
    {chk(bool_field("outcome_required_help_prevent_harm", not is_severe))} Required help to prevent permanent harm
    {chk(bool_field("outcome_disability"))} Disability or health problem
    {chk(bool_field("outcome_birth_defect"))} Birth defect
    {chk(bool_field("outcome_life_threatening", is_severe))} Life-threatening
    {chk(bool_field("outcome_death"))} Death
    {chk(bool_field("outcome_other_serious_event"))} Other serious medical event
    3. Date the problem occurred
    {filled(date_occurred)}
    4. Tell us what happened
    {narrative}
    5. Relevant tests / laboratory results
    {escape(field("relevant_tests"))}
    Section B – Product Availability
    1. Do you still have the product?
    {filled(field("product_available"))}
    2. Do you have a picture of the product?
    {filled(field("product_picture_available"))}
    Section C – About the Products
    1. Name of product
    {filled(field("product_name", primary_rx.get("Medication", "Not listed"))) if primary_rx or field("product_name") != "Not on file" else nof()}
    2. Therapy on-going
    {chk(is_ongoing)} On-going
    3. Company / manufacturer
    {filled(field("manufacturer"))}
    4. Product type
    {filled(field("product_type"))}
    8. Strength / Dose
    {filled(field("strength", primary_rx.get("Dose","Not listed")))}
    6. Lot number
    {filled(field("lot_number"))}
    7. NDC number
    {filled(field("ndc_number"))}
    9. Quantity
    {filled(field("quantity"))}
    10. Frequency
    {filled(field("frequency"))}
    11. How was it taken or used?
    {filled(field("route"))}
    13. Duration
    {filled(field("duration"))}
    12a. Date first started
    {filled(field("therapy_start_date", primary_rx.get("Start Date","Not listed")))}
    12b. Date stopped
    {filled(field("therapy_stop_date"))}
    14. Why was the person using the product?
    {filled(field("reason_for_use", profile.get("condition",""))) if field("reason_for_use", profile.get("condition","")) else nof()}
    Section E – About the Person Who Had the Problem
    1. Initials
    {filled(field("patient_initials", initials))}
    2. Sex
    {chk(field("patient_sex", profile.get("sex","")).lower()=="male")} Male   {chk(field("patient_sex", profile.get("sex","")).lower()=="female")} Female
    3. Age
    {filled(field("patient_age", profile.get("age","")))} Year(s)
    4. Date of Birth
    {filled(field("patient_dob", profile.get("dob",""))) if field("patient_dob", profile.get("dob","")) else nof()}
    5. Weight
    {filled(field("patient_weight", profile.get("weight",""))) if field("patient_weight", profile.get("weight","")) else nof()}
    6. Race / Ethnicity
    {chk(field("patient_race_ethnicity", profile.get("race",""))=="American Indian or Alaska Native")} American Indian or Alaska Native
    {chk(field("patient_race_ethnicity", profile.get("race",""))=="Asian")} Asian
    {chk(field("patient_race_ethnicity", profile.get("race",""))=="Black or African American")} Black or African American
    {chk(field("patient_race_ethnicity", profile.get("race",""))=="Hispanic or Latino")} Hispanic or Latino
    {chk(field("patient_race_ethnicity", profile.get("race",""))=="White")} White
    7. Known medical conditions
    {filled(field("known_medical_conditions", profile.get("relevant_history",""))) if field("known_medical_conditions", profile.get("relevant_history","")) else nof()}
    8. Allergies
    {filled(field("allergies", profile.get("allergies",""))) if field("allergies", profile.get("allergies","")) else nof()}
    10. OTC medications, vitamins, supplements
    {escape(otc_text)}
    11. Current prescription medications
    {escape(rx_text)}
    Section F – About the Person Filling Out This Form
    1 & 2. Name
    {filled((field("reporter_first_name", "SentinelPlus Automated System") + " " + field("reporter_last_name", "")).strip())}
    7. Telephone number
    {filled(field("reporter_phone"))}
    3. Number / Street
    {filled(field("reporter_address"))}
    4. City and State / Province
    {filled(field("reporter_city_state"))}
    5. ZIP or Postal code
    {filled(field("reporter_zip"))}
    6. Country
    {filled(field("reporter_country"))}
    8. Email
    {filled(field("reporter_email", "DrOzOffice@Stayhealthy.com"))}
    9. Today's date
    {filled(field("reporter_today_date", "Pre-filled by SentinelPlus"))}
    10. Reported to manufacturer?
    {chk(bool_field("reported_to_manufacturer"))} Yes   {chk(not bool_field("reported_to_manufacturer"))} No
    11. Identity disclosure preference
    {chk(bool_field("do_not_disclose_identity"))} Do not disclose identity
    """ def highlight_text(text: str, extraction: Dict, severity: Dict) -> str: terms = [] for bucket in ["symptoms", "medications", "timing"]: terms.extend(v for v in extraction.get(bucket, []) if v != "Not detected") terms.extend(severity.get("drivers", [])) terms = sorted(set(terms), key=len, reverse=True) safe = escape(text) for term in terms: if not term: continue pattern = re.compile(re.escape(escape(term)), re.IGNORECASE) safe = pattern.sub(lambda m: f"{m.group(0)}", safe) return f"""
    {safe}

    Highlighted terms were detected by the extraction model or flagged as severity evidence cues.

    """ def results_table(severity: Dict, extraction: Dict) -> pd.DataFrame: rows = [ {"Output Type": "Severity Label", "Value": severity["label"]}, {"Output Type": "Evidence Cues", "Value": ", ".join(severity["drivers"])}, {"Output Type": "Medications Extracted", "Value": ", ".join(extraction["medications"])}, {"Output Type": "Symptoms Extracted", "Value": ", ".join(extraction["symptoms"])}, {"Output Type": "Timing Extracted", "Value": ", ".join(extraction["timing"])}, {"Output Type": "Patient Context Extracted", "Value": ", ".join(extraction["patient_context"])}, ] return pd.DataFrame(rows) def triage_markdown(triage: Dict) -> str: priority = triage["priority"] if "non-urgent" in priority.lower(): badge_color = "#e67e00" elif "urgent" in priority.lower(): badge_color = "#c0392b" else: badge_color = "#27ae60" missing_items = "".join( f'
  • {m}
    • ' for m in triage["missing"] ) return f"""
    SentinelPlus Triage Agent
    Priority
    {priority}
    Recommended Next Step
    {triage['next_step']}
    Agent Explanation
    {triage['explanation']}
    Missing Information Checklist
    """ def _chat_append(history: List[Dict], question: str, answer: str) -> List[Dict]: return history + [ {"role": "user", "content": question or ""}, {"role": "assistant", "content": answer}, ] def answer_triage_question(question: str, history: List[Dict], analysis_state: Dict, testimony: str): history = history or [] allowed, message = is_question_in_scope(question or "") if not allowed: return "", _chat_append(history, question, message) try: analysis_state = analysis_state or {} if not analysis_state.get("triage"): return "", _chat_append( history, question, "Run Analyze Updated Testimony first so I can answer from the current ADR analysis.", ) if (testimony or "").strip() != analysis_state.get("testimony", ""): return "", _chat_append( history, question, "The testimony has changed since the last analysis. Click Analyze Updated Testimony, then ask your question again.", ) case = analysis_state["case"] severity = analysis_state["severity"] extraction = analysis_state["extraction"] triage = analysis_state["triage"] client = _openai_client() if client is None: fallback = ( "Live AI Q&A is not connected yet. Based on the local triage output, " f"priority is: {triage['priority']}. The safest next step is: {triage['next_step']} " "I can help identify missing report details once OPENAI_API_KEY is configured." ) return "", _chat_append(history, question, fallback) response = client.responses.create( model=LIVE_AGENT_MODEL, input=[ {"role": "system", "content": TRIAGE_QA_SYSTEM_PROMPT}, { "role": "user", "content": json.dumps( { "case_context": build_agent_context(severity, extraction, case), "triage_output": { "priority": triage.get("priority"), "next_step": triage.get("next_step"), "explanation": triage.get("explanation"), "missing": triage.get("missing"), }, "recent_chat": history[-8:], "user_question": question, }, ensure_ascii=True, ), }, ], max_output_tokens=500, safety_identifier=f"sentinelplus_qa_{case.get('patient_name', 'unknown').replace(' ', '_')}", ) answer = getattr(response, "output_text", "").strip() if not answer: answer = "I could not generate a live response. Please ask a question about symptoms, timing, medications, or missing report details." return "", _chat_append(history, question, answer) except Exception as exc: return "", _chat_append(history, question, _friendly_live_ai_error(exc)) # --------------------------------------------------------------------- # MAIN ANALYSIS PIPELINE # --------------------------------------------------------------------- def analyze(case_name: str, testimony: str): try: case = DEMO_CASES[case_name].copy() testimony = (testimony or "").strip() if not testimony: msg = ( '
    ' 'Testimony needed
    ' 'Enter patient testimony, then click Analyze Updated Testimony.
    ' ) empty_df = pd.DataFrame([{"Output Type": "Input Needed", "Value": "Patient testimony is empty"}]) return msg, empty_df, msg, msg, "", build_fda_3500b_html(case, heuristic_severity_from_text("")), {} case["testimony"] = testimony severity = run_severity_model(testimony) extraction = run_medical_extraction_model(testimony) severity["drivers"] = extraction["symptoms"][:6] triage = triage_agent(severity, extraction, case) analysis_state = { "case_name": case_name, "testimony": testimony, "case": case, "severity": severity, "extraction": extraction, "triage": triage, } return ( build_severity_html(severity), results_table(severity, extraction), highlight_text(testimony, extraction, severity), triage_markdown(triage), build_severity_alert_html(severity), build_fda_3500b_html(case, severity, triage.get("form_3500b")), analysis_state, ) except Exception as exc: err_html = ( f'
    ' f'⚠ Model Error
    {escape(str(exc))}
    ' f'Ensure transformers, torch, and sentencepiece are installed ' f'and the model files are accessible.
    ' ) empty_df = pd.DataFrame([{"Output Type": "Error", "Value": str(exc)}]) return err_html, empty_df, err_html, err_html, "", err_html, {} # --------------------------------------------------------------------- # GRADIO UI # --------------------------------------------------------------------- _first_case = list(DEMO_CASES.keys())[0] CUSTOM_CSS = """ .gradio-container { background: #ffffff !important; } #main_tabs_container { background: #ffffff !important; } .tabitem { background: #ffffff !important; } #patient_dropdown, #testimony_box, #profile_panel, #biometric_panel, #meds_panel, #highlighted_panel, #severity_panel, #extraction_table, #triage_panel { border: 2px solid #6366F1 !important; border-radius: 10px !important; background: #ffffff !important; } #triage_chatbot, #triage_question { border: 2px solid #6366F1 !important; border-radius: 10px !important; background: #ffffff !important; } #extraction_table .table-wrap, #extraction_table .wrap, #extraction_table > div, #meds_panel .prose, #meds_panel > div { padding: 14px !important; } #biometric_panel, #biometric_panel *, #meds_panel, #meds_panel .prose, #meds_panel .prose *, #meds_panel table, #meds_panel td, #meds_panel th { color: #1f2937 !important; -webkit-text-fill-color: #1f2937 !important; } #meds_panel th { font-weight: 700 !important; } #extraction_table label, #extraction_table table, #extraction_table td, #extraction_table th { font-family: ui-sans-serif, system-ui, -apple-system, sans-serif !important; font-size: 14px !important; } """ with gr.Blocks(title="SentinelPlus - ADR Intelligent Assistant", theme=gr.themes.Soft(), css=CUSTOM_CSS) as demo: gr.HTML("""
    SentinelPlus - ADR Intelligent Assistant
    AI-assisted adverse drug reaction detection and FDA MedWatch reporting.
    """) main_tabs = gr.Tabs(elem_id="main_tabs_container") latest_analysis_state = gr.State({}) with main_tabs: # ---------------------------------------------------------------- Tab 1 with gr.Tab("1. Patient Check-in", id=0): gr.Markdown( "Select a patient to load their SentinelPlus profile, wearable biometrics, " "medications, and testimony." ) case_dropdown = gr.Dropdown( choices=list(DEMO_CASES.keys()), value=_first_case, label="Select Patient", elem_id="patient_dropdown", ) gr.HTML('
    PATIENT TESTIMONY
    ') testimony_box = gr.Textbox( label="Patient Testimony", value=DEMO_CASES[_first_case]["testimony"], lines=8, interactive=True, elem_id="testimony_box", ) analyze_btn = gr.Button("Analyze Updated Testimony", variant="primary", size="lg") gr.HTML('
    PATIENT PROFILE & BIOMETRICS
    ') with gr.Row(): profile_md = gr.HTML(format_profile(DEMO_CASES[_first_case]), elem_id="profile_panel") biometric_html = gr.HTML(format_biometrics(DEMO_CASES[_first_case]), elem_id="biometric_panel") gr.HTML('
    MEDICATIONS
    ') meds_md = gr.Markdown(format_medications(DEMO_CASES[_first_case]), elem_id="meds_panel") # ---------------------------------------------------------------- Tab 2 with gr.Tab("2. Severity Score", id=1): gr.Markdown( "ADR severity output from the DeBERTa v2 fine-tuned classifier. " "Detected terms in the patient testimony are highlighted below." ) highlighted_html = gr.HTML(label="Highlighted Testimony", elem_id="highlighted_panel") severity_display = gr.HTML(elem_id="severity_panel") # ---------------------------------------------------------------- Tab 3 with gr.Tab("3. ADR Triage", id=2): gr.Markdown( "Medical entity extraction (Clinical-AI-Apollo/Medical-NER) and " "SentinelPlus agentic triage summary." ) output_table = gr.Dataframe(label="Extraction Results", wrap=True, elem_id="extraction_table") triage_output = gr.HTML(elem_id="triage_panel") gr.Markdown( "Ask SentinelPlus documentation questions about symptoms, medication context, " "missing report details, or what to clarify with a clinician." ) triage_chatbot = gr.Chatbot( label="Live ADR Questions", height=240, elem_id="triage_chatbot", ) with gr.Row(): triage_question = gr.Textbox( label="Ask about this ADR case", placeholder="Example: What details should I clarify about my symptoms before this goes to the doctor?", lines=2, elem_id="triage_question", ) triage_ask_btn = gr.Button("Ask Agent", variant="primary") severity_alert = gr.HTML() send_btn = gr.Button("📋 Send to Doctor", variant="secondary", size="lg") # ---------------------------------------------------------------- Tab 4 with gr.Tab("4. Doctor's View", id=3, elem_id="doctors_view_tab"): gr.Markdown( "Pre-filled FDA MedWatch Form 3500B. " "Click **Submit to MedWatch** to simulate report submission." ) fda_form_output = gr.HTML() submit_btn = gr.Button("📤 Submit to MedWatch", variant="primary", size="lg") modal_html = gr.HTML(value="") # Outputs from analyze() in order _analysis_outputs = [ severity_display, output_table, highlighted_html, triage_output, severity_alert, fda_form_output, latest_analysis_state, ] # Dropdown change → reload scenario fields, then auto-run analysis case_dropdown.change( fn=load_scenario, inputs=case_dropdown, outputs=[testimony_box, profile_md, meds_md, biometric_html, triage_chatbot], ).then( fn=analyze, inputs=[case_dropdown, testimony_box], outputs=_analysis_outputs, ) analyze_btn.click( fn=analyze, inputs=[case_dropdown, testimony_box], outputs=_analysis_outputs, ) # Send to Doctor → rerun current testimony, then jump to Tab 4 send_btn.click( fn=analyze, inputs=[case_dropdown, testimony_box], outputs=_analysis_outputs, ).then( fn=lambda: gr.update(selected=3), outputs=main_tabs, ) triage_ask_btn.click( fn=answer_triage_question, inputs=[triage_question, triage_chatbot, latest_analysis_state, testimony_box], outputs=[triage_question, triage_chatbot], ) triage_question.submit( fn=answer_triage_question, inputs=[triage_question, triage_chatbot, latest_analysis_state, testimony_box], outputs=[triage_question, triage_chatbot], ) # Submit to MedWatch → show confirmation popup submit_btn.click( fn=lambda: MODAL_HTML, outputs=modal_html, ) # Page load → run analysis on the default patient demo.load( fn=analyze, inputs=[case_dropdown, testimony_box], outputs=_analysis_outputs, ) gr.Markdown(""" --- *SentinelPlus does not diagnose. All outputs require human clinical review before action.* """) if __name__ == "__main__": demo.launch()