|
|
import gradio as gr |
|
|
import os |
|
|
import sys |
|
|
import pandas as pd |
|
|
import pdfplumber |
|
|
|
|
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src"))) |
|
|
from txagent.txagent import TxAgent |
|
|
|
|
|
|
|
|
def extract_all_text_from_csv(file_path): |
|
|
try: |
|
|
df = pd.read_csv(file_path, low_memory=False) |
|
|
return df.to_string(index=False) |
|
|
except Exception as e: |
|
|
return f"Error parsing CSV: {e}" |
|
|
|
|
|
|
|
|
def extract_all_text_from_pdf(file_path): |
|
|
extracted = [] |
|
|
try: |
|
|
with pdfplumber.open(file_path) as pdf: |
|
|
for page in pdf.pages: |
|
|
tables = page.extract_tables() |
|
|
for table in tables: |
|
|
for row in table: |
|
|
if any(row): |
|
|
extracted.append("\t".join([cell or "" for cell in row])) |
|
|
return "\n".join(extracted) |
|
|
except Exception as e: |
|
|
return f"Error parsing PDF: {e}" |
|
|
|
|
|
|
|
|
def create_ui(agent: TxAgent): |
|
|
with gr.Blocks(theme=gr.themes.Soft()) as demo: |
|
|
gr.Markdown("<h1 style='text-align: center;'>🧠 CPS: Clinical Processing System</h1>") |
|
|
chatbot = gr.Chatbot(label="CPS Assistant", height=600, type="messages") |
|
|
|
|
|
|
|
|
with gr.Row(): |
|
|
uploaded_files = gr.File( |
|
|
label="📎", file_types=[".pdf", ".txt", ".docx", ".jpg", ".png", ".csv"], |
|
|
file_count="multiple", visible=False |
|
|
) |
|
|
|
|
|
with gr.Column(scale=10): |
|
|
message_input = gr.Textbox( |
|
|
placeholder="Type your medical question or upload files...", show_label=False, scale=10 |
|
|
) |
|
|
|
|
|
with gr.Column(scale=1, min_width=60): |
|
|
file_icon = gr.UploadButton("📎", file_types=[".pdf", ".csv", ".docx", ".txt", ".jpg", ".png"], file_count="multiple") |
|
|
|
|
|
send_button = gr.Button("Send", variant="primary") |
|
|
|
|
|
conversation_state = gr.State([]) |
|
|
|
|
|
def handle_chat(message, history, conversation, new_files): |
|
|
context = ( |
|
|
"You are a clinical AI reviewing medical interview or form data. " |
|
|
"Analyze the extracted content and reason step-by-step about what the doctor could have missed. " |
|
|
"Don't answer yet — just reason." |
|
|
) |
|
|
|
|
|
if new_files: |
|
|
extracted_text = "" |
|
|
for file in new_files: |
|
|
path = file.name |
|
|
if path.endswith(".csv"): |
|
|
extracted_text += extract_all_text_from_csv(path) + "\n" |
|
|
elif path.endswith(".pdf"): |
|
|
extracted_text += extract_all_text_from_pdf(path) + "\n" |
|
|
else: |
|
|
extracted_text += f"(Uploaded file: {os.path.basename(path)})\n" |
|
|
|
|
|
message = f"{context}\n\n---\n{extracted_text.strip()}\n---\n\nNow reason what the doctor might have missed." |
|
|
|
|
|
generator = agent.run_gradio_chat( |
|
|
message=message, |
|
|
history=history, |
|
|
temperature=0.3, |
|
|
max_new_tokens=1024, |
|
|
max_token=8192, |
|
|
call_agent=False, |
|
|
conversation=conversation, |
|
|
uploaded_files=new_files, |
|
|
max_round=30 |
|
|
) |
|
|
for update in generator: |
|
|
yield update |
|
|
|
|
|
|
|
|
file_icon.upload(fn=None, inputs=[], outputs=[uploaded_files]) |
|
|
send_button.click(fn=handle_chat, inputs=[message_input, chatbot, conversation_state, uploaded_files], outputs=chatbot) |
|
|
message_input.submit(fn=handle_chat, inputs=[message_input, chatbot, conversation_state, uploaded_files], outputs=chatbot) |
|
|
|
|
|
gr.Examples([["Upload your medical form and ask what the doctor might’ve missed."]], inputs=message_input) |
|
|
|
|
|
return demo |
|
|
|
