Update app.py

Added RAG arch for verification

app.py

@@ -6,8 +6,53 @@ from pydantic import BaseModel, Field
 from PIL import Image
 import requests
 from datetime import datetime
+import pandas as pd
+from langchain.docstore.document import Document
+from langchain_community.vectorstores import Chroma
+from langchain_google_genai import GoogleGenerativeAIEmbeddings
 
-# 1. Enhanced Schema (Added NPI for the API call)
+# --- 1. RAG CONFIGURATION & INGESTION ---
+@st.cache_resource
+def get_vector_db(file_path):
+    """Ingests the 2026 Alert List and caches the vector database."""
+    if not os.path.exists(file_path):
+        st.error(f"Alert List file not found at {file_path}")
+        return None
+        
+    df = pd.read_csv(file_path)
+    df.columns = df.columns.str.strip()
+    
+    documents = []
+    for _, row in df.iterrows():
+        content = (
+            f"Provider Name: {row['Name']}\n"
+            f"License Number: {row['License Number']}\n"
+            f"Action Taken: {row['Action Type']}\n"
+            f"Effective Date: {row['Effective Date']}\n"
+            f"Summary: {row.get('Description', 'Administrative disciplinary action recorded.')}"
+        )
+        metadata = {
+            "license": str(row['License Number']),
+            "provider_name": row['Name']
+        }
+        documents.append(Document(page_content=content, metadata=metadata))
+
+    embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001")
+    # Using an in-memory Chroma for the demo; use persist_directory for production
+    vector_db = Chroma.from_documents(documents, embeddings)
+    return vector_db
+
+def check_yellow_flags(license_number, vector_db):
+    """Searches for the license in the ingested RAG database."""
+    # Perform similarity search
+    results = vector_db.similarity_search(license_number, k=1)
+    
+    # Check if the result is actually a match (similarity search can return 'close' matches)
+    if results and license_number in results[0].page_content:
+        return results[0].page_content
+    return None
+
+# --- 2. SCHEMAS & HELPERS ---
 class ProviderLicense(BaseModel):
     provider_name: str = Field(description="Full name of the healthcare provider")
     license_number: str = Field(description="The professional license number")
@@ -15,11 +60,6 @@ class ProviderLicense(BaseModel):
     state: str = Field(description="The state where the license was issued")
     expiration_date: str = Field(description="Format: YYYY-MM-DD")
 
-# 2. Page Configuration
-st.set_page_config(page_title="AI Credentialing Assistant", layout="wide")
-st.title("🩺 Provider Credentialing AI")
-
-# 3. Helper Function for NPPES
 def get_nppes_data(npi_number):
     url = "https://npiregistry.cms.hhs.gov/api/?version=2.1"
     params = {"number": npi_number}
@@ -30,14 +70,20 @@ def get_nppes_data(npi_number):
             return data["results"][0] if data.get("result_count", 0) > 0 else None
         return None
     except Exception as e:
-        st.error(f"NPPES API Error: {e}")
         return None
 
-# 4. Initialize Client
+# --- 3. PAGE SETUP & INITIALIZATION ---
+st.set_page_config(page_title="AI Credentialing Assistant", layout="wide")
+st.title("🩺 Provider Credentialing AI")
+
+# Load the RAG Database (Point to your uploaded CSV)
+alert_list_path = "alert-actions-2026.xlsx - Sheet1.csv"
+vdb = get_vector_db(alert_list_path)
+
 api_key = os.environ.get("GEMINI_API_KEY")
 client = genai.Client(api_key=api_key)
 
-# 5. Sidebar & Main Logic
+# --- 4. MAIN WORKFLOW ---
 uploaded_file = st.sidebar.file_uploader("Upload Medical License", type=["jpg", "jpeg", "png", "pdf"])
 
 if uploaded_file:
@@ -49,12 +95,12 @@ if uploaded_file:
         st.image(image, use_container_width=True)
 
     with col2:
