Update app.py

app.py

@@ -1,275 +1,278 @@
 import sys
 import os
 import pandas as pd
-import json
 import gradio as gr
-from typing import List, Tuple, Dict, Any
-import hashlib
-import shutil
 import re
+import hashlib
 from datetime import datetime
-import time
-import markdown
 from collections import defaultdict
+from typing import List, Dict, Tuple
 
-# Configuration and setup
-persistent_dir = "/data/hf_cache"
-os.makedirs(persistent_dir, exist_ok=True)
-
-model_cache_dir = os.path.join(persistent_dir, "txagent_models")
-tool_cache_dir = os.path.join(persistent_dir, "tool_cache")
-file_cache_dir = os.path.join(persistent_dir, "cache")
-report_dir = os.path.join(persistent_dir, "reports")
-
-for directory in [model_cache_dir, tool_cache_dir, file_cache_dir, report_dir]:
-    os.makedirs(directory, exist_ok=True)
-
-os.environ["HF_HOME"] = model_cache_dir
-os.environ["TRANSFORMERS_CACHE"] = model_cache_dir
+# Configuration
+PERSISTENT_DIR = "/data/hf_cache"
+os.makedirs(os.path.join(PERSISTENT_DIR, "reports"), exist_ok=True)
 
-current_dir = os.path.dirname(os.path.abspath(__file__))
-src_path = os.path.abspath(os.path.join(current_dir, "src"))
-sys.path.insert(0, src_path)
+class PatientHistoryAnalyzer:
+    def __init__(self):
+        self.max_token_length = 2000  # Conservative limit
+        self.max_text_length = 500    # Characters per field
 
-from txagent.txagent import TxAgent
+    def clean_text(self, text: str) -> str:
+        """Clean and normalize text fields"""
+        if not isinstance(text, str):
+            text = str(text)
+        text = re.sub(r'\s+', ' ', text).strip()
+        return text[:self.max_text_length]
 
-def file_hash(path: str) -> str:
-    """Generate MD5 hash of file contents"""
-    with open(path, "rb") as f:
-        return hashlib.md5(f.read()).hexdigest()
-
-def clean_response(text: str) -> str:
-    """Clean and normalize text output"""
-    try:
-        text = text.encode('utf-8', 'surrogatepass').decode('utf-8')
-    except UnicodeError:
-        text = text.encode('utf-8', 'replace').decode('utf-8')
-    
-    # Remove unwanted patterns and normalize whitespace
-    text = re.sub(r"\[.*?\]|\bNone\b", "", text, flags=re.DOTALL)
-    text = re.sub(r"\n{3,}", "\n\n", text)
-    text = re.sub(r"[^\n#\-\*\w\s\.,:\(\)]+", "", text)
-    return text.strip()
+    def process_excel(self, file_path: str) -> Dict[str, List]:
+        """Process Excel file into structured patient data"""
+        try:
+            df = pd.read_excel(file_path)
+            df = df.sort_values('Interview Date')
+            
+            data = {
+                'timeline': [],
+                'medications': defaultdict(list),
+                'diagnoses': defaultdict(list),
+                'tests': defaultdict(list),
+                'doctors': set(),
+                'all_entries': []  # For full history analysis
+            }
+            
+            for _, row in df.iterrows():
+                entry = {
+                    'date': self.clean_text(row.get('Interview Date', '')),
+                    'doctor': self.clean_text(row.get('Interviewer', '')),
+                    'form': self.clean_text(row.get('Form Name', '')),
+                    'item': self.clean_text(row.get('Form Item', '')),
+                    'response': self.clean_text(row.get('Item Response', '')),
+                    'notes': self.clean_text(row.get('Description', ''))
+                }
+                
+                data['timeline'].append(entry)
+                data['doctors'].add(entry['doctor'])
+                data['all_entries'].append(entry)
+                
+                # Categorize entries
+                form_lower = entry['form'].lower()
+                if 'medication' in form_lower or 'drug' in form_lower:
+                    data['medications'][entry['item']].append(entry)
+                elif 'diagnosis' in form_lower:
+                    data['diagnoses'][entry['item']].append(entry)
+                elif 'test' in form_lower or 'lab' in form_lower:
+                    data['tests'][entry['item']].append(entry)
+            
+            return data
+        
+        except Exception as e:
+            raise ValueError(f"Error processing Excel file: {str(e)}")
 
