Update app.py

app.py

@@ -1,102 +1,97 @@
+#%% md
+# ## Nuru HIV Informational Chatbot
+#%%
+# Import libraries
 import os
-os.environ["OPENAI_API_KEY"]
-
-from llama_index.llms.openai import OpenAI
-from llama_index.core.schema import MetadataMode
-import openai
-from openai import OpenAI as OpenAIOG
+from dotenv import load_dotenv
 import logging
 import sys
-llm = OpenAI(temperature=0.0, model="gpt-3.5-turbo")
+import re
+import json
+from datetime import datetime
+from langdetect import detect, DetectorFactory
+from deep_translator import GoogleTranslator
+from lingua import Language, LanguageDetectorBuilder
+import gradio as gr
+from openai import OpenAI as OpenAIOG
+from llama_index.llms.openai import OpenAI
+from llama_index.core import VectorStoreIndex, StorageContext, load_index_from_storage
+from deep_translator import GoogleTranslator
+
+# Set OpenAI API Key (Ensure this is set in the environment)
+# load_dotenv("config.env")
+os.environ.get("OPENAI_API_KEY")
+
+# Initialize OpenAI clients
+llm = OpenAI(temperature=0.0, model="gpt-4o")
 client = OpenAIOG()
 
-from langdetect import detect
-from langdetect import DetectorFactory
+# Set seed for language detection consistency
 DetectorFactory.seed = 0
-from deep_translator import GoogleTranslator
-from lingua import Language, LanguageDetectorBuilder
 
-# Load index
-from llama_index.core import VectorStoreIndex
-from llama_index.core import StorageContext
-from llama_index.core import load_index_from_storage
+# Load index for retrieval
 storage_context = StorageContext.from_defaults(persist_dir="arv_metadata")
 index = load_index_from_storage(storage_context)
 query_engine = index.as_query_engine(similarity_top_k=3, llm=llm)
-retriever = index.as_retriever(similarity_top_k = 3)
-
-import gradio as gr
-import re
-import json
-from datetime import datetime
+retriever = index.as_retriever(similarity_top_k=3)
 
-acknowledgment_keywords_sw = ["sawa", "ndiyo", "naam", "hakika", "asante", "nimeelewa", "nimekupata", "ni kweli",
-                             "kwa hakika", "nimesikia", "ahsante"]
+# Define keyword lists
+acknowledgment_keywords_sw = ["sawa", "ndiyo", "naam", "hakika", "asante", "nimeelewa", "nimekupata", "ni kweli", "kwa hakika", "nimesikia", "ahsante"]
 acknowledgment_keywords_en = ["thanks", "thank you", "thx", "ok", "okay", "great", "got it", "appreciate", "good", "makes sense"]
-follow_up_keywords = ["but", "also", "and", "what", "how", "why", "when", "is", "?",
-                     "lakini", "pia", "na", "nini", "vipi", "kwanini", "wakati"]
+follow_up_keywords = ["but", "also", "and", "what", "how", "why", "when", "is", "?", "lakini", "pia", "na", "nini", "vipi", "kwanini", "wakati"]
 greeting_keywords_sw = ["sasa", "niaje", "habari", "mambo", "jambo", "shikamoo", "marahaba", "hujambo", "hamjambo", "salama", "vipi"]
 greeting_keywords_en = ["hi", "hello", "hey", "how's it", "what's up", "yo", "howdy"]
+#%%
+# Define helper functions
 
 def contains_exact_word_or_phrase(text, keywords):
+    """Check if the given text contains any exact keyword from the list."""
     text = text.lower()
-    for keyword in keywords:
-        if re.search(r'\b' + re.escape(keyword) + r'\b', text):
-            return True
-    return False
+    return any(re.search(r'\b' + re.escape(keyword) + r'\b', text) for keyword in keywords)
 
-def contains_greeting_sw(question):
-    # Check if the question contains acknowledgment keywords
-    return contains_exact_word_or_phrase(question, greeting_keywords_sw)
+def contains_greeting_sw(text):
+    return contains_exact_word_or_phrase(text, greeting_keywords_sw)
 
-def contains_greeting_en(question):
-    # Check if the question contains acknowledgment keywords
-    return contains_exact_word_or_phrase(question, greeting_keywords_en)
+def contains_greeting_en(text):
+    return contains_exact_word_or_phrase(text, greeting_keywords_en)
 
-def contains_acknowledgment_sw(question):
-    # Check if the question contains acknowledgment keywords
-    return contains_exact_word_or_phrase(question, acknowledgment_keywords_sw)
+def contains_acknowledgment_sw(text):
+    return contains_exact_word_or_phrase(text, acknowledgment_keywords_sw)
 
