app.py

#%% md
# ## Nuru HIV Informational Chatbot
#%%
# Import libraries
import os
from dotenv import load_dotenv
import logging
import sys
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()

# Set seed for language detection consistency
DetectorFactory.seed = 0

# 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)

# 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"]
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()
    return any(re.search(r'\b' + re.escape(keyword) + r'\b', text) for keyword in keywords)

def contains_greeting_sw(text):
    return contains_exact_word_or_phrase(text, greeting_keywords_sw)

def contains_greeting_en(text):
    return contains_exact_word_or_phrase(text, greeting_keywords_en)

def contains_acknowledgment_sw(text):
    return contains_exact_word_or_phrase(text, acknowledgment_keywords_sw)

def contains_acknowledgment_en(text):
    return contains_exact_word_or_phrase(text, acknowledgment_keywords_en)

def contains_follow_up(text):
    return contains_exact_word_or_phrase(text, follow_up_keywords)

def convert_to_date(date_str):
    """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(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]):

    """Process user query, detect language, handle greetings, acknowledgments, and retrieve relevant information."""
    context = " ".join([item["user"] + " " + item["chatbot"] for item in conversation_history])
    user_params = json.loads(user_params)
    
    # Extract user information
    consent = user_params.get("CONSENT")
    person_info = user_params.get("PERSON_INFO", {})
    gender = person_info.get("GENDER", "")
    age = person_info.get("AGE", "")
    vl_result = person_info.get("VIRAL_LOAD", "")
    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", "")

    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))
    
    # 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":
        question = GoogleTranslator(source='sw', target='en').translate(question)
    
    # Retrieve relevant sources
    sources = retriever.retrieve(question)
    retrieved_text = "\n\n".join([f"Source {i+1}: {source.text}" for i, source in enumerate(sources[:3])])

    # Combine into new user question - conversation history, new question, retrieved sources
    question_final = (
        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"
        "- Only answers questions about HIV (Human Immunodeficiency Virus). Recognize that users may type 'HIV' with any capitalization (e.g., HIV, hiv, Hiv, etc.) or make minor typos (e.g., hvi, hiv/aids). Use your best judgment to understand when a user intends to refer to HIV. Politely correct any significant misunderstandings, but otherwise proceed to answer normally.\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=messages
    )

    # Collect response
    reply_to_user = completion.choices[0].message.content

    # add question and reply to conversation history
    conversation_history.append({"user": question, "chatbot": reply_to_user})  

    # If initial question was in swahili, translate response to swahili
    if lang_question=="sw":
        reply_to_user = GoogleTranslator(source='auto', target='sw').translate(reply_to_user) 

    # return system_prompt, conversation_history 
    return reply_to_user, conversation_history    

#%%
demo = gr.Interface(
    title = "Nishauri Chatbot Demo",
    fn=nishauri,
    inputs=["text", gr.State(value=[])],
    outputs=["text", gr.State()],
)

demo.launch()