Update app.py

app.py

@@ -3,12 +3,9 @@ import os
 import pandas as pd
 import json
 import gradio as gr
-from typing import List, Tuple, Union, Generator, BinaryIO
-import hashlib
-import shutil
+from typing import List, Tuple, Union, Generator, BinaryIO, Dict, Any
 import re
 from datetime import datetime
-from concurrent.futures import ThreadPoolExecutor, as_completed
 
 # Setup directories
 persistent_dir = "/data/hf_cache"
@@ -42,22 +39,20 @@ def clean_response(text: str) -> str:
 def estimate_tokens(text: str) -> int:
     return len(text) // 3.5 + 1
 
-def extract_text_from_excel(file_obj) -> str:
-    """Handle both Gradio file objects and direct file paths"""
+def extract_text_from_excel(file_obj: Union[str, Dict[str, Any]]) -> str:
+    """Handle Gradio file upload object which is a dictionary with 'name' and other keys"""
     all_text = []
     try:
-        # Handle Gradio file object
-        if hasattr(file_obj, 'name'):
-            file_path = file_obj.name
-        # Handle direct file path
-        elif isinstance(file_obj, (str, os.PathLike)):
+        # Handle Gradio file upload object
+        if isinstance(file_obj, dict) and 'name' in file_obj:
+            file_path = file_obj['name']
+        elif isinstance(file_obj, str):
             file_path = file_obj
         else:
             raise ValueError("Unsupported file input type")
             
-        # Verify file exists
         if not os.path.exists(file_path):
-            raise FileNotFoundError(f"File not found at path: {file_path}")
+            raise FileNotFoundError(f"Temporary upload file not found at: {file_path}")
             
         xls = pd.ExcelFile(file_path)
         
@@ -76,44 +71,41 @@ def extract_text_from_excel(file_obj) -> str:
     except Exception as e:
         raise ValueError(f"❌ Error processing Excel file: {str(e)}")
 
-def split_text_into_chunks(text: str, max_tokens: int = MAX_CHUNK_TOKENS, max_chunks: int = 30) -> List[str]:
-    effective_max = max_tokens - PROMPT_OVERHEAD
+def split_text_into_chunks(text: str) -> List[str]:
+    effective_max = MAX_CHUNK_TOKENS - PROMPT_OVERHEAD
     lines, chunks, curr_chunk, curr_tokens = text.split("\n"), [], [], 0
     for line in lines:
         t = estimate_tokens(line)
         if curr_tokens + t > effective_max:
             if curr_chunk:
                 chunks.append("\n".join(curr_chunk))
-            if len(chunks) >= max_chunks:
-                break
             curr_chunk, curr_tokens = [line], t
         else:
             curr_chunk.append(line)
             curr_tokens += t
-    if curr_chunk and len(chunks) < max_chunks:
+    if curr_chunk:
         chunks.append("\n".join(curr_chunk))
     return chunks
 
 def build_prompt_from_text(chunk: str) -> str:
     return f"""
-### Unstructured Clinical Records
+### Clinical Records Analysis
 
-Analyze the following clinical notes and provide a detailed, concise summary focusing on:
-- Diagnostic Patterns
-- Medication Issues
-- Missed Opportunities
-- Inconsistencies
-- Follow-up Recommendations
+Please analyze these clinical notes and provide:
+- Key diagnostic indicators
+- Current medications and potential issues
+- Recommended follow-up actions
+- Any inconsistencies or concerns
 
 ---
 
 {chunk}
 
 ---
-Respond in well-structured bullet points with medical reasoning.
+Provide a structured response with clear medical reasoning.
 """
 