-def contains_acknowledgment_en(question):
-    # Check if the question contains acknowledgment keywords
-    return contains_exact_word_or_phrase(question, acknowledgment_keywords_en)
+def contains_acknowledgment_en(text):
+    return contains_exact_word_or_phrase(text, acknowledgment_keywords_en)
 
-def contains_follow_up(question):
-    # Check if the question contains follow-up indicators
-    return contains_exact_word_or_phrase(question, follow_up_keywords)
+def contains_follow_up(text):
+    return contains_exact_word_or_phrase(text, follow_up_keywords)
 
 def convert_to_date(date_str):
-    date_obj = datetime.strptime(date_str, "%Y%m%d")
-    return date_obj.strftime("%Y-%m-%d")
-
-def detect_language(question):
-    # Check if the text has less than 5 words
-    if len(question.split()) < 5:
-        languages = [Language.ENGLISH, Language.SWAHILI]  # Add more languages as needed
+    """Convert date string in YYYYMMDD format to YYYY-MM-DD."""
+    try:
+        return datetime.strptime(date_str, "%Y%m%d").strftime("%Y-%m-%d")
+    except ValueError:
+        return "Unknown Date"
+
+def detect_language(text):
+    """Detect language of a given text using Lingua for short texts and langdetect for longer ones."""
+    if len(text.split()) < 5:
+        languages = [Language.ENGLISH, Language.SWAHILI]
         detector = LanguageDetectorBuilder.from_languages(*languages).build()
-        detected_language = detector.detect_language_of(question)
-        # Return language code for consistency
-        if detected_language == Language.SWAHILI:
-            return "sw"
-        elif detected_language == Language.ENGLISH:
-            return "en"
-    else:
-        try:
-            lang_detect = detect(question)
-            return lang_detect  
-        except Exception as e:
-            print(f"Error with langdetect: {e}")
-            return "unknown"  
-
+        detected_language = detector.detect_language_of(text)
+        return "sw" if detected_language == Language.SWAHILI else "en"
+    try:
+        return detect(text)
+    except Exception as e:
+        logging.warning(f"Language detection error: {e}")
+        return "unknown"
+#%%
+# Define Gradio function
 def nishauri(user_params: str, conversation_history: list[str]):
 
-    # Get conversation history
+    """Process user query, detect language, handle greetings, acknowledgments, and retrieve relevant information."""
     context = " ".join([item["user"] + " " + item["chatbot"] for item in conversation_history])
-    
-    # Convert the user_params_str to a dictionary
     user_params = json.loads(user_params)
     
-    ## Parse user params
+    # Extract user information
     consent = user_params.get("CONSENT")
     person_info = user_params.get("PERSON_INFO", {})
     gender = person_info.get("GENDER", "")
@@ -105,160 +100,94 @@ def nishauri(user_params: str, conversation_history: list[str]):
     vl_date = convert_to_date(person_info.get("VIRAL_LOAD_DATETIME", ""))
     next_appt_date = convert_to_date(person_info.get("APPOINTMENT_DATETIME", ""))
     regimen = person_info.get("REGIMEN", "")