-        st.subheader("AI Extraction & Registry Match")
-        with st.spinner("Analyzing and Verifying..."):
+        st.subheader("AI Extraction & Risk Analysis")
+        with st.spinner("Extracting & Verifying..."):
             try:
-                # AI Extraction
+                # A. Extraction
                 response = client.models.generate_content(
-                    model="gemini-2.5-flash", # Verify model ID in your environment
+                    model="gemini-2.0-flash", 
                     contents=["Extract details from this license.", image],
                     config=types.GenerateContentConfig(
                         response_mime_type="application/json",
@@ -63,43 +109,46 @@ if uploaded_file:
                 )
                 data = response.parsed
 
-                # Logic: Is Valid? (Calculated, not guessed)
+                # B. Expiration Logic
                 expiry = datetime.strptime(data.expiration_date, "%Y-%m-%d").date()
                 is_active = expiry >= datetime.today().date()
 
-                # Display Results
                 st.metric("Provider Name", data.provider_name)
                 st.write(f"**License:** {data.license_number} ({data.state})")
                 
                 status_color = "green" if is_active else "red"
                 st.markdown(f"**Status:** :{status_color}[{ 'Valid' if is_active else 'Expired'}] (Expires: {data.expiration_date})")
 
-                # --- NEW: Automated NPPES Verification ---
+                # C. Federal Verification
                 st.divider()
-                st.subheader("Federal Registry Verification")
+                st.subheader("Federal Registry (NPPES)")
                 registry_data = get_nppes_data(data.npi_number)
 
                 if registry_data:
-                    # Based on your st.json, registry_data is the single record dictionary
-                    basic_info = registry_data.get('basic', {})
-                    
-                    # Check if it's an Organization (NPI-2) or Individual (NPI-1)
-                    if registry_data.get('enumeration_type') == 'NPI-2':
-                        display_name = basic_info.get('organization_name', 'Unnamed Organization')
-                        st.info(f"Verified as Organizational NPI: {display_name}")
-                    else:
-                        f_name = basic_info.get('first_name', '')
-                        l_name = basic_info.get('last_name', '')
-                        display_name = f"{f_name} {l_name}".strip()
-                        st.success(f"Verified as Individual Provider: {display_name}")
+                    basic = registry_data.get('basic', {})
+                    name = basic.get('organization_name') if registry_data.get('enumeration_type') == 'NPI-2' else f"{basic.get('first_name')} {basic.get('last_name')}"
+                    st.success(f"NPI Verified: {name}")
+                else:
+                    st.warning("NPI not found in Federal Registry")
+
+                # D. NEW: RAG-based Yellow Flag Detection
+                st.divider()
+                st.subheader("⚠️ Risk Intelligence (RAG)")
                 
-                    st.write(f"**Registry Name:** {display_name}")
-                    
-                    # Taxonomy parsing (based on your JSON structure)
-                    taxonomies = registry_data.get('taxonomies', [])
-                    if taxonomies:
-                        st.write(f"**Primary Taxonomy:** {taxonomies[0].get('desc', 'N/A')}")
-                        st.write(f"**Associated License:** {taxonomies[0].get('license', 'N/A')}")
-            
+                flag_context = check_yellow_flags(data.license_number, vdb)
+                
+                if flag_context:
+                    st.error(f"YELLOW FLAG DETECTED for License {data.license_number}")
+                    # Use Gemini to summarize the disciplinary action for the user
+                    risk_summary = client.models.generate_content(
+                        model="gemini-2.0-flash",
+                        contents=[f"Based on this medical board record, summarize the risk in one sentence: {flag_context}"]
+                    )
+                    st.warning(risk_summary.text)
+                    with st.expander("View Raw Alert Detail"):
+                        st.text(flag_context)
+                else:
+                    st.success("No active flags found in the 2026 Medical Board Alert List.")
+
             except Exception as e:
                 st.error(f"Processing Error: {e}")