#%% md # ## Nuru HIV Informational Chatbot #%% # Import libraries import os import json import logging from datetime import datetime 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 StorageContext, load_index_from_storage from llama_index.core.postprocessor.llm_rerank import LLMRerank from deep_translator import GoogleTranslator from dotenv import load_dotenv import utils.helpers as helpers # Load environment variables from .env file load_dotenv() # 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() # Load index for retrieval storage_context = StorageContext.from_defaults(persist_dir="arv_metadata") index = load_index_from_storage(storage_context) # retriever = index.as_retriever(similarity_top_k=10, # # Similarity threshold for filtering # similarity_threshold=0.5) #%% # 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]) # formatted_history = convert_conversation_format(conversation_history) # summary = summarize_conversation(formatted_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 = helpers.convert_to_date(person_info.get("VIRAL_LOAD_DATETIME", ""), datetime) next_appt_date = helpers.convert_to_date(person_info.get("APPOINTMENT_DATETIME", ""), 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)) # detect language of user lang_question = helpers.detect_language(question, Language, LanguageDetectorBuilder, client, logging) print(lang_question) # If user is making a greeting or acknowledgement, address that accordingly intent = helpers.detect_intention(question, client = client) if intent == "greeting": prompt = f""" The user greeted you as follows: {question}. Respond by asking if they have any questions about HIV. Respond in {"Swahili" if lang_question == "sw" else "English"}. """ elif intent == "acknowledgment": prompt = f""" The user acknowledged a response you gave to a prior question as follows {question}. Respond by saying you are ready to help if they have any more questions. Respond in {"Swahili" if lang_question == "sw" else "English"}. """ else: prompt = None if prompt: completion = client.chat.completions.create( model="gpt-3.5-turbo", 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 # If the user is asking a question, proceed with the RAG pipeline # Translate if needed if lang_question == "sw": question = GoogleTranslator(source='sw', target='en').translate(question) # Retrieve relevant sources # sources = retriever.retrieve(question) # Summarize the conversation history history_summary = " ".join( [f"User: {turn['user']} Assistant: {turn['chatbot']}" for turn in conversation_history] ) query_with_context = f"Current question: {question}\n\nSummary of prior context: {history_summary}" # Initialize the LLMRerank postprocessor reranker = LLMRerank(top_n=3) # Attach the reranker to the retriever retriever_with_rerank = index.as_retriever( similarity_top_k=10, similarity_threshold=0.6, postprocessors=[reranker] ) # Retrieve and re-rank sources sources = retriever_with_rerank.retrieve(query_with_context) # Combine the top-ranked sources retrieved_text = "\n\n".join([f"Source {i+1}: {source.text}" for i, source in enumerate(sources)]) # 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" "- 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).\n" "- Recognize that users may type 'HIV' with any capitalization (e.g., HIV, hiv, Hiv, etc.) or make minor typos (e.g., hvi, hiv/aids).\n" "- If a question is ambiguous or might be indirectly related to HIV (e.g., symptoms, illness, or general health concerns), assume it could be relevant to HIV and respond accordingly.\n" "- If a question is about using the Nishauri app, such as finding viral load results, regimen details, or the next appointment, provide clear instructions on how to navigate the app to find this information.\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" "- The user may user lowercase or slang for HIV or related terms.\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 #%% app = gr.Interface( title = "Nuru Chatbot Demo", description="Enter a question and see the processed outputs in collapsible boxes", fn=nishauri, inputs=["text", gr.State(value=[])], outputs=[ gr.Textbox(label = "Nuru Response", type = "text"), gr.State() ], ) # for testing purposes to run as script if __name__ == "__main__": app.launch()