# gradio_app.py """ LLM Clinical Assistant Gradio App (Gradio ≥ 4.0) Flow: 1. Chief concern → AI‑generated questions. 2. Clinician types answers (individual textboxes) + vitals/ROS → AI drafts Subjective & Objective. 3. Clinician edits S&O → AI suggests tests. 4. Whether or not tests are required, clinician can always proceed to (optionally) record test results → AI returns Assessment & Plan. """ import json import os import gradio as gr from helper_functions import ( generate_question_dict, generate_ros_dict, generating_so_pipeline, identify_tests_pipeline, ap_llm_pipeline, create_initial_submission, update_submission_with_so, update_submission_with_ap, update_submission_with_feedback, update_submission_with_qa_and_so, log_to_airtable, whisper_stt ) # ----------------------------------------------------------------------------- # Constants # ----------------------------------------------------------------------------- MAX_QUESTIONS = 12 # pre‑allocate this many answer boxes; extras hidden MAX_ROS = 5 # pre-allocate this many answer boxes for ROS; extras hidden # ----------------------------------------------------------------------------- # Helper functions # ----------------------------------------------------------------------------- def _qa_pairs_to_string(q_dict: dict, answers: list[str]) -> str: """Combine {question} + answer list → multiline string for the LLM.""" lines: list[str] = [] for i, question in enumerate(q_dict.values()): if i >= len(answers): break answer = answers[i] or "" lines.append(f"{question} {answer}") return "\n".join(lines) # ----------------------------------------------------------------------------- # UI – Gradio Blocks # ----------------------------------------------------------------------------- ORANGE_CSS = """ .orange-btn {background-color: #ff7f2a !important; color: white !important;} .orange-btn:hover {background-color: #e66a1f !important;} #logger-title {text-align: center; width: 100%; font-weight: bold; font-size: 1.1rem; margin-bottom: 4px;} #patient-title {text-align: center; width: 100%; font-weight: bold; font-size: 1.1rem; margin-bottom: 4px;} """ with gr.Blocks(title="Clinic LLM Assistant", css=ORANGE_CSS) as demo: # stt_interface = gr.Interface(fn=whisper_stt, nputs=gr.Audio(sources="upload", type="filepath"), outputs="text") gr.Markdown( """# Clinic LLM Assistant AI‑assisted SOAP‑note generator for community health workers in Kano, Nigeria.""" ) question_state = gr.State({}) ros_question_state = gr.State({}) so_state = gr.State("") submission_state = gr.State(None) # add submission id state to track airtable record gr.Markdown("Logger Information", elem_id="logger-title") with gr.Row(): name_input = gr.Textbox(label="Full Name", placeholder="Enter full name") email_input = gr.Textbox(label="Email Address", placeholder="Enter email address") gr.Markdown("Patient Information", elem_id="patient-title") with gr.Row(): age_input = gr.Number(label="Age (years)", precision=0, value=None) gender_input = gr.Dropdown(["Male", "Female"], label="Gender") cc_input = gr.Textbox(lines=2, label="Chief Complaint", elem_id="cc_input") cc_audio_input = gr.Audio(label="Record Chief Complaint", sources=["microphone"], type="filepath") cc_audio_input.change(fn=whisper_stt, inputs=cc_audio_input, outputs=cc_input) gen_q_btn = gr.Button("Generate Questions", elem_classes="orange-btn") # Answer section for initial questions (subjective) with gr.Column(visible=False) as answer_sec: answer_boxes = [] audio_boxes = [] for i in range(MAX_QUESTIONS): with gr.Row() as answer_row: answer_box = gr.Textbox( lines=2, label=f"Answer {i+1}", visible=True, elem_id=f"answer_box_{i}", ) # audio_box = gr.Audio( # sources=["microphone"], # type="filepath", # label="🎤", # visible=True, # elem_id = f"subjective_audio_{i}" # ) answer_boxes.append(answer_box) # audio_boxes.append(audio_box) submit_answers_btn = gr.Button("Submit Answers", elem_classes="orange-btn") # for textbox, audio in zip(answer_boxes, audio_boxes): # audio.change( # fn=whisper_stt, # Your speech-to-text function # inputs=audio, # outputs=textbox, # ) # ROS + Vitals + Other Info section (initially hidden) with gr.Column(visible=False) as ros_sec: gr.Markdown("### Review of Systems Questions") ros_answer_boxes = [ gr.Textbox( lines=2, label=f"ROS Answer {i+1}", visible=False, elem_id=f"ros_answer_box_{i}", ) for i in range(MAX_ROS) ] ros_answer_boxes = [] ros_audio_boxes = [] for i in range(MAX_ROS): with gr.Row() as ros_row: ros_box = gr.Textbox( lines=2, label=f"ROS Answer {i+1}", visible=True, elem_id=f"ros_answer_box_{i}", ) # ros_audio = gr.Audio( # sources=["microphone"], # type="filepath", # label="🎤", # visible=True, # ) ros_answer_boxes.append(ros_box) # ros_audio_boxes.append(ros_audio) vitals_input = gr.Textbox(lines=3, label="Vital Signs", elem_id="vitals_input") other_input = gr.Textbox( lines=3, label="Other Info / Physical Exam", elem_id="other_input", ) gen_so_btn = gr.Button("Create Subjective & Objective", elem_classes="orange-btn") # for textbox, audio in zip(ros_answer_boxes, ros_audio_boxes): # audio.change( # fn=whisper_stt, # Your speech-to-text function # inputs=audio, # outputs=textbox, # ) so_edit = gr.Textbox( lines=16, label="Subjective & Objective (editable)", visible=False, elem_id="so_edit", ) confirm_so_btn = gr.Button( "Confirm Subjective & Objective", visible=False, elem_classes="orange-btn" ) tests_md = gr.Markdown(visible=False) with gr.Column(visible=False) as results_sec: malaria_dd = gr.Dropdown( ["Normal", "Abnormal", "Not Done"], value="Not Done", label="Malaria RDT" ) typhoid_dd = gr.Dropdown( ["Normal", "Abnormal", "Not Done"], value="Not Done", label="Typhoid/Salmonella RDT" ) hpylori_dd = gr.Dropdown( ["Normal", "Abnormal", "Not Done"], value="Not Done", label="H Pylori RDT" ) pcv_dd = gr.Dropdown( ["Normal", "Abnormal", "Not Done"], value="Not Done", label="PCV / Anemia" ) urine_dd = gr.Dropdown( ["Normal", "Abnormal", "Not Done"], value="Not Done", label="Urinalysis" ) gen_ap_btn = gr.Button("Generate Assessment & Plan", visible=False, elem_classes="orange-btn") ap_edit_tb = gr.Textbox( lines=14, label="Assessment & Plan (editable)", visible=False, elem_id="ap_edit_tb", ) feedback_tb = gr.Textbox( lines=4, label="Optional Feedback", placeholder="How was this experience? Any suggestions or comments?", visible=False, elem_id="feedback_tb", ) submit_btn = gr.Button("Submit Feedback", elem_classes="orange-btn", visible=False) submission_status = gr.Markdown(visible=False) loading_md = gr.Markdown("⏳ Loading ...", visible=False) # ------------------------------------------------------------------------- # Callbacks # ------------------------------------------------------------------------- def _gen_questions(name, email, age, gender, cc): if not name or not email: return ( *[gr.update(visible=False) for _ in range(MAX_QUESTIONS * 2)], # hides both answer + audio boxes gr.update(visible=False), {}, None, gr.update( value="⚠️ Please enter both Full Name and Email address.", visible=True ), ) try: q_dict = generate_question_dict(age, gender, cc) submission_id = create_initial_submission( name=name, email=email, age=age, gender=gender, chief_complaint=cc ) except json.JSONDecodeError: return ( *[gr.update(visible=False) for _ in range(MAX_QUESTIONS)], gr.update(visible=False), {}, None, gr.update(value="⚠️ LLM failed to return valid JSON.", visible=True), ) except Exception as e: return ( *[gr.update(visible=False) for _ in range(MAX_QUESTIONS)], gr.update(visible=False), {}, None, gr.update(value=f"⚠️ Failed to create submission: {str(e)}", visible=True), ) box_updates = [ gr.update(label=list(q_dict.values())[i], visible=True, value="") if i < len(q_dict) else gr.update(visible=False) for i in range(MAX_QUESTIONS) ] # audio_updates = [ # gr.update(visible=True) # if i < len(q_dict) # else gr.update(visible=False) # for i in range(MAX_QUESTIONS) # ] return ( *box_updates, # *audio_updates, gr.update(visible=True), # show answers section q_dict, submission_id, gr.update(visible=False), ) gen_q_btn.click( lambda: gr.update(value="⏳ Loading ...", visible=True), None, loading_md, queue=False, ).then( _gen_questions, [name_input, email_input, age_input, gender_input, cc_input], [*answer_boxes, answer_sec, question_state, submission_state, loading_md], ) # Submit subjective answers → generate ROS questions def _generate_ros(*args): answers = list(args[:MAX_QUESTIONS]) q_dict = args[MAX_QUESTIONS] age, gender, cc = args[MAX_QUESTIONS + 1 : MAX_QUESTIONS + 4] qa_str = _qa_pairs_to_string(q_dict, answers) ros_dict = generate_ros_dict(age, gender, cc, qa_str) ros_updates = [ gr.update(label=list(ros_dict.values())[i], visible=True, value="") if i < len(ros_dict) else gr.update(visible=False) for i in range(MAX_ROS) ] return *ros_updates, gr.update(visible=True), ros_dict, gr.update(visible=False) submit_answers_btn.click( lambda: gr.update(visible=True), None, loading_md, queue=False, ).then( _generate_ros, [*answer_boxes, question_state, age_input, gender_input, cc_input], [*ros_answer_boxes, ros_sec, ros_question_state, loading_md], ) # Generate Subjective & Objective note from all answers def _gen_so(*args): answers = list(args[:MAX_QUESTIONS]) ros_answers = list(args[MAX_QUESTIONS : MAX_QUESTIONS + MAX_ROS]) vitals, other = args[MAX_QUESTIONS + MAX_ROS : MAX_QUESTIONS + MAX_ROS + 2] q_dict, ros_dict = args[-2:] qa_str = _qa_pairs_to_string(q_dict, answers) ros_str = _qa_pairs_to_string(ros_dict, ros_answers) so_text = generating_so_pipeline(qa_str, vitals, ros_str, other) # Optionally update submission with Q&A and S&O submission_id = submission_state.value if submission_id: try: # Combine Q&A pairs for logging qa_pairs = [] if isinstance(q_dict, dict): questions = list(q_dict.values()) for idx, question in enumerate(questions): if idx < len(answers): ans = answers[idx] or "" qa_pairs.append((question, ans)) if isinstance(ros_dict, dict): ros_questions = list(ros_dict.values()) for idx, question in enumerate(ros_questions): if idx < len(ros_answers): ans = ros_answers[idx] or "" qa_pairs.append((question, ans)) update_submission_with_qa_and_so(submission_id, qa_pairs, so_text) except Exception as e: print(f"Failed to update submission with Q&A and S&O: {e}") return ( gr.update(value=so_text, visible=True), gr.update(visible=True), so_text, gr.update(visible=False), ) gen_so_btn.click( lambda: gr.update(visible=True), None, loading_md, queue=False, ).then( _gen_so, [*answer_boxes, *ros_answer_boxes, vitals_input, other_input, question_state, ros_question_state], [so_edit, confirm_so_btn, so_state, loading_md], ) # Confirm S&O → Test suggestion def _confirm_so(so_text): suggestion = identify_tests_pipeline(so_text) need_results = suggestion.lower().startswith("the following tests") return ( gr.update(value=suggestion, visible=True), gr.update(visible=True), gr.update(visible=True), gr.update(visible=False), ) confirm_so_btn.click( lambda: gr.update(visible=True), None, loading_md, queue=False, ).then(_confirm_so, so_edit, [tests_md, results_sec, gen_ap_btn, loading_md]) # Generate Assessment & Plan def _gen_ap(so_text, mal, typ, hp, pcv, uri): ap_text = ap_llm_pipeline(so_text, mal, typ, hp, pcv, uri) return ( gr.update(value=ap_text, visible=True), gr.update(visible=False), gr.update(visible=True), gr.update(visible=True), gr.update(visible=True), ) gen_ap_btn.click( lambda: gr.update(visible=True), None, loading_md, queue=False, ).then( _gen_ap, [so_state, malaria_dd, typhoid_dd, hpylori_dd, pcv_dd, urine_dd], [ap_edit_tb, loading_md, feedback_tb, submit_btn, submission_status], ) # Submit feedback & log all info to Airtable def _log_to_airtable(ap_text, feedback_text, name, email, age, gender, cc, so_txt, q_state, *answers): try: qa_pairs = [ (list(q_state.values())[i], answers[i] or "") for i in range(min(len(q_state), len(answers))) ] log_to_airtable( ap_text, feedback_text, name=name, email=email, age=age, gender=gender, chief_complaint=cc, qa_pairs=qa_pairs, so_text=so_txt, ) return "Feedback submitted successfully!" except Exception as e: return f"Failed to submit feedback: {str(e)}" submit_btn.click( _log_to_airtable, inputs=[ ap_edit_tb, feedback_tb, name_input, email_input, age_input, gender_input, cc_input, so_edit, question_state, *answer_boxes, ], outputs=[submission_status], ) # ----------------------------------------------------------------------------- # Launch the app # ----------------------------------------------------------------------------- # Launch demo.queue().launch(share=True)