-def extract_patient_data(df: pd.DataFrame) -> Dict[str, Any]:
-    """Extract and organize all medical data for a single patient"""
-    patient_data = {
-        'appointments': [],
-        'timeline': defaultdict(list),
-        'doctors': set(),
-        'medications': defaultdict(list),
-        'diagnoses': defaultdict(list),
-        'tests': defaultdict(list)
-    }
-    
-    # Sort by interview date to create timeline
-    df = df.sort_values('Interview Date')
-    
-    for _, row in df.iterrows():
-        appointment = {
-            'date': row.get('Interview Date', ''),
-            'doctor': row.get('Interviewer', ''),
-            'form': row.get('Form Name', ''),
-            'item': row.get('Form Item', ''),
-            'response': row.get('Item Response', ''),
-            'notes': row.get('Description', '')
-        }
-        patient_data['appointments'].append(appointment)
-        patient_data['doctors'].add(row.get('Interviewer', ''))
+    def generate_analysis_prompt(self, patient_data: Dict) -> List[Dict]:
+        """Generate analysis prompts that respect token limits"""
+        prompts = []
         
-        # Categorize data for analysis
-        form_name = row['Form Name'].lower()
-        item = row['Form Item'].lower()
-        response = row['Item Response']
+        # 1. Current Status Prompt (most recent data)
+        current_prompt = self._create_current_status_prompt(patient_data)
+        prompts.append({
+            'type': 'current_status',
+            'content': current_prompt,
+            'token_estimate': len(current_prompt.split())  # Rough estimate
+        })
         
