import os import pandas as pd import numpy as np import gzip from sentence_transformers import SentenceTransformer from sklearn.metrics.pairwise import cosine_similarity import gradio as gr from together import Together # Load pre-trained Sentence Transformer model model = SentenceTransformer('LaBSE') # Load questions and answers from the CSV file df = pd.read_csv('combined_questions_and_answers.csv') # Encode all questions in the dataset question_embeddings = model.encode(df['Question'].tolist()) # Together API setup client = Together(api_key=os.environ.get("TOGETHER_API_KEY")) def compressed_length(s): return len(gzip.compress(s.encode('utf-8'))) def ncd(x, y): """ Normalized Compression Distance for strings x and y. """ Cx = compressed_length(x) Cy = compressed_length(y) Cxy = compressed_length(x + " " + y) return (Cxy - min(Cx, Cy)) / max(Cx, Cy) def normalize_scores(scores, reverse=False): """ Scale a list of scores to [0,1], optionally reversing (1 - x). """ min_score = min(scores) max_score = max(scores) if max_score == min_score: return [0] * len(scores) if reverse: return [(max_score - x) / (max_score - min_score) for x in scores] return [(x - min_score) / (max_score - min_score) for x in scores] def hybrid_retrieval(query, passages, embeddings, alpha=0.7, beta=0.3): """ Combine cosine similarity (SentenceTransformer) and Normalized Compression Distance (NCD) for retrieval. """ query_embedding = model.encode(query) cosine_similarities = cosine_similarity([query_embedding], embeddings)[0] # Normalize normalized_cosine_similarities = normalize_scores(cosine_similarities) # Calculate NCD ncd_values = [ncd(query, passage) for passage in passages] normalized_ncd_values = normalize_scores(ncd_values, reverse=True) # Combine final_scores = [ alpha * cos_sim + beta * ncd_sim for cos_sim, ncd_sim in zip(normalized_cosine_similarities, normalized_ncd_values) ] most_similar_index = np.argmax(final_scores) return most_similar_index, cosine_similarities[most_similar_index], ncd_values[most_similar_index], final_scores[most_similar_index] def llama_query(prompt, system_content): """ Send a prompt to the Together LLaMa model and return the response. """ response = client.chat.completions.create( model="meta-llama/Llama-3.3-70B-Instruct-Turbo", messages=[ {"role": "system", "content": system_content}, {"role": "user", "content": prompt} ], max_tokens=512, temperature=0.7, top_p=0.7, top_k=50, repetition_penalty=1, stop=["<|eot_id|>", "<|eom_id|>"], stream=False ) return response.choices[0].message.content def check_blood_donation_relevance(question): """ Use LLaMa to check whether 'question' is about blood donation. """ prompt = f"Is the following question related to blood donation? Answer ONLY with 'Yes' or 'No': {question}" system_content = "You are an assistant that determines if a question is related to blood donation." response = llama_query(prompt, system_content).strip().lower() return response == 'yes' def detect_language(text): """ Use LLaMa to detect language (English or Swahili). Returns 'swahili' or 'english'. """ prompt = ( "Detect the language of this text. If it's Swahili, return 'Swahili'. " "If it's English, return 'English'. Here's the text:\n\n" f"{text}" ) system_content = "You are a language detection assistant." response = llama_query(prompt, system_content).strip().lower() # Attempt to match strictly 'swahili' or 'english' from the response if "swahili" in response: return "swahili" if "english" in response: return "english" # Fallback: default to English return "english" def translate_to_english(text): """ Translate Swahili text to English using LLaMa. """ prompt = f"Translate the following Swahili text to English: {text}" system_content = "You are a translation assistant that translates from Swahili to English." response = llama_query(prompt, system_content) return response.strip() def translate_to_swahili(text): """ Translate any text to simple Swahili using LLaMa. """ prompt = f"Translate the following text to simple Swahili, avoiding difficult words: {text}" system_content = "You are a translation assistant that translates to simple Swahili." response = llama_query(prompt, system_content) return response.strip() def refine_answer(question, retrieved_answer): """ Refine the retrieved answer, making it more relevant and natural. """ prompt = ( f"Question: {question}\n\n" f"Retrieved Answer: {retrieved_answer}\n\n" "Please refine the retrieved answer so it's direct, clear, and specifically addresses the question." ) system_content = "You are an assistant that refines answers to make them more relevant and natural." return llama_query(prompt, system_content).strip() def get_answer(user_question, threshold=0.3): # 1) Detect user language language = detect_language(user_question) # 2) Convert user question to English for checking & retrieval if language == 'swahili': english_question = translate_to_english(user_question) else: english_question = user_question # 3) Check if the question is about blood donation using LLaMa is_blood_related = check_blood_donation_relevance(english_question) if not is_blood_related: # Off-topic response off_topic_message = "I'm sorry, but your question doesn't seem to be related to blood donation. Could you please ask a question about blood donation?" if language == 'swahili': off_topic_message = translate_to_swahili(off_topic_message) return off_topic_message, 0.0 # If it is about blood donation, proceed with hybrid retrieval index, cosine_sim, ncd_value, final_score = hybrid_retrieval( english_question, df['Question'].tolist(), question_embeddings ) # 4) If retrieval confidence is high enough, refine the CSV answer if final_score > threshold: retrieved_answer = df.iloc[index]['Answer'] refined_answer_english = refine_answer(english_question, retrieved_answer) # Translate back to user language if needed if language == 'swahili': return translate_to_swahili(refined_answer_english), final_score else: return refined_answer_english, final_score else: # 5) If retrieval is below threshold, ask LLaMa for a general blood-donation-related answer llama_response_english = llama_query( f"Please provide a concise, accurate answer about blood donation for the question: {english_question}", "You are an assistant knowledgeable about blood donation. Provide concise, accurate answers." ) llama_response_english = llama_response_english.strip() # Translate back to user language if needed if language == 'swahili': return translate_to_swahili(llama_response_english), final_score else: return llama_response_english, final_score # Gradio app def gradio_app(user_question): answer, similarity = get_answer(user_question) return f"Similarity: {similarity:.2f}\nAnswer: {answer}" iface = gr.Interface( fn=gradio_app, inputs=gr.Textbox(label="Enter your question"), outputs=gr.Textbox(label="Answer"), title="Blood Donation Q&A", description="Ask questions about blood donation in English or Swahili. The system first checks if it's related to blood donation." ) iface.launch()