Upload app.py

app.py

@@ -17,6 +17,7 @@ import torch
 import faiss
 import numpy as np
 import gradio as gr
+from sklearn.metrics.pairwise import cosine_similarity
 # from google.colab import drive
 from transformers import pipeline, AutoModelForCausalLM, AutoTokenizer
 from sentence_transformers import SentenceTransformer
@@ -39,7 +40,10 @@ whisper_model = whisper.load_model("base")
 
 load_dotenv()
 
+
+
 hf_token = os.getenv("HF_TOKEN")
+
 login(token=hf_token)
 
 
@@ -50,7 +54,7 @@ login(token=hf_token)
 # 🔧 Configuration
 # -------------------------------
 base_model_path = "google/gemma-2-9b-it"
-peft_model_path = "Jaamie/gemma-mental-health-qlora"
+#peft_model_path = "Jaamie/gemma-mental-health-qlora"
 device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 
 embedding_model_bge = "BAAI/bge-base-en-v1.5"
@@ -205,6 +209,30 @@ def extract_diagnosis(response_text: str) -> str:
             return line.split(":")[-1].strip()
     return "Unknown"
 
+# calculating the correctness of the answer - Hallucination
+def calculate_rag_confidence(query_embedding, top_k_docs_embeddings, generation_logprobs=None):
+    """
+    Combines retriever and generation signals to compute a confidence score.
+    Args:
+        query_embedding (np.ndarray): Embedding vector of the user query (shape: [1, dim]).
+        top_k_docs_embeddings (np.ndarray): Embedding matrix of top-k retrieved documents (shape: [k, dim]).
+        generation_logprobs (list, optional): List of logprobs for generated tokens.
+    Returns:
+        float: Final confidence score (0 to 1).
+    """
+    retriever_similarities = cosine_similarity(query_embedding, top_k_docs_embeddings)
+    retriever_confidence = float(np.max(retriever_similarities))  
+
+    if generation_logprobs:
+        gen_confidence = float(np.exp(np.mean(generation_logprobs)))
+    else:
+        gen_confidence = 0.0  # fallback if unavailable
+
+    alpha, beta = 0.6, 0.4
+    final_confidence = alpha * retriever_confidence + beta * gen_confidence
+    return round(final_confidence, 4)
+
+# Main Process
 def process_query(user_query, input_type="text"):
     # Embed the query
     query_embedding = embedding_model.encode(user_query, normalize_embeddings=True)
@@ -223,6 +251,7 @@ def process_query(user_query, input_type="text"):
     print(f"Detected emotion: {emotion_result}")
     emotion = emotion_result['label']
     value = emotion_result['score']