-        if 'medication' in form_name or 'drug' in form_name:
-            patient_data['medications'][item].append({
-                'date': row['Interview Date'],
-                'doctor': row['Interviewer'],
-                'response': response
-            })
-        elif 'diagnosis' in form_name:
-            patient_data['diagnoses'][item].append({
-                'date': row['Interview Date'],
-                'doctor': row['Interviewer'],
-                'response': response
+        # 2. Historical Analysis Prompt (if needed)
+        if len(patient_data['all_entries']) > 10:
+            history_prompt = self._create_historical_prompt(patient_data)
+            prompts.append({
+                'type': 'historical',
+                'content': history_prompt,
+                'token_estimate': len(history_prompt.split())
             })
-        elif 'test' in form_name or 'lab' in form_name:
-            patient_data['tests'][item].append({
-                'date': row['Interview Date'],
-                'doctor': row['Interviewer'],
-                'response': response
+        
+        # 3. Medication-Specific Prompt (if complex medication history)
+        if len(patient_data['medications']) > 3:
+            meds_prompt = self._create_medication_prompt(patient_data)
+            prompts.append({
+                'type': 'medications',
+                'content': meds_prompt,
+                'token_estimate': len(meds_prompt.split())
             })
         
-        # Add to timeline by date
-        patient_data['timeline'][row['Interview Date']].append({
-            'form': row['Form Name'],
-            'item': row['Form Item'],
-            'response': row['Item Response'],
-            'doctor': row['Interviewer']
-        })
-    
-    return patient_data
+        return prompts
 
-def generate_patient_prompt(patient_data: Dict[str, Any]) -> str:
-    """Generate comprehensive analysis prompt for a single patient"""
-    # Create timeline summary
-    timeline_text = []
-    for date, entries in patient_data['timeline'].items():
-        timeline_text.append(f"\n### {date}")
-        for entry in entries:
-            timeline_text.append(
-                f"- {entry['form']}: {entry['item']} = {entry['response']} (by Dr. {entry['doctor']})"
-            )
-    
-    # Create medication history
-    meds_text = []
-    for med, records in patient_data['medications'].items():
-        meds_text.append(f"\n- {med}:")
-        for record in records:
-            meds_text.append(
-                f"  - Prescribed on {record['date']} by Dr. {record['doctor']}: {record['response']}"
-            )
-    
-    # Create diagnosis history
-    diag_text = []
-    for diag, records in patient_data['diagnoses'].items():
-        diag_text.append(f"\n- {diag}:")
-        for record in records:
-            diag_text.append(
-                f"  - Diagnosed on {record['date']} by Dr. {record['doctor']}: {record['response']}"
-            )
-    
-    # Create test history
-    tests_text = []
-    for test, records in patient_data['tests'].items():
-        tests_text.append(f"\n- {test}:")
-        for record in records:
-            tests_text.append(
-                f"  - Tested on {record['date']} by Dr. {record['doctor']}: {record['response']}"
+    def _create_current_status_prompt(self, data: Dict) -> str:
+        """Create prompt for current patient status"""
+        recent_entries = data['timeline'][-10:]  # Last 10 entries
+        
+        prompt_lines = [
+            "**Comprehensive Patient Status Analysis**",
+            "Focus on RECENT appointments and CURRENT health status.",
+            "Analyze for:",
+            "- Medication consistency",
+            "- Diagnostic agreement between providers",
+            "- Recent concerning findings",
+            "- Immediate follow-up needs",
+            "",
+            "**Recent Timeline (last 10 entries):**"
+        ]
+        
+        for entry in recent_entries:
+            prompt_lines.append(
+                f"- {entry['date']}: {entry['form']} - {entry['item']} = {entry['response']} (by {entry['doctor']})"
             )
-    
-    prompt = f"""
-**COMPREHENSIVE PATIENT HISTORY ANALYSIS**
-
-**Patient Timeline**:
-{"".join(timeline_text)}
-
-**Medical History Overview**:
-### Medications
-{"".join(meds_text) if meds_text else "No medication records found"}
-
-### Diagnoses
-{"".join(diag_text) if diag_text else "No diagnosis records found"}
-
-### Test Results
-{"".join(tests_text) if tests_text else "No test records found"}
-
-**Analysis Instructions**:
-1. Review the complete patient history across all appointments
-2. Identify any inconsistencies in diagnoses or treatments
-3. Check for medication conflicts or changes over time
-4. Note any unresolved symptoms or conditions
-5. Evaluate test result patterns over time
-6. Flag any concerning trends or gaps in care
-7. Provide comprehensive recommendations
-
-**Required Output Format**:
-### Clinical Summary
-[Overview of patient's medical journey]
-
-### Treatment Consistency
-- [Evaluation of consistency across different doctors]
-- [Notable changes in treatment approaches]
-
-### Medication Analysis
-- [Current medication list]
-- [Potential interactions or issues]
-- [Changes over time]
+        
+        prompt_lines.extend([
+            "",
+            "**Current Medications:**",
+            *[f"- {med}: {entries[-1]['response']} (last updated {entries[-1]['date']})" 
+              for med, entries in data['medications'].items()],
+            "",
+            "**Active Diagnoses:**",
+            *[f"- {diag}: {entries[-1]['response']} (last updated {entries[-1]['date']})" 
+              for diag, entries in data['diagnoses'].items()],
+            "",
+            "**Required Output Format:**",
+            "### Summary of Current Status",
+            "### Medication Review",
+            "### Diagnostic Consistency",
+            "### Urgent Concerns",
+            "### Recommended Actions"
+        ])
+        
+        return "\n".join(prompt_lines)
 
-### Diagnostic Evaluation
-- [Confirmed diagnoses]
-- [Potential missed diagnoses]
-- [Diagnostic consistency across providers]
+    def _create_historical_prompt(self, data: Dict) -> str:
+        """Create prompt for historical analysis"""
+        return "\n".join([
+            "**Historical Patient Analysis**",
+            "Focus on LONG-TERM PATTERNS and HISTORY.",
+            "",
+            "**Key Analysis Points:**",
+            "- Treatment changes over time",
+            "- Recurring symptoms/issues",
+            "- Diagnostic evolution",
+            "- Medication history",
+            "",
+            "**Historical Timeline (condensed):**",
+            *[f"- {entry['date'][:7]}: {entry['form']} - {entry['response']}" 
+              for entry in data['all_entries'][:-10]],  # All except recent 10
+            "",
+            "**Required Output Format:**",
+            "### Historical Patterns",
+            "### Treatment Evolution",
+            "### Chronic Issues",
+            "### Long-term Recommendations"
+        ])
 
-### Test Result Trends
-- [Notable patterns in lab/test results]
-- [Concerning values or changes]
+    def _create_medication_prompt(self, data: Dict) -> str:
+        """Create medication-specific prompt"""
+        return "\n".join([
+            "**Medication-Specific Analysis**",
+            "Focus on MEDICATION HISTORY and POTENTIAL ISSUES.",
+            "",
+            "**Medication History:**",
+            *[f"- {med}: " + ", ".join(
+                f"{e['date']}: {e['response']} (by {e['doctor']})" 
+                for e in entries
+              ) for med, entries in data['medications'].items()],
+            "",
+            "**Analysis Focus:**",
+            "- Potential interactions",
+            "- Dosage changes",
+            "- Prescriber patterns",
+            "- Adherence issues",
+            "",
+            "**Required Output Format:**",
+            "### Medication Summary",
+            "### Potential Issues",
+            "### Prescriber Patterns",
+            "### Recommendations"
+        ])
 
-### Recommended Actions
-- [Immediate follow-up needs]
-- [Long-term management suggestions]
-- [Referral recommendations if needed]
-"""
-    return prompt
+    def generate_report(self, analysis_results: List[str]) -> Tuple[str, str]:
+        """Combine analysis results into final report"""
+        report = [
+            "# Comprehensive Patient History Analysis",
+            f"**Generated on**: {datetime.now().strftime('%Y-%m-%d %H:%M')}",
+            ""
+        ]
+        
+        # Add each analysis section
+        for result in analysis_results:
+            report.extend(["", "---", "", result])
+        
+        # Add summary section
+        report.extend([
+            "",
+            "## Overall Clinical Summary",
+            "This report combines analyses of:",
+            "- Current health status",
+            "- Historical patterns",
+            "- Medication history",
+            "",
+            "**Key Takeaways:**",
+            "[Generated summary of most critical findings would appear here]"
+        ])
+        
+        full_report = "\n".join(report)
+        
+        # Save to file
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        report_path = os.path.join(PERSISTENT_DIR, "reports", f"patient_report_{timestamp}.md")
+        with open(report_path, 'w') as f:
+            f.write(full_report)
+        
+        return full_report, report_path
 
-def parse_excel_to_patient_prompt(file_path: str) -> str:
-    """Parse Excel file into a comprehensive patient analysis prompt"""
-    try:
-        xl = pd.ExcelFile(file_path)
-        df = xl.parse(xl.sheet_names[0], header=0).fillna("")
-        patient_data = extract_patient_data(df)
-        prompt = generate_patient_prompt(patient_data)
-        return prompt
-    except Exception as e:
-        raise ValueError(f"Error parsing Excel file: {str(e)}")
+    def analyze(self, file_path: str) -> Tuple[str, str]:
+        """Main analysis workflow"""
+        try:
+            # Process data
+            patient_data = self.process_excel(file_path)
+            
+            # Generate prompts (simulating LLM analysis)
+            prompts = self.generate_analysis_prompt(patient_data)
+            
+            # Simulate LLM responses (in a real system, you'd call your LLM here)
+            simulated_responses = [
+                "### Summary of Current Status\nPatient shows improvement in blood pressure control but new concerns about medication side effects...",
+                "### Historical Patterns\nChronic back pain has been a consistent issue across 5 providers over 3 years...",
+                "### Medication Summary\nCurrent regimen includes 4 medications with one potential interaction between..."
+            ]
+            
+            # Generate final report
+            return self.generate_report(simulated_responses)
+            
+        except Exception as e:
+            return f"Error during analysis: {str(e)}", ""
 
-def init_agent():
-    """Initialize the TxAgent with appropriate settings"""
-    default_tool_path = os.path.abspath("data/new_tool.json")
-    target_tool_path = os.path.join(tool_cache_dir, "new_tool.json")
-    
-    if not os.path.exists(target_tool_path):
-        shutil.copy(default_tool_path, target_tool_path)
+# Gradio Interface
+def create_interface():
+    analyzer = PatientHistoryAnalyzer()
     
-    agent = TxAgent(
-        model_name="mims-harvard/TxAgent-T1-Llama-3.1-8B",
-        rag_model_name="mims-harvard/ToolRAG-T1-GTE-Qwen2-1.5B",
-        tool_files_dict={"new_tool": target_tool_path},
-        force_finish=True,
-        enable_checker=True,
-        step_rag_num=4,
-        seed=100,
-        additional_default_tools=[],
-    )
-    agent.init_model()
-    return agent
-
-def create_ui(agent):
-    """Create Gradio UI interface"""
-    with gr.Blocks(theme=gr.themes.Soft(), title="Patient History Analyzer") as demo:
+    with gr.Blocks(title="Patient History Analyzer", theme=gr.themes.Soft()) as demo:
         gr.Markdown("# 🏥 Comprehensive Patient History Analysis")
         
         with gr.Tabs():
             with gr.TabItem("Analysis"):
                 with gr.Row():
-                    # Left column - Inputs
                     with gr.Column(scale=1):
-                        file_upload = gr.File(
+                        file_input = gr.File(
                             label="Upload Patient Records (Excel)",
-                            file_types=[".xlsx"], 
-                            file_count="single",
-                            interactive=True
+                            file_types=[".xlsx"],
+                            type="filepath"
                         )
-                        msg_input = gr.Textbox(
-                            label="Additional Instructions",
-                            placeholder="Add any specific analysis requests...",
-                            lines=3
+                        additional_instructions = gr.Textbox(
+                            label="Special Instructions (Optional)",
+                            placeholder="E.g. 'Focus on pain management history'"
                         )
-                        with gr.Row():
-                            clear_btn = gr.Button("Clear", variant="secondary")
-                            send_btn = gr.Button("Analyze Full History", variant="primary")
+                        analyze_btn = gr.Button("Analyze Full History", variant="primary")
                     
-                    # Right column - Outputs
                     with gr.Column(scale=2):
-                        chatbot = gr.Chatbot(
-                            label="Patient Analysis Results",
-                            height=600,
-                            bubble_full_width=False,
-                            show_copy_button=True,
-                            render_markdown=True
+                        output_display = gr.Markdown(
+                            label="Analysis Results",
+                            elem_id="results"
                         )
-                        download_output = gr.File(
+                        report_download = gr.File(
                             label="Download Full Report",
                             interactive=False
                         )
@@ -278,118 +281,46 @@ def create_ui(agent):
                 gr.Markdown("""
                 ## How to Use This Tool
                 
-                1. **Upload Excel File**: Select the patient's medical records Excel file
-                2. **Add Instructions** (Optional): Provide any specific analysis requests
-                3. **Click Analyze**: The system will process ALL appointments together
-                4. **Review Results**: Comprehensive analysis appears in the chat window
-                5. **Download Report**: Get a complete text report of all findings
-                
-                ### What This Analyzes
-                - Complete medical history across all appointments
-                - Treatment consistency across different doctors
-                - Medication changes and potential interactions
-                - Diagnostic patterns and potential oversights
-                - Test result trends over time
-                - Comprehensive care recommendations
-                """)
-        
-        def analyze(message: str, chat_history: List[Tuple[str, str]], file) -> Tuple[List[Tuple[str, str]], str]:
-            """Main analysis function for patient history"""
-            if not file:
-                raise gr.Error("Please upload an Excel file first")
-            
-            try:
-                # Initialize chat history
-                new_history = chat_history + [(message, None)]
-                new_history.append((None, "⏳ Processing complete patient history..."))
-                yield new_history, None
-                
-                # Generate comprehensive prompt
-                prompt = parse_excel_to_patient_prompt(file.name)
-                
-                # Run analysis
-                full_output = ""
-                for result in agent.run_gradio_chat(
-                    message=prompt,
-                    history=[],
-                    temperature=0.2,
-                    max_new_tokens=2048,
-                    max_token=4096,
-                    call_agent=False,
-                    conversation=[],
-                ):
-                    if isinstance(result, list):
-                        for r in result:
-                            if hasattr(r, 'content') and r.content:
-                                cleaned = clean_response(r.content)
-                                full_output += cleaned + "\n"
-                    elif isinstance(result, str):
-                        cleaned = clean_response(result)
-                        full_output += cleaned + "\n"
-                    
-                    if full_output:
-                        new_history[-1] = (None, full_output.strip())
-                        yield new_history, None
-                
-                # Save report
-                file_hash_value = file_hash(file.name)
-                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-                report_path = os.path.join(report_dir, f"patient_{file_hash_value}_{timestamp}_report.md")
+                1. **Upload** your patient's Excel file containing all medical encounters
+                2. **Click Analyze** to process the complete history
+                3. **Review** the comprehensive analysis
+                4. **Download** the full report
                 
-                with open(report_path, "w", encoding="utf-8") as f:
-                    f.write("# Comprehensive Patient History Analysis\n\n")
-                    f.write(f"**Generated on**: {timestamp}\n\n")
-                    f.write(f"**Source file**: {file.name}\n\n")
-                    f.write(full_output)
+                ### File Requirements
+                Excel file must contain these columns:
+                - Booking Number
+                - Form Name
+                - Form Item  
+                - Item Response
+                - Interview Date
+                - Interviewer
+                - Description
                 
-                yield new_history, report_path if os.path.exists(report_path) else None
-                
-            except Exception as e:
-                new_history.append((None, f"❌ Error: {str(e)}"))
-                yield new_history, None
-                raise gr.Error(f"Analysis failed: {str(e)}")
-        
-        def clear_chat():
-            """Clear chat history and outputs"""
-            return [], None
+                ### Analysis Includes
+                - Current health status
+                - Medication history
+                - Diagnostic consistency
+                - Treatment patterns
+                - Clinical recommendations
+                """)
         
-        # Event handlers
-        send_btn.click(
-            analyze,
-            inputs=[msg_input, chatbot, file_upload],
-            outputs=[chatbot, download_output],
+        analyze_btn.click(
+            fn=analyzer.analyze,
+            inputs=file_input,
+            outputs=[output_display, report_download],
             api_name="analyze"
         )
-        
-        msg_input.submit(
-            analyze,
-            inputs=[msg_input, chatbot, file_upload],
-            outputs=[chatbot, download_output]
-        )
-        
-        clear_btn.click(
-            clear_chat,
-            inputs=[],
-            outputs=[chatbot, download_output]
-        )
     
     return demo
 
 if __name__ == "__main__":
     try:
-        agent = init_agent()
-        demo = create_ui(agent)
-        
-        demo.queue(
-            api_open=False,
-            max_size=20
-        ).launch(
+        demo = create_interface()
+        demo.launch(
             server_name="0.0.0.0",
             server_port=7860,
-            show_error=True,
-            allowed_paths=[report_dir],
-            share=False
+            show_error=True
         )
     except Exception as e:
-        print(f"Failed to launch application: {str(e)}")
+        print(f"Error launching application: {str(e)}")
         sys.exit(1)