-    question = user_params.get("QUESTION")
-
-        # Keep those user params that are present
-    info_pieces = []
-
-    if gender:
-        info_pieces.append(f"The person is {gender}.")
-    
-    if age:
-        info_pieces.append(f"The person is age {age}.")
-    
-    if next_appt_date:
-        info_pieces.append(f"The person's next clinical check-in is scheduled for {next_appt_date}.")
-    
-    if regimen:
-        info_pieces.append(f"The person is on the following regimen for HIV: {regimen}.")
-    
-    if vl_result:
-        info_pieces.append(f"The person's most recent viral load result was {vl_result}.")
-    
-    if vl_date:
-        info_pieces.append(f"The person's most recent viral load was taken on {vl_date}.")
-    
-    full_text = " ".join(info_pieces)
-    
-    ## Process greeting
-    # greet_response = process_greeting_response(question)
-    if contains_greeting_en(question) and not contains_follow_up(question):
-            greeting = (
-                f" The user previously asked and answered the following: {context}. "
-                f" The user just provided the following greeting: {question}. "
-                "Please respond accordingly in English."
-            )
-            completion = client.chat.completions.create(
-              model="gpt-4o",
-                messages=[
-                {"role": "user", "content": greeting}
-              ]
-            )
-            reply_to_user = completion.choices[0].message.content
-            conversation_history.append({"user": question, "chatbot": reply_to_user})   
-            return reply_to_user, conversation_history
-
-    if contains_greeting_sw(question) and not contains_follow_up(question):
-            greeting = (
-                f" The user previously asked and answered the following: {context}. "
-                f" The user just provided the following greeting: {question}. "
-                "Please respond accordingly in Swahili."
-            )
-            completion = client.chat.completions.create(
-              model="gpt-4o",
-                messages=[
-                {"role": "user", "content": greeting}
-              ]
-            )
-            reply_to_user = completion.choices[0].message.content
-            conversation_history.append({"user": question, "chatbot": reply_to_user})   
-            return reply_to_user, conversation_history
-    
-    ## Process acknowledgment
-    if contains_acknowledgment_en(question) and not contains_follow_up(question):
-            acknowledgment = (
-                f" The user previously asked and answered the following: {context}. "
-                f" The user just provided the following acknowledgement: {question}. "
-                "Please respond accordingly in English."
-            )
-            completion = client.chat.completions.create(
-              model="gpt-4o",
-                messages=[
-                {"role": "user", "content": acknowledgment}
-              ]
-            )
-            reply_to_user = completion.choices[0].message.content
-            conversation_history.append({"user": question, "chatbot": reply_to_user})   
-            return reply_to_user, conversation_history
-
-    if contains_acknowledgment_sw(question) and not contains_follow_up(question):
-            acknowledgment = (
-                f" The user previously asked and answered the following: {context}. "
-                f" The user just provided the following acknowledgment: {question}. "
-                "Please respond accordingly in Swahili."
-            )
-            completion = client.chat.completions.create(
-              model="gpt-4o",
-                messages=[
-                {"role": "user", "content": acknowledgment}
-              ]
-            )
-            reply_to_user = completion.choices[0].message.content
-            conversation_history.append({"user": question, "chatbot": reply_to_user})   
-            return reply_to_user, conversation_history
-    
-    # context = " ".join([item["user"] + " " + item["chatbot"] for item in conversation_history])
-
-    ## If not greeting or acknowledgement, then proceed with RAG
+    question = user_params.get("QUESTION", "")
+
+    info_pieces = [
+        "Here is information about the person asking the question."
+        f"The person is {gender}." if gender else "",
+        f"The person is age {age}." if age else "",
+        f"The person's next clinical check-in is scheduled for {next_appt_date}." if next_appt_date else "",
+        f"The person is on the following regimen for HIV: {regimen}." if regimen else "",
+        f"The person's most recent viral load result was {vl_result}." if vl_result else "",
+        f"The person's most recent viral load was taken on {vl_date}." if vl_date else "",
+    ]
+    full_text = " ".join(filter(None, info_pieces))
     
-    ## Detect language of question - if Swahili, translate to English
+    # Process greetings and acknowledgments
+    for lang, contains_greeting, contains_acknowledgment in [("en", contains_greeting_en, contains_acknowledgment_en), ("sw", contains_greeting_sw, contains_acknowledgment_sw)]:
+        if contains_greeting(question) and not contains_follow_up(question):
+            prompt = f"The user said: {question}. Respond accordingly in {lang}."
+        elif contains_acknowledgment(question) and not contains_follow_up(question):
+            prompt = f"The user acknowledged: {question}. Respond accordingly in {lang}."
+        else:
+            continue
+        completion = client.chat.completions.create(
+            model="gpt-4o",
+            messages=[{"role": "user", "content": prompt}]
+        )
+        reply_to_user = completion.choices[0].message.content
+        conversation_history.append({"user": question, "chatbot": reply_to_user})
+        return reply_to_user, conversation_history
+
+    # Detect language and translate if needed
     lang_question = detect_language(question)
-    if lang_question=="sw":
+    if lang_question == "sw":
         question = GoogleTranslator(source='sw', target='en').translate(question)
     
-    # Retrieve sources
+    # Retrieve relevant sources
     sources = retriever.retrieve(question)
