Create updated_diagnostics.py

updated_diagnostics.py

@@ -0,0 +1,243 @@
+"""
+Updated diagnostics module for the Pharmaceutical Analytics application.
+Uses the simplified Planning Agent implementation.
+"""
+
+import streamlit as st
+import os
+import json
+import traceback
+import time
+from datetime import datetime
+from typing import Dict, Any, Tuple
+
+def test_claude_connectivity() -> Tuple[bool, str]:
+    """
+    Test basic connectivity to Claude API
+    
+    Returns:
+        Tuple of (success_flag, response_or_error)
+    """
+    try:
+        # Check API key
+        api_key = os.getenv("ANTHROPIC_API_KEY")
+        if not api_key:
+            return False, "No ANTHROPIC_API_KEY found in environment variables"
+        
+        # Import anthropic
+        try:
+            import anthropic
+        except ImportError:
+            return False, "Failed to import anthropic library. Make sure it's installed (pip install anthropic)"
+        
+        # Create client and send request
+        client = anthropic.Anthropic(api_key=api_key)
+        
+        response = client.messages.create(
+            model="claude-3-haiku-20240307",
+            max_tokens=300,
+            messages=[
+                {"role": "user", "content": "Briefly tell me about Egyptian history in 2-3 sentences."}
+            ]
+        )
+        
+        if response and response.content:
+            return True, response.content[0].text
+        else:
+            return False, "Received empty response from Anthropic API"
+        
+    except Exception as e:
+        error_msg = f"Error testing Claude API: {str(e)}\n\n{traceback.format_exc()}"
+        return False, error_msg
+
+def test_simplified_agent() -> Tuple[bool, Dict[str, Any]]:
+    """
+    Test the simplified Planning Agent functionality
+    
+    Returns:
+        Tuple of (success_flag, results_dict)
+    """
+    results = {
+        "steps": [],
+        "errors": [],
+        "plan": None
+    }
+    
+    try:
+        # Test Planning Agent import
+        results["steps"].append({"step": "import", "status": "started"})
+        try:
+            from simplified_planning_agent import PlanningAgent
+            results["steps"].append({"step": "import", "status": "success"})
+        except Exception as import_error:
+            error_msg = f"Error importing simplified PlanningAgent: {str(import_error)}\n\n{traceback.format_exc()}"
+            results["errors"].append({"phase": "import", "error": error_msg})
+            return False, results
+        
+        # Test agent initialization
+        results["steps"].append({"step": "initialization", "status": "started"})
+        try:
+            planning_agent = PlanningAgent()
+            results["steps"].append({"step": "initialization", "status": "success"})
+        except Exception as init_error:
+            error_msg = f"Error initializing PlanningAgent: {str(init_error)}\n\n{traceback.format_exc()}"
+            results["errors"].append({"phase": "initialization", "error": error_msg})
+            return False, results
+        
+        # Test plan creation
+        results["steps"].append({"step": "plan_creation", "status": "started"})
+        try:
+            test_alert = "Sales of DrugX down 15% in Northeast region over past 30 days."
+            analysis_plan, plan_dict = planning_agent.create_analysis_plan(test_alert)
+            
+            results["steps"].append({"step": "plan_creation", "status": "success"})
+            results["plan"] = plan_dict
+            return True, results
+        except Exception as plan_error:
+            error_msg = f"Error creating analysis plan: {str(plan_error)}\n\n{traceback.format_exc()}"
+            results["errors"].append({"phase": "plan_creation", "error": error_msg})
+            return False, results
+            
+    except Exception as e:
+        error_msg = f"Unexpected error in agent functionality test: {str(e)}\n\n{traceback.format_exc()}"
+        results["errors"].append({"phase": "unknown", "error": error_msg})
+        return False, results
+
+def render_diagnostics_tab():
+    """
+    Render the diagnostics tab in the Streamlit app
+    """
+    st.header("🔧 System Diagnostics")
+    
+    # Environment information
+    st.subheader("Environment Information")
+    
+    # Check API key
+    api_key = os.getenv("ANTHROPIC_API_KEY")
+    api_status = "✅ Set" if api_key else "❌ Not Set"
+    
+    environment_data = {
+        "API Key": api_status,
+        "Python Version": os.sys.version,
+        "Timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+    }
+    
+    # Try to get library versions
+    try:
+        import anthropic
+        environment_data["Anthropic Library"] = anthropic.__version__
+    except (ImportError, AttributeError):
+        environment_data["Anthropic Library"] = "Not found"
+    
+    try:
+        import langgraph
+        environment_data["LangGraph Library"] = langgraph.__version__
+    except (ImportError, AttributeError):
+        environment_data["LangGraph Library"] = "Not found"
+    
+    # Display environment table
+    for key, value in environment_data.items():
+        cols = st.columns([1, 3])
+        with cols[0]:
+            st.markdown(f"**{key}:**")
+        with cols[1]:
+            st.markdown(f"{value}")
+    
+    # Claude connectivity test
+    st.subheader("Claude LLM Test")
+    if st.button("Test Claude Connectivity"):
+        with st.spinner("Testing connection to Claude..."):
+            success, response = test_claude_connectivity()
+        
+        if success:
+            st.success("✅ Claude API is working properly!")
+            st.markdown("**Response:**")
+            st.info(response)
+        else:
+            st.error("❌ Claude API test failed")
+            st.expander("Error Details", expanded=True).error(response)
+    
+    # Simplified Agent functionality test
+    st.subheader("Simplified Planning Agent Test")
+    if st.button("Test Simplified Planning Agent"):
+        with st.spinner("Testing Simplified Planning Agent..."):
+            success, results = test_simplified_agent()
+        
+        if success:
+            st.success("✅ Simplified Planning Agent is working properly!")
+            st.json(results["plan"])
+        else:
+            st.error("❌ Simplified Planning Agent test failed")
+            for error in results["errors"]:
+                st.expander(f"Error in {error['phase']} phase", expanded=True).error(error["error"])
+    
+    # Original Agent functionality test
+    st.subheader("Original Planning Agent Test (Likely to Fail)")
+    if st.button("Test Original Planning Agent"):
+        with st.spinner("Testing Original Planning Agent..."):
+            try:
+                from agents.planning_agent import PlanningAgent
+                planning_agent = PlanningAgent()
+                test_alert = "Sales of DrugX down 15% in Northeast region over past 30 days."
+                analysis_plan, plan_dict = planning_agent.create_analysis_plan(test_alert)
+                
+                st.success("✅ Original Planning Agent is working!")
+                st.json(plan_dict)
+            except Exception as e:
+                st.error("❌ Original Planning Agent test failed")
+                st.expander("Error Details", expanded=True).error(f"{str(e)}\n\n{traceback.format_exc()}")
+    
+    # Database Connectivity
+    st.subheader("Database Connectivity")
+    if st.button("Test Database Connection"):
+        with st.spinner("Testing database connection..."):
+            try:
+                import sqlite3
+                conn = sqlite3.connect("data/pharma_db.sqlite")
+                
+                # Test query
+                cursor = conn.cursor()
+                cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
+                tables = cursor.fetchall()
+                
+                # Close connection
+                conn.close()
+                
+                st.success(f"✅ Successfully connected to database")
+                st.write(f"Found {len(tables)} tables:")
+                st.json([table[0] for table in tables])
+                
+            except Exception as e:
+                st.error(f"❌ Database connection failed: {str(e)}")
+                st.expander("Error Details", expanded=True).error(traceback.format_exc())
+    
+    # Add guidance
+    st.markdown("---")
+    st.markdown("""
+    ### Using the Simplified Agent
+    
+    The original Planning Agent implementation has issues with the LangChain prompt templates. 
+    We've created a simplified version that uses direct API calls to Claude.
+    
+    To use the simplified agent:
+    
+    1. Copy the `simplified_planning_agent.py` file to your project
+    2. Update your workflow to import from `simplified_planning_agent` instead of `agents.planning_agent`
+    
+    ### Troubleshooting Tips
+    
+    1. If Claude test fails:
+       - Check your API key in environment variables
+       - Verify internet connectivity
+       - Confirm your API key has sufficient credits
+    
+    2. If Agent test fails:
+       - Check import errors (are all libraries installed?)
+       - Verify agent implementation
+       - Check for syntax errors in agent code
+    
+    3. If Database test fails:
+       - Verify database file exists
+       - Check file permissions
+       - Run setup_environment() function again
+    """)