NCTCMumbai's picture
Update app.py
97badab verified
raw
history blame
24.7 kB
import requests
import gradio as gr
from ragatouille import RAGPretrainedModel
import logging
from pathlib import Path
from time import perf_counter
from sentence_transformers import CrossEncoder
from huggingface_hub import InferenceClient
from jinja2 import Environment, FileSystemLoader
import numpy as np
from os import getenv
from backend.query_llm import generate_hf, generate_openai
from backend.semantic_search import table, retriever
from huggingface_hub import InferenceClient
# Bhashini API translation function
api_key = getenv('API_KEY')
user_id = getenv('USER_ID')
def bhashini_translate(text: str, from_code: str = "en", to_code: str = "hi") -> dict:
"""Translates text from source language to target language using the Bhashini API."""
print(f'Starting translation process from {from_code} to {to_code}...')
gr.Warning(f'Translating to {to_code}...')
url = 'https://meity-auth.ulcacontrib.org/ulca/apis/v0/model/getModelsPipeline'
headers = {
"Content-Type": "application/json",
"userID": user_id,
"ulcaApiKey": api_key
}
payload = {
"pipelineTasks": [{"taskType": "translation", "config": {"language": {"sourceLanguage": from_code, "targetLanguage": to_code}}}],
"pipelineRequestConfig": {"pipelineId": "64392f96daac500b55c543cd"}
}
print('Sending initial request to get the pipeline...')
response = requests.post(url, json=payload, headers=headers)
if response.status_code != 200:
print(f'Error in initial request: {response.status_code}')
return {"status_code": response.status_code, "message": "Error in translation request", "translated_content": None}
print('Initial request successful, processing response...')
response_data = response.json()
service_id = response_data["pipelineResponseConfig"][0]["config"][0]["serviceId"]
callback_url = response_data["pipelineInferenceAPIEndPoint"]["callbackUrl"]
print(f'Service ID: {service_id}, Callback URL: {callback_url}')
headers2 = {
"Content-Type": "application/json",
response_data["pipelineInferenceAPIEndPoint"]["inferenceApiKey"]["name"]: response_data["pipelineInferenceAPIEndPoint"]["inferenceApiKey"]["value"]
}
compute_payload = {
"pipelineTasks": [{"taskType": "translation", "config": {"language": {"sourceLanguage": from_code, "targetLanguage": to_code}, "serviceId": service_id}}],
"inputData": {"input": [{"source": text}], "audio": [{"audioContent": None}]}
}
print(f'Sending translation request with text: "{text}"')
compute_response = requests.post(callback_url, json=compute_payload, headers=headers2)
if compute_response.status_code != 200:
print(f'Error in translation request: {compute_response.status_code}')
return {"status_code": compute_response.status_code, "message": "Error in translation", "translated_content": None}
print('Translation request successful, processing translation...')
compute_response_data = compute_response.json()
translated_content = compute_response_data["pipelineResponse"][0]["output"][0]["target"]
print(f'Translation successful. Translated content: "{translated_content}"')
return {"status_code": 200, "message": "Translation successful", "translated_content": translated_content}
# Existing chatbot functions
VECTOR_COLUMN_NAME = "vector"
TEXT_COLUMN_NAME = "text"
HF_TOKEN = getenv("HUGGING_FACE_HUB_TOKEN")
proj_dir = Path(__file__).parent
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
client = InferenceClient("mistralai/Mixtral-8x7B-Instruct-v0.1", token=HF_TOKEN)
env = Environment(loader=FileSystemLoader(proj_dir / 'templates'))
template = env.get_template('template.j2')
template_html = env.get_template('template_html.j2')
def add_text(history, text):
history = [] if history is None else history
history = history + [(text, None)]
return history, gr.Textbox(value="", interactive=False)
def bot(history, cross_encoder):
top_rerank = 25
top_k_rank = 20
query = history[-1][0]
if not query:
gr.Warning("Please submit a non-empty string as a prompt")
raise ValueError("Empty string was submitted")
logger.warning('Retrieving documents...')
if cross_encoder == '(HIGH ACCURATE) ColBERT':
gr.Warning('Retrieving using ColBERT.. First time query will take a minute for model to load..pls wait')
RAG = RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0")
RAG_db = RAG.from_index('.ragatouille/colbert/indexes/cbseclass10index')
documents_full = RAG_db.search(query, k=top_k_rank)
documents = [item['content'] for item in documents_full]
prompt = template.render(documents=documents, query=query)
prompt_html = template_html.render(documents=documents, query=query)
generate_fn = generate_hf
history[-1][1] = ""
for character in generate_fn(prompt, history[:-1]):
history[-1][1] = character
yield history, prompt_html
else:
document_start = perf_counter()
query_vec = retriever.encode(query)
doc1 = table.search(query_vec, vector_column_name=VECTOR_COLUMN_NAME).limit(top_k_rank)
documents = table.search(query_vec, vector_column_name=VECTOR_COLUMN_NAME).limit(top_rerank).to_list()
documents = [doc[TEXT_COLUMN_NAME] for doc in documents]
query_doc_pair = [[query, doc] for doc in documents]
if cross_encoder == '(FAST) MiniLM-L6v2':
cross_encoder1 = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
elif cross_encoder == '(ACCURATE) BGE reranker':
cross_encoder1 = CrossEncoder('BAAI/bge-reranker-base')
cross_scores = cross_encoder1.predict(query_doc_pair)
sim_scores_argsort = list(reversed(np.argsort(cross_scores)))
documents = [documents[idx] for idx in sim_scores_argsort[:top_k_rank]]
document_time = perf_counter() - document_start
prompt = template.render(documents=documents, query=query)
prompt_html = template_html.render(documents=documents, query=query)
generate_fn = generate_hf
history[-1][1] = ""
for character in generate_fn(prompt, history[:-1]):
history[-1][1] = character
yield history, prompt_html
def translate_text(response_text, selected_language):
iso_language_codes = {
"Hindi": "hi",
"Gom": "gom",
"Kannada": "kn",
"Dogri": "doi",
"Bodo": "brx",
"Urdu": "ur",
"Tamil": "ta",
"Kashmiri": "ks",
"Assamese": "as",
"Bengali": "bn",
"Marathi": "mr",
"Sindhi": "sd",
"Maithili": "mai",
"Punjabi": "pa",
"Malayalam": "ml",
"Manipuri": "mni",
"Telugu": "te",
"Sanskrit": "sa",
"Nepali": "ne",
"Santali": "sat",
"Gujarati": "gu",
"Odia": "or"
}
to_code = iso_language_codes[selected_language]
translation = bhashini_translate(response_text, to_code=to_code)
return translation['translated_content']
# Gradio interface
with gr.Blocks(theme='NoCrypt/miku') as CHATBOT:
with gr.Row():
with gr.Column(scale=10):
gr.HTML(value="""<div style="color: #FF4500;"><h1>ADWITIYA-</h1> <h1><span style="color: #008000">Custom Manual Chatbot and Quizbot</span></h1></div>""")
gr.HTML(value=f"""<p style="font-family: sans-serif; font-size: 16px;">Using GenAI for CBIC Capacity Building - A free chat bot developed by National Customs Targeting Center using Open source LLMs for CBIC Officers</p>""")
gr.HTML(value=f"""<p style="font-family: Arial, sans-serif; font-size: 14px;">Developed by NCTC,Mumbai. Suggestions may be sent to <a href="mailto:[email protected]" style="color: #00008B; font-style: italic;">[email protected]</a>.</p>""")
with gr.Column(scale=3):
gr.Image(value='logo.png', height=200, width=200)
chatbot = gr.Chatbot(
[],
elem_id="chatbot",
avatar_images=('https://aui.atlassian.com/aui/8.8/docs/images/avatar-person.svg',
'https://huggingface.co/datasets/huggingface/brand-assets/resolve/main/hf-logo.svg'),
bubble_full_width=False,
show_copy_button=True,
show_share_button=True,
)
with gr.Row():
txt = gr.Textbox(
scale=3,
show_label=False,
placeholder="Enter text and press enter",
container=False,
)
txt_btn = gr.Button(value="Submit text", scale=1)
cross_encoder = gr.Radio(choices=['(FAST) MiniLM-L6v2', '(ACCURATE) BGE reranker', '(HIGH ACCURATE) ColBERT'], value='(ACCURATE) BGE reranker', label="Embeddings", info="Only First query to Colbert may take little time)")
language_dropdown = gr.Dropdown(
choices=[
"Hindi", "Gom", "Kannada", "Dogri", "Bodo", "Urdu", "Tamil", "Kashmiri", "Assamese", "Bengali", "Marathi",
"Sindhi", "Maithili", "Punjabi", "Malayalam", "Manipuri", "Telugu", "Sanskrit", "Nepali", "Santali",
"Gujarati", "Odia"
],
value="Hindi", # default to Hindi
label="Select Language for Translation"
)
prompt_html = gr.HTML()
translated_textbox = gr.Textbox(label="Translated Response")
txt_msg = txt_btn.click(add_text, [chatbot, txt], [chatbot, txt], queue=False).then(
bot, [chatbot, cross_encoder], [chatbot, prompt_html]).then(
translate_text, [txt, language_dropdown], translated_textbox
)
txt_msg = txt.submit(add_text, [chatbot, txt], [chatbot, txt], queue=False).then(
bot, [chatbot, cross_encoder], [chatbot, prompt_html]).then(
translate_text, [txt, language_dropdown], translated_textbox
)
# Launch the Gradio application
CHATBOT.launch(share=True)
# from ragatouille import RAGPretrainedModel
# import subprocess
# import json
# import spaces
# import firebase_admin
# from firebase_admin import credentials, firestore
# import logging
# from pathlib import Path
# from time import perf_counter
# from datetime import datetime
# import gradio as gr
# from jinja2 import Environment, FileSystemLoader
# import numpy as np
# from sentence_transformers import CrossEncoder
# from huggingface_hub import InferenceClient
# from os import getenv
# from backend.query_llm import generate_hf, generate_openai
# from backend.semantic_search import table, retriever
# from huggingface_hub import InferenceClient
# VECTOR_COLUMN_NAME = "vector"
# TEXT_COLUMN_NAME = "text"
# HF_TOKEN = getenv("HUGGING_FACE_HUB_TOKEN")
# proj_dir = Path(__file__).parent
# # Setting up the logging
# logging.basicConfig(level=logging.INFO)
# logger = logging.getLogger(__name__)
# client = InferenceClient("mistralai/Mixtral-8x7B-Instruct-v0.1",token=HF_TOKEN)
# # Set up the template environment with the templates directory
# env = Environment(loader=FileSystemLoader(proj_dir / 'templates'))
# # Load the templates directly from the environment
# template = env.get_template('template.j2')
# template_html = env.get_template('template_html.j2')
# def add_text(history, text):
# history = [] if history is None else history
# history = history + [(text, None)]
# return history, gr.Textbox(value="", interactive=False)
# def bot(history, cross_encoder):
# top_rerank = 25
# top_k_rank = 20
# query = history[-1][0]
# if not query:
# gr.Warning("Please submit a non-empty string as a prompt")
# raise ValueError("Empty string was submitted")
# logger.warning('Retrieving documents...')
# # if COLBERT RAGATATOUILLE PROCEDURE :
# if cross_encoder=='(HIGH ACCURATE) ColBERT':
# gr.Warning('Retrieving using ColBERT.. First time query will take a minute for model to load..pls wait')
# RAG= RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0")
# RAG_db=RAG.from_index('.ragatouille/colbert/indexes/cbseclass10index')
# documents_full=RAG_db.search(query,k=top_k_rank)
# documents=[item['content'] for item in documents_full]
# # Create Prompt
# prompt = template.render(documents=documents, query=query)
# prompt_html = template_html.render(documents=documents, query=query)
# generate_fn = generate_hf
# history[-1][1] = ""
# for character in generate_fn(prompt, history[:-1]):
# history[-1][1] = character
# yield history, prompt_html
# print('Final history is ',history)
# #store_message(db,history[-1][0],history[-1][1],cross_encoder)
# else:
# # Retrieve documents relevant to query
# document_start = perf_counter()
# query_vec = retriever.encode(query)
# logger.warning(f'Finished query vec')
# doc1 = table.search(query_vec, vector_column_name=VECTOR_COLUMN_NAME).limit(top_k_rank)
# logger.warning(f'Finished search')
# documents = table.search(query_vec, vector_column_name=VECTOR_COLUMN_NAME).limit(top_rerank).to_list()
# documents = [doc[TEXT_COLUMN_NAME] for doc in documents]
# logger.warning(f'start cross encoder {len(documents)}')
# # Retrieve documents relevant to query
# query_doc_pair = [[query, doc] for doc in documents]
# if cross_encoder=='(FAST) MiniLM-L6v2' :
# cross_encoder1 = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
# elif cross_encoder=='(ACCURATE) BGE reranker':
# cross_encoder1 = CrossEncoder('BAAI/bge-reranker-base')
# cross_scores = cross_encoder1.predict(query_doc_pair)
# sim_scores_argsort = list(reversed(np.argsort(cross_scores)))
# logger.warning(f'Finished cross encoder {len(documents)}')
# documents = [documents[idx] for idx in sim_scores_argsort[:top_k_rank]]
# logger.warning(f'num documents {len(documents)}')
# document_time = perf_counter() - document_start
# logger.warning(f'Finished Retrieving documents in {round(document_time, 2)} seconds...')
# # Create Prompt
# prompt = template.render(documents=documents, query=query)
# prompt_html = template_html.render(documents=documents, query=query)
# generate_fn = generate_hf
# history[-1][1] = ""
# for character in generate_fn(prompt, history[:-1]):
# history[-1][1] = character
# yield history, prompt_html
# print('Final history is ',history)
# #store_message(db,history[-1][0],history[-1][1],cross_encoder)
# # def system_instructions(question_difficulty, topic,documents_str):
# # return f"""<s> [INST] Your are a great teacher and your task is to create 10 questions with 4 choices with a {question_difficulty} difficulty about topic request " {topic} " only from the below given documents, {documents_str} then create an answers. Index in JSON format, the questions as "Q#":"" to "Q#":"", the four choices as "Q#:C1":"" to "Q#:C4":"", and the answers as "A#":"Q#:C#" to "A#":"Q#:C#". [/INST]"""
# RAG_db = gr.State()
# # def load_model():
# # try:
# # # Initialize the model
# # RAG = RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0")
# # # Load the RAG database
# # RAG_db.value = RAG.from_index('.ragatouille/colbert/indexes/cbseclass10index')
# # return 'Ready to Go!!'
# # except Exception as e:
# # return f"Error loading model: {e}"
# # def generate_quiz(question_difficulty, topic):
# # if not topic.strip():
# # return ['Please enter a valid topic.'] + [gr.Radio(visible=False) for _ in range(10)]
# # top_k_rank = 10
# # # Load the model and database within the generate_quiz function
# # try:
# # RAG = RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0")
# # RAG_db_ = RAG.from_index('.ragatouille/colbert/indexes/cbseclass10index')
# # gr.Warning('Model loaded!')
# # except Exception as e:
# # return [f"Error loading model: {e}"] + [gr.Radio(visible=False) for _ in range(10)]
# # RAG_db_ = RAG_db.value
# # documents_full = RAG_db_.search(topic, k=top_k_rank)
# # generate_kwargs = dict(
# # temperature=0.2,
# # max_new_tokens=4000,
# # top_p=0.95,
# # repetition_penalty=1.0,
# # do_sample=True,
# # seed=42,
# # )
# # question_radio_list = []
# # count = 0
# # while count <= 3:
# # try:
# # documents = [item['content'] for item in documents_full]
# # document_summaries = [f"[DOCUMENT {i+1}]: {summary}{count}" for i, summary in enumerate(documents)]
# # documents_str = '\n'.join(document_summaries)
# # formatted_prompt = system_instructions(question_difficulty, topic, documents_str)
# # pre_prompt = [
# # {"role": "system", "content": formatted_prompt}
# # ]
# # response = client.text_generation(
# # formatted_prompt, **generate_kwargs, stream=False, details=False, return_full_text=False,
# # )
# # output_json = json.loads(f"{response}")
# # global quiz_data
# # quiz_data = output_json
# # for question_num in range(1, 11):
# # question_key = f"Q{question_num}"
# # answer_key = f"A{question_num}"
# # question = quiz_data.get(question_key)
# # answer = quiz_data.get(quiz_data.get(answer_key))
# # if not question or not answer:
# # continue
# # choice_keys = [f"{question_key}:C{i}" for i in range(1, 5)]
# # choice_list = [quiz_data.get(choice_key, "Choice not found") for choice_key in choice_keys]
# # radio = gr.Radio(choices=choice_list, label=question, visible=True, interactive=True)
# # question_radio_list.append(radio)
# # if len(question_radio_list) == 10:
# # break
# # else:
# # count += 1
# # continue
# # except Exception as e:
# # count += 1
# # if count == 3:
# # return ['Sorry. Pls try with another topic!'] + [gr.Radio(visible=False) for _ in range(10)]
# # continue
# # return ['Quiz Generated!'] + question_radio_list
# # def compare_answers(*user_answers):
# # user_answer_list = user_answers
# # answers_list = [quiz_data.get(quiz_data.get(f"A{question_num}")) for question_num in range(1, 11)]
# # score = sum(1 for answer in user_answer_list if answer in answers_list)
# # if score > 7:
# # message = f"### Excellent! You got {score} out of 10!"
# # elif score > 5:
# # message = f"### Good! You got {score} out of 10!"
# # else:
# # message = f"### You got {score} out of 10! Don’t worry, you can prepare well and try better next time!"
# # return message
# #with gr.Blocks(theme='Insuz/SimpleIndigo') as demo:
# with gr.Blocks(theme='NoCrypt/miku') as CHATBOT:
# with gr.Row():
# with gr.Column(scale=10):
# # gr.Markdown(
# # """
# # # Theme preview: `paris`
# # To use this theme, set `theme='earneleh/paris'` in `gr.Blocks()` or `gr.Interface()`.
# # You can append an `@` and a semantic version expression, e.g. @>=1.0.0,<2.0.0 to pin to a given version
# # of this theme.
# # """
# # )
# gr.HTML(value="""<div style="color: #FF4500;"><h1>ADWITIYA-</h1> <h1><span style="color: #008000">Custom Manual Chatbot and Quizbot</span></h1>
# </div>""", elem_id='heading')
# gr.HTML(value=f"""
# <p style="font-family: sans-serif; font-size: 16px;">
# Using GenAI for CBIC Capacity Building - A free chat bot developed by National Customs Targeting Center using Open source LLMs for CBIC Officers
# </p>
# """, elem_id='Sub-heading')
# #usage_count = get_and_increment_value_count(db,collection_name, field_name)
# gr.HTML(value=f"""<p style="font-family: Arial, sans-serif; font-size: 14px;">Developed by NCTC,Mumbai . Suggestions may be sent to <a href="mailto:[email protected]" style="color: #00008B; font-style: italic;">[email protected]</a>.</p>""", elem_id='Sub-heading1 ')
# with gr.Column(scale=3):
# gr.Image(value='logo.png',height=200,width=200)
# chatbot = gr.Chatbot(
# [],
# elem_id="chatbot",
# avatar_images=('https://aui.atlassian.com/aui/8.8/docs/images/avatar-person.svg',
# 'https://huggingface.co/datasets/huggingface/brand-assets/resolve/main/hf-logo.svg'),
# bubble_full_width=False,
# show_copy_button=True,
# show_share_button=True,
# )
# with gr.Row():
# txt = gr.Textbox(
# scale=3,
# show_label=False,
# placeholder="Enter text and press enter",
# container=False,
# )
# txt_btn = gr.Button(value="Submit text", scale=1)
# cross_encoder = gr.Radio(choices=['(FAST) MiniLM-L6v2','(ACCURATE) BGE reranker','(HIGH ACCURATE) ColBERT'], value='(ACCURATE) BGE reranker',label="Embeddings", info="Only First query to Colbert may take litte time)")
# prompt_html = gr.HTML()
# # Turn off interactivity while generating if you click
# txt_msg = txt_btn.click(add_text, [chatbot, txt], [chatbot, txt], queue=False).then(
# bot, [chatbot, cross_encoder], [chatbot, prompt_html])#.then(update_count_html,[],[count_html])
# # Turn it back on
# txt_msg.then(lambda: gr.Textbox(interactive=True), None, [txt], queue=False)
# # Turn off interactivity while generating if you hit enter
# txt_msg = txt.submit(add_text, [chatbot, txt], [chatbot, txt], queue=False).then(
# bot, [chatbot, cross_encoder], [chatbot, prompt_html])#.then(update_count_html,[],[count_html])
# # Turn it back on
# txt_msg.then(lambda: gr.Textbox(interactive=True), None, [txt], queue=False)
# # Examples
# gr.Examples(examples, txt)
# # with gr.Blocks(title="Quiz Maker", theme=gr.themes.Default(primary_hue="green", secondary_hue="green"), css="style.css") as QUIZBOT:
# # with gr.Column(scale=4):
# # gr.HTML("""
# # <center>
# # <h1><span style="color: purple;">ADWITIYA</span> Customs Manual Quizbot</h1>
# # <h2>Generative AI-powered Capacity building for Training Officers</h2>
# # <i>⚠️ NACIN Faculties create quiz from any topic dynamically for classroom evaluation after their sessions! ⚠️</i>
# # </center>
# # """)
# # with gr.Column(scale=2):
# # gr.HTML("""
# # <center>
# # <h2>Ready!</h2>
# # </center>
# # """)
# # # load_btn = gr.Button("Click to Load!🚀")
# # # load_text = gr.Textbox()
# # # load_btn.click(fn=load_model, outputs=load_text)
# # topic = gr.Textbox(label="Enter the Topic for Quiz", placeholder="Write any topic/details from Customs Manual")
# # with gr.Row():
# # radio = gr.Radio(["easy", "average", "hard"], label="How difficult should the quiz be?")
# # generate_quiz_btn = gr.Button("Generate Quiz!🚀")
# # quiz_msg = gr.Textbox()
# # question_radios = [gr.Radio(visible=False) for _ in range(10)]
# # generate_quiz_btn.click(
# # fn=generate_quiz,
# # inputs=[radio, topic],
# # outputs=[quiz_msg] + question_radios
# # )
# # check_button = gr.Button("Check Score")
# # score_textbox = gr.Markdown()
# # check_button.click(
# # fn=compare_answers,
# # inputs=question_radios,
# # outputs=score_textbox
# # )
# #demo = gr.TabbedInterface([CHATBOT, QUIZBOT], ["AI ChatBot", "AI Quizbot"])
# CHATBOT.queue()
# CHATBOT.launch(debug=True)