-    source0 = sources[0].text
-    source1 = sources[1].text
-    source2 = sources[2].text
-
-    # If user consented, add user parameters, otherwise proceed with out
-    if consent == "YES":
-        background = ("The person who asked the question is a person living with HIV."
-                      f" The person is {info_pieces} "
-                      " They are asking questions about HIV. Do not talk about anything that is not related to HIV. "
-                      " Recognize that they already have HIV and do not suggest that they have to get tested"
-                      " for HIV or take post-exposure prophylaxis, as that is not relevant, though their partners perhaps should."  
-                      " Do not suggest anything that is not relevant to someone who already has HIV."
-                      " Do not mention in the response that the person is living with HIV."
-                      " The following information about viral loads is authoritative for any question about viral loads:"
-                      " A high viral load or non-suppressed viral load is any viral load above 200 copies/ml."
-                      " A viral load above 1000 copies/ml suggests treatment failure."
-                      " A suppressed viral load is one below 200 copies / ml.")
-    else:
-        background = ("The person who asked the question is a person living with HIV."
-                      " They are asking questions about HIV. Do not talk about anything that is not related to HIV. "
-                      " Recognize that they already have HIV and do not suggest that they have to get tested"
-                      " for HIV or take post-exposure prophylaxis, as that is not relevant, though their partners perhaps should."  
-                      " Do not suggest anything that is not relevant to someone who already has HIV."
-                      " Do not mention in the response that the person is living with HIV."
-                      " The following information about viral loads is authoritative for any question about viral loads:"
-                      " A high viral load or non-suppressed viral load is any viral load above 200 copies/ml."
-                      " A viral load above 1000 copies/ml suggests treatment failure."
-                      " A suppressed viral load is one below 200 copies / ml.")
+    retrieved_text = "\n\n".join([f"Source {i+1}: {source.text}" for i, source in enumerate(sources[:3])])
 
-    # Combine into final prompt - user background, conversation history, new question, retrieved sources
+    # Combine into new user question - conversation history, new question, retrieved sources
     question_final = (
-        f" The user previously asked and answered the following: {context}. "
-        f" The user just asked the following question: {question}."
-        f" Please use the following content to generate a response: {source0} {source1} {source2}."
-        f" Please consider the following background information when generating a response: {background}."
-        " Keep answers brief and limited to the question that was asked."
-        " If they share a greeting, just greet them in return and ask if they have a question."
-        " Do not change the subject or address anything the user didn't directly ask about."
-        " If they respond with an acknowledgement, simply thank them."
-        " Do not discuss anything other than HIV. If they ask a question that is not about HIV, respond that"
-        " you are only able to discuss HIV."
-        " Keep the response to under 50 words and use simple language. The person asking the question does not know technical terms."
+        f"The user asked the following question: \"{question}\"\n\n"
+        f"Use only the content below to answer the question:\n\n{retrieved_text}\n\n"
+        "Guidelines:\n"
+        "- Only answer the question that was asked.\n"
+        "- Do not change the subject or include unrelated information.\n"
+        "- Only discuss HIV. If the question is not about HIV, say that you can only answer HIV-related questions.\n"
+    )
+
+    # Set LLM instructions. If user consented, add user parameters, otherwise proceed without
+    system_prompt = (
+        "You are a helpful assistant who only answers questions about HIV.\n"
+        "- Do not answer questions about other topics (e.g., malaria or tuberculosis).\n"
+        "- If a question is unrelated to HIV, politely respond that you can only answer HIV-related questions.\n\n"
+    
+        "The person asking the question is living with HIV.\n"
+        "- Do not suggest they get tested for HIV or take post-exposure prophylaxis (PEP).\n"
+        "- You may mention that their partners might benefit from testing or PEP, if relevant.\n"
+        "- Do not mention in your response that the person is living with HIV.\n"
+        "- Only suggest things relevant to someone who already has HIV.\n\n"
+        "- Keep the answer under 50 words.\n"
+        "- Use simple, easy-to-understand language. Avoid medical jargon.\n"
+    
+        "Use the following authoritative information about viral loads:\n"
+        "- A high or non-suppressed viral load is above 200 copies/ml.\n"
+        "- A viral load above 1000 copies/ml suggests treatment failure.\n"
+        "- A suppressed viral load is one below 200 copies/ml.\n\n"
     )
 
+    if consent == "YES":
+        system_prompt = f"{system_prompt} {full_text}."
+ 
+    # Start with context
+    messages = [{"role": "system", "content": system_prompt}]
+
+    # Add conversation history
+    for turn in conversation_history:
+        messages.append({"role": "user", "content": turn["user"]})
+        messages.append({"role": "assistant", "content": turn["chatbot"]})
+    
+    # Finally, add the current question
+    messages.append({"role": "user", "content": question_final})
+
     # Generate response
     completion = client.chat.completions.create(
-      model="gpt-4o",
-        messages=[
-        {"role": "user", "content": question_final}
-      ]
+        model="gpt-4o",
+        messages=messages
     )
+
     # Collect response
     reply_to_user = completion.choices[0].message.content
 
@@ -269,8 +198,10 @@ def nishauri(user_params: str, conversation_history: list[str]):
     if lang_question=="sw":
         reply_to_user = GoogleTranslator(source='auto', target='sw').translate(reply_to_user) 
 
-    return reply_to_user, conversation_history
+    # return system_prompt, conversation_history 
+    return reply_to_user, conversation_history    
 
+#%%
 demo = gr.Interface(
     title = "Nishauri Chatbot Demo",
     fn=nishauri,