-def init_agent():
+def init_agent() -> TxAgent:
     tool_path = os.path.join(tool_cache_dir, "new_tool.json")
     if not os.path.exists(tool_path):
         default_tool = {
@@ -137,7 +129,7 @@ def init_agent():
     agent.init_model()
     return agent
 
-def stream_report(agent, input_file, full_output: str) -> Generator[Tuple[str, Union[str, None], str], None, None]:
+def stream_report(agent: TxAgent, input_file: Union[str, Dict[str, Any]], full_output: str) -> Generator[Tuple[str, Union[str, None], str], None, None]:
     accumulated_text = ""
     try:
         if input_file is None:
@@ -146,12 +138,11 @@ def stream_report(agent, input_file, full_output: str) -> Generator[Tuple[str, U
 
         try:
             text = extract_text_from_excel(input_file)
+            chunks = split_text_into_chunks(text)
         except Exception as e:
             yield f"❌ {str(e)}", None, ""
             return
 
-        chunks = split_text_into_chunks(text)
-
         for i, chunk in enumerate(chunks):
             prompt = build_prompt_from_text(chunk)
             partial = ""
@@ -160,87 +151,50 @@ def stream_report(agent, input_file, full_output: str) -> Generator[Tuple[str, U
                 max_new_tokens=MAX_NEW_TOKENS, max_token=MAX_MODEL_TOKENS,
                 call_agent=False, conversation=[]
             ):
-                if isinstance(res, str):
-                    partial += res
-                elif hasattr(res, "content"):
-                    partial += res.content
+                partial += res if isinstance(res, str) else res.content
+                
             cleaned = clean_response(partial)
-            accumulated_text += f"\n\n📄 **Chunk {i+1}**:\n{cleaned}"
+            accumulated_text += f"\n\n📄 Analysis Part {i+1}:\n{cleaned}"
             yield accumulated_text, None, ""
 
-        summary_prompt = f"Summarize this analysis in a final structured report:\n\n" + accumulated_text
+        summary_prompt = f"Please summarize this analysis:\n\n{accumulated_text}"
         final_report = ""
         for res in agent.run_gradio_chat(
             message=summary_prompt, history=[], temperature=0.2,
             max_new_tokens=MAX_NEW_TOKENS, max_token=MAX_MODEL_TOKENS,
             call_agent=False, conversation=[]
         ):
-            if isinstance(res, str):
-                final_report += res
-            elif hasattr(res, "content"):
-                final_report += res.content
+            final_report += res if isinstance(res, str) else res.content
 
         cleaned = clean_response(final_report)
-        accumulated_text += f"\n\n📊 **Final Summary**:\n{cleaned}"
         report_path = os.path.join(report_dir, f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md")
         with open(report_path, 'w') as f:
-            f.write(f"# 🧠 Final Patient Report\n\n{cleaned}")
+            f.write(f"# Clinical Analysis Report\n\n{cleaned}")
 
-        yield accumulated_text, report_path, cleaned
+        yield f"{accumulated_text}\n\n📊 Final Summary:\n{cleaned}", report_path, cleaned
 
     except Exception as e:
-        yield f"❌ Unexpected error: {str(e)}", None, ""
-
-def create_ui(agent):
-    with gr.Blocks(css="""
-        body {
-            background: #10141f;
-            color: #ffffff;
-            font-family: 'Inter', sans-serif;
-            margin: 0;
-            padding: 0;
-        }
-        .gradio-container {
-            padding: 30px;
-            width: 100vw;
-            max-width: 100%;
-            border-radius: 0;
-            background-color: #1a1f2e;
-        }
-        .output-markdown {
-            background-color: #131720;
-            border-radius: 12px;
-            padding: 20px;
-            min-height: 600px;
-            overflow-y: auto;
-            border: 1px solid #2c3344;
-        }
-        .gr-button {
-            background: linear-gradient(135deg, #4b4ced, #37b6e9);
-            color: white;
-            font-weight: 500;
-            border: none;
-            padding: 10px 20px;
-            border-radius: 8px;
-            transition: background 0.3s ease;
-        }
-        .gr-button:hover {
-            background: linear-gradient(135deg, #37b6e9, #4b4ced);
-        }
-    """) as demo:
-        gr.Markdown("""# 🧠 Clinical Reasoning Assistant
-Upload clinical Excel records below and click **Analyze** to generate a medical summary.
-""")
-        file_upload = gr.File(label="Upload Excel File", file_types=[".xlsx"])
-        analyze_btn = gr.Button("Analyze")
-        report_output_markdown = gr.Markdown(elem_classes="output-markdown")
-        report_file = gr.File(label="Download Report", visible=False)
-        full_output = gr.State(value="")
+        yield f"❌ Processing error: {str(e)}", None, ""
+
+def create_ui(agent: TxAgent) -> gr.Blocks:
+    with gr.Blocks(theme=gr.themes.Soft(), css=".gradio-container {max-width: 900px !important}") as demo:
+        gr.Markdown("""# Clinical Records Analyzer""")
+        with gr.Row():
+            file_upload = gr.File(label="Upload Excel File", file_types=[".xlsx"])
+            analyze_btn = gr.Button("Analyze", variant="primary")
+        
+        with gr.Row():
+            with gr.Column(scale=2):
+                report_output = gr.Markdown()
+            with gr.Column(scale=1):
+                report_file = gr.File(label="Download Report", visible=False)
+        
+        full_output = gr.State()
 
         analyze_btn.click(
             fn=stream_report,
             inputs=[file_upload, full_output],
-            outputs=[report_output_markdown, report_file, full_output]
+            outputs=[report_output, report_file, full_output]
         )
 
     return demo
@@ -250,12 +204,10 @@ if __name__ == "__main__":
         agent = init_agent()
         demo = create_ui(agent)
         demo.launch(
-            server_name="0.0.0.0", 
-            server_port=7860, 
-            allowed_paths=["/data/hf_cache/reports"], 
-            share=True,
-            show_error=True
+            server_name="0.0.0.0",
+            server_port=7860,
+            share=False
         )
     except Exception as e:
-        print(f"Error: {str(e)}", file=sys.stderr)
+        print(f"Application error: {str(e)}", file=sys.stderr)
         sys.exit(1)