app.py

import gradio as gr
from sentence_transformers import SentenceTransformer
from huggingface_hub import InferenceClient
import pandas as pd
import torch
import math
import httpcore
import pickle
import time
setattr(httpcore, 'SyncHTTPTransport', 'AsyncHTTPProxy')
from googletrans import Translator

"""
For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
"""
client = InferenceClient("HuggingFaceH4/zephyr-7b-beta")
model = SentenceTransformer('intfloat/multilingual-e5-large-instruct')

def get_detailed_instruct(task_description: str, query: str) -> str:
    return f'Instruct: {task_description}\nQuery: {query}'


def respond(
    message,
    max_tokens,
    temperature,
    top_p,
):
    translator = Translator()
    
    #messages = [{"role": "system", "content": "You are a sunni moslem bot that always give answer based on quran, hadith, and the companions of prophet Muhammad!"}]
    messages = [{"role": "system", "content": "Act as a sunni moslem bot that submits fully to Allah, and always give answer after giving reference of verses in quran and/or hadiths"}]
    ##make a moslem bot
    messages.append({"role": "user", "content": "I want you to answer strictly based on quran and hadith"})
    messages.append({"role": "assistant", "content": "I'd be happy to help! Please go ahead and provide the sentence you'd like me to analyze. Please specify whether you're referencing a particular verse or hadith (Prophetic tradition) from the Quran or Hadith, or if you're asking me to analyze a general statement."})

    #adding fatwa references
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    selected_references = torch.load('selected_references.sav', map_location=torch.device(device))
    encoded_questions = torch.load('encoded_questions.sav', map_location=torch.device(device))
    
    task = 'Given a web search query, retrieve relevant passages that answer the query'
    queries = [
        get_detailed_instruct(task, message)
    ]
    print("start\n")
    print(time.time())

    query_embeddings = model.encode(queries, convert_to_tensor=True, normalize_embeddings=True)
    scores = (query_embeddings @ encoded_questions.T) * 100
    selected_references['similarity'] = scores.tolist()[0]
    sorted_references = selected_references.sort_values(by='similarity', ascending=False)
    sorted_references = sorted_references.iloc[:3]
    sorted_references = sorted_references.sort_values(by='similarity', ascending=True)
    print(sorted_references.shape[0])
    print(sorted_references['similarity'].tolist())
    print("sorted references\n")
    print(time.time())

    
    for index, row in sorted_references.iterrows():
        if(type(row["user"]) is str and type(row['assistant']) is str):
            try:
                translator = Translator()
                print(index)
                print(f'{row["user"]}')
                translated = translator.translate(f'{row["user"]}', src='ar', dest='en')
                print(translated)
                user = translated.text
                print(user)
                assistant = translator.translate(row['assistant']).text
                messages.append({"role": "user", "content":user })
                messages.append({"role": "assistant", "content": assistant})
            except Exception as error:
                print("1. An error occurred:", error)
                print("adding fatwa references exception occurred")
    
    print("append references\n")
    print(time.time())
    
    #adding more references
    df = pd.read_csv("moslem-bot-reference.csv", sep='|')
    for index, row in df.iterrows():
        messages.append({"role": "user", "content": row['user']})
        messages.append({"role": "assistant", "content": row['assistant']})
    
    #latest user question
    translator = Translator()
    en_message = ""
    message_language = "en"
    print("===message===")
    print(message)
    print("============")
    try:
        translator = Translator()
        print(translator.detect(message))
        message_language = translator.detect(message).lang
        print(message_language)
        print(translator.translate(message))
        en_message = translator.translate(message).text
        messages.append({"role": "user", "content": en_message})
    except Exception as error:
        messages.append({"role": "user", "content": message})
        print("An error occurred:", error)
        print("en_message exception occurred")

    response = ""

    for message in client.chat_completion(
        messages,
        max_tokens=max_tokens,
        stream=True,
        temperature=temperature,
        top_p=top_p,
    ):
        token = message.choices[0].delta.content

        response += token

        yield response
        '''
        if(len(token)==0 and message_language!='en'):
            translated_response = translator.translate(response, src='en', dest=message_language)
            if(translated_response and translated_response.text):
                yield translated_response.text
            else:
                yield response
        else:
            yield response
        '''


"""
For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
"""
demo = gr.Interface(
    respond,
    additional_inputs=[
        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
        gr.Slider(
            minimum=0.1,
            maximum=1.0,
            value=0.95,
            step=0.05,
            label="Top-p (nucleus sampling)",
        ),
    ],
    inputs="textbox", 
    outputs="textbox",   
    cache_examples="lazy",
    examples=[
                ["Why is men created?"],
                ["Please tell me about superstition!"],
                ["How moses defeat pharaoh?"],
            ],
)


if __name__ == "__main__":
    demo.launch()