+
     # Define RAG prompt
     prompt_in_chat_format = [
         {"role": "user", "content": f"""
@@ -268,17 +297,24 @@ def process_query(user_query, input_type="text"):
         print("❌ Error during generation:", e)
         answer = "⚠️ An error occurred while generating the response."
 
-    # Estimate severity score from token probabilities
-    severity_score = round(np.random.uniform(0.6, 1.0), 2)
-    answer += f"\n\n🧭 Confidence Score: {value}"
-    answer += f"\n\n*Confidence Score is the correctness of the answer"
+    # Get embeddings of retrieved docs
+    retrieved_doc_embeddings = embedding_model.encode(retrieved_docs, normalize_embeddings=True)
+    retrieved_doc_embeddings = np.array(retrieved_doc_embeddings, dtype=np.float32)
+
+    # Calculate RAG-based confidence
+    confidence_score = calculate_rag_confidence(query_embedding, retrieved_doc_embeddings)
+
+    # Add to response
+    answer += f"\n\n🧭 Accuracy & Closeness of the Answer: {confidence_score:.2f}"
+    answer += "\n\n*Derived from semantic similarity and generation certainty."
+
 
     # Extracting diagnosis
     diagnosis = extract_diagnosis(answer)
     status = "fallback" if diagnosis.lower() == "unknown" else "success"
 
     # Log interaction
-    log_query(input_type=input_type, query=user_query, diagnosis=diagnosis, confidence_score=severity_score, status=status)
+    log_query(input_type=input_type, query=user_query, diagnosis=diagnosis, confidence_score=confidence_score, status=status)
     download_path = create_summary_txt(answer)
 
     timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@@ -371,27 +407,27 @@ def show_logs():
         return f"⚠️ Error: {e}"
 
 
-def create_summary_pdf(text, filename_prefix="diagnosis_report"):
-    try:
-        filename = f"{filename_prefix}_{uuid.uuid4().hex[:6]}.pdf"
-        filepath = os.path.join(".", filename)  # Save in current directory
-        pdf = FPDF()
-        pdf.add_page()
-        pdf.set_font("Arial", style='B', size=14)
-        pdf.cell(200, 10, txt="🧠 Mental Health Diagnosis Report", ln=True, align='C')
-        pdf.set_font("Arial", size=12)
-        pdf.ln(10)
-
-        wrapped = textwrap.wrap(text, width=90)
-        for line in wrapped:
-            pdf.cell(200, 10, txt=line, ln=True)
-
-        pdf.output(filepath)
-        print(f"✅ PDF created at: {filepath}")
-        return filepath
-    except Exception as e:
-        print(f"❌ Error creating PDF: {e}")
-        return None
+# def create_summary_pdf(text, filename_prefix="diagnosis_report"):
+#     try:
+#         filename = f"{filename_prefix}_{uuid.uuid4().hex[:6]}.pdf"
+#         filepath = os.path.join(".", filename)  # Save in current directory
+#         pdf = FPDF()
+#         pdf.add_page()
+#         pdf.set_font("Arial", style='B', size=14)
+#         pdf.cell(200, 10, txt="🧠 Mental Health Diagnosis Report", ln=True, align='C')
+#         pdf.set_font("Arial", size=12)
+#         pdf.ln(10)
+
+#         wrapped = textwrap.wrap(text, width=90)
+#         for line in wrapped:
+#             pdf.cell(200, 10, txt=line, ln=True)
+
+#         pdf.output(filepath)
+#         print(f"✅ PDF created at: {filepath}")
+#         return filepath
+#     except Exception as e:
+#         print(f"❌ Error creating PDF: {e}")
+#         return None
 
 
 
@@ -491,22 +527,12 @@ def unified_handler(audio, text):
     else:
         response, _ = process_query(text, input_type="text")
 
-    download_path = create_summary_txt(response)  # <- save as txt instead
+    download_path = create_summary_txt(response)  
 
     return response, download_path
 
-#Agentic Framework from HF spaces
-# agent_iframe = gr.HTML(
-#     '<iframe src="https://jaamie-mental-health-agent.hf.space" width="100%" height="700px" style="border:none;"></iframe>'
-# )
 
 
-    # if email:
-    #     send_status = send_email_report(to_email=email, response=response)
-    #     response += f"\n\n{send_status}"
-
-    # return response, download_path
-
 
 # Gradio UI
 
@@ -540,23 +566,6 @@ logs_tab = gr.Interface(
 )
 
 
-# 📝 Anonymous Feedback
-# feedback_tab = gr.Interface(
-#     fn=lambda fb, inp_type, query, diag, score, status: submit_feedback(fb, inp_type, query, diag, score, status),
-#     inputs=[
-#         gr.Textbox(label="📝 Feedback"),
-#         gr.Textbox(label="Input Type"),
-#         gr.Textbox(label="Query"),
-#         gr.Textbox(label="Diagnosis"),
-#         gr.Textbox(label="Confidence Score"),
-#         gr.Textbox(label="Status")
-#     ],
-#     outputs="text",
-#     title="📝 Submit Feedback With Session Metadata"
-# )
-
-# def feedback_handler(fb, inp_type, query, diag, score, status):
-#     return submit_feedback(fb, inp_type, query, diag, score, status)
 
 feedback_tab = gr.Interface(
     fn=submit_feedback,