Spaces:
Sleeping
Sleeping
| # 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) |