NCTCMumbai/Customs_Manual_Chatbot
0
1 2"""3Credit to Derek Thomas, derek@huggingface.co4"""5 6import subprocess7 8# subprocess.run(["pip", "install", "--upgrade", "transformers[torch,sentencepiece]==4.34.1"])9import spaces10import logging11from pathlib import Path12from time import perf_counter13 14import gradio as gr15from jinja2 import Environment, FileSystemLoader16import numpy as np17from sentence_transformers import CrossEncoder18 19from backend.query_llm import generate_hf, generate_openai,generate_gemini20from backend.semantic_search import table, retriever21 22VECTOR_COLUMN_NAME = "vector"23TEXT_COLUMN_NAME = "text"24 25proj_dir = Path(__file__).parent26# Setting up the logging27logging.basicConfig(level=logging.INFO)28logger = logging.getLogger(__name__)29 30# Set up the template environment with the templates directory31env = Environment(loader=FileSystemLoader(proj_dir / 'templates'))32 33# Load the templates directly from the environment34template = env.get_template('template.j2')35template_html = env.get_template('template_html.j2')36 37# crossEncoder38#cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') 39cross_encoder = CrossEncoder('BAAI/bge-reranker-base')40# Examples41examples = ['My transhipment cargo is missing','can u explain and tabulate difference between b 17 bond and a warehousing bond',42 'What are benefits of the AEO Scheme and eligibility criteria?',43 'What are penalties for customs offences? ', 'what are penalties to customs officers misusing their powers under customs act?','What are eligibility criteria for exemption from cost recovery charges','list in detail what is procedure for obtaining new approval for openeing a CFS attached to an ICD']44 45 46def add_text(history, text):47 history = [] if history is None else history48 history = history + [(text, None)]49 print('add_text function done..returning history' ,history)50 return history, gr.Textbox(value="", interactive=False)51 52 53def bot(history, api_kind):54 top_rerank = 1555 top_k_rank = 1056 query = history[-1][0]57 print('history[-1][0]',history[-1][0])58 print('api kind ',api_kind)59 60 if not query:61 gr.Warning("Please submit a non-empty string as a prompt")62 raise ValueError("Empty string was submitted")63 64 logger.warning('Retrieving documents...')65 # Retrieve documents relevant to query66 document_start = perf_counter()67 68 query_vec = retriever.encode(query)69 print(query)70 query_vec_flat = [arr.flatten() for arr in query_vec]71 logger.warning(f'Finished query vec')72 #documents = table.search(query_vec_flat, vector_column_name=VECTOR_COLUMN_NAME).limit(top_k_rank).to_list()73 74 75 76 logger.warning(f'Finished search')77 documents = table.search(query_vec, vector_column_name=VECTOR_COLUMN_NAME).limit(top_rerank).to_list()78 documents = [doc[TEXT_COLUMN_NAME] for doc in documents]79 logger.warning(f'start cross encoder {len(documents)}')80 # Retrieve documents relevant to query81 query_doc_pair = [[query, doc] for doc in documents]82 cross_scores = cross_encoder.predict(query_doc_pair)83 sim_scores_argsort = list(reversed(np.argsort(cross_scores)))84 logger.warning(f'Finished cross encoder {len(documents)}')85 86 documents = [documents[idx] for idx in sim_scores_argsort[:top_k_rank]]87 logger.warning(f'num documents {len(documents)}')88 89 document_time = perf_counter() - document_start90 logger.warning(f'Finished Retrieving documents in {round(document_time, 2)} seconds...')91 92 # Create Prompt93 prompt = template.render(documents=documents, query=query)94 prompt_html = template_html.render(documents=documents, query=query)95 96 if api_kind == "HuggingFace":97 generate_fn = generate_hf98 elif api_kind == "Gemini":99 print("Gemini condition satisfied")100 generate_fn = generate_gemini101 elif api_kind is None:102 gr.Warning("API name was not provided")103 raise ValueError("API name was not provided")104 else:105 gr.Warning(f"API {api_kind} is not supported")106 raise ValueError(f"API {api_kind} is not supported")107 try:108 count_tokens = lambda text: len([token.strip() for token in text.split() if token.strip()])109 print(prompt_html,'token count is',count_tokens(prompt_html))110 history[-1][1] = ""111 for character in generate_fn(prompt, history[:-1]):112 history[-1][1] = character113 yield history, prompt_html114 print('final history is ',history)115 # return history[-1][1], prompt_html116 117 except Exception as e: # Catch any exception118 print('An unexpected error occurred during generation:', str(e))119 yield f"An unexpected error occurred during generation: {str(e)}"120 121with gr.Blocks(theme='WeixuanYuan/Soft_dark') as CHATBOT:122 # Beautiful heading with logo123 gr.HTML(value="""124 <div style="display: flex; align-items: center; justify-content: space-between;">125 <h1 style="color: #008000">ADWITIYA - <span style="color: #008000">Customs Manual Chatbot</span></h1>126 <img src='logo.png' alt="Chatbot" width="50" height="50" />127 </div>128 """, elem_id="heading")129 130 # Formatted description131 gr.HTML(value="""<p style="font-family: sans-serif; font-size: 16px;">A free chat bot developed by National Customs Targeting Center using Open source LLMs.(Dedicated to 75th Batch IRS Probationers)</p>""", elem_id="description")132 133 chatbot = gr.Chatbot(134 [],135 elem_id="chatbot",136 avatar_images=('https://aui.atlassian.com/aui/8.8/docs/images/avatar-person.svg',137 'https://huggingface.co/datasets/huggingface/brand-assets/resolve/main/hf-logo.svg'),138 bubble_full_width=False,139 show_copy_button=True,140 show_share_button=True,141 )142 143 with gr.Row():144 txt = gr.Textbox(145 scale=3,146 show_label=False,147 placeholder="Enter text and press enter",148 container=False,149 )150 txt_btn = gr.Button(value="Submit text", scale=1)151 152 api_kind = gr.Radio(choices=["HuggingFace","Gemini"], value="HuggingFace")153 154 prompt_html = gr.HTML()155 #prompt_html = gr.Textbox(label='Retrieved Documents')156 try:157 # Turn off interactivity while generating if you click158 txt_msg = txt_btn.click(add_text, [chatbot, txt], [chatbot, txt], queue=False).then(159 bot, [chatbot, api_kind], [chatbot, prompt_html])160 except Exception as e:161 print ('Exception txt btn click ' ,str(e))162 # Turn it back on163 txt_msg.then(lambda: gr.Textbox(interactive=True), None, [txt], queue=False)164 try:165 # Turn off interactivity while generating if you hit enter166 txt_msg = txt.submit(add_text, [chatbot, txt], [chatbot, txt], queue=False).then(167 bot, [chatbot, api_kind], [chatbot, prompt_html])168 except Exception as e:169 print ('Exception ' ,str(e))170 171 # Turn it back on172 txt_msg.then(lambda: gr.Textbox(interactive=True), None, [txt], queue=False)173 174 # Examples175 gr.Examples(examples, txt)176 177 # QUIZBOT CODE178 RAG_db=gr.State()179 180 181 182with gr.Blocks(title="Quiz Maker", theme=gr.themes.Default(primary_hue="green", secondary_hue="green"), css="style.css") as QUIZBOT:183 def system_instructions(question_difficulty, topic,documents_str):184 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]"""185 186 def load_model():187 RAG= RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0")188 RAG_db.value=RAG.from_index('.ragatouille/colbert/indexes/cbseclass10index')189 return 'Ready to Go!!'190 with gr.Column(scale=4):191 gr.HTML("""192 <center>193 <h1><span style="color: purple;">AI NANBAN</span> - CBSE Class Quiz Maker</h1>194 <h2>AI-powered Learning Game</h2>195 <i>⚠️ Students create quiz from any topic /CBSE Chapter ! ⚠️</i>196 </center>197 """)198 #gr.Warning('Retrieving using ColBERT.. First time query will take a minute for model to load..pls wait')199 with gr.Column(scale=2):200 load_btn = gr.Button("Click to Load!🚀")201 load_text=gr.Textbox()202 load_btn.click(load_model,[],load_text)203 204 205 topic = gr.Textbox(label="Enter the Topic for Quiz", placeholder="Write any topic from CBSE notes")206 207 with gr.Row():208 radio = gr.Radio(209 ["easy", "average", "hard"], label="How difficult should the quiz be?"210 )211 212 213 generate_quiz_btn = gr.Button("Generate Quiz!🚀")214 quiz_msg=gr.Textbox()215 216 question_radios = [gr.Radio(visible=False), gr.Radio(visible=False), gr.Radio(217 visible=False), gr.Radio(visible=False), gr.Radio(visible=False), gr.Radio(visible=False), gr.Radio(visible=False), gr.Radio(218 visible=False), gr.Radio(visible=False), gr.Radio(visible=False)]219 220 print(question_radios)221 222 @spaces.GPU223 @generate_quiz_btn.click(inputs=[radio, topic], outputs=[quiz_msg]+question_radios, api_name="generate_quiz")224 def generate_quiz(question_difficulty, topic):225 top_k_rank=10226 RAG_db_=RAG_db.value227 documents_full=RAG_db_.search(topic,k=top_k_rank)228 229 230 231 generate_kwargs = dict(232 temperature=0.2,233 max_new_tokens=4000,234 top_p=0.95,235 repetition_penalty=1.0,236 do_sample=True,237 seed=42,238 )239 question_radio_list = []240 count=0241 while count<=3:242 try:243 documents=[item['content'] for item in documents_full]244 document_summaries = [f"[DOCUMENT {i+1}]: {summary}{count}" for i, summary in enumerate(documents)]245 documents_str='\n'.join(document_summaries)246 formatted_prompt = system_instructions(247 question_difficulty, topic,documents_str)248 print(formatted_prompt)249 pre_prompt = [250 {"role": "system", "content": formatted_prompt}251 ]252 response = client.text_generation(253 formatted_prompt, **generate_kwargs, stream=False, details=False, return_full_text=False,254 )255 output_json = json.loads(f"{response}")256 257 258 print(response)259 print('output json', output_json)260 261 global quiz_data262 263 quiz_data = output_json264 265 266 267 for question_num in range(1, 11):268 question_key = f"Q{question_num}"269 answer_key = f"A{question_num}"270 271 question = quiz_data.get(question_key)272 answer = quiz_data.get(quiz_data.get(answer_key))273 274 if not question or not answer:275 continue276 277 choice_keys = [f"{question_key}:C{i}" for i in range(1, 5)]278 choice_list = []279 for choice_key in choice_keys:280 choice = quiz_data.get(choice_key, "Choice not found")281 choice_list.append(f"{choice}")282 283 radio = gr.Radio(choices=choice_list, label=question,284 visible=True, interactive=True)285 286 question_radio_list.append(radio)287 if len(question_radio_list)==10:288 break289 else:290 print('10 questions not generated . So trying again!')291 count+=1292 continue293 except Exception as e:294 count+=1295 print(f"Exception occurred: {e}")296 if count==3:297 print('Retry exhausted')298 gr.Warning('Sorry. Pls try with another topic !')299 else:300 print(f"Trying again..{count} time...please wait")301 continue302 303 print('Question radio list ' , question_radio_list)304 305 return ['Quiz Generated!']+ question_radio_list306 307 check_button = gr.Button("Check Score")308 309 score_textbox = gr.Markdown()310 311 @check_button.click(inputs=question_radios, outputs=score_textbox)312 def compare_answers(*user_answers):313 user_anwser_list = []314 user_anwser_list = user_answers315 316 answers_list = []317 318 for question_num in range(1, 20):319 answer_key = f"A{question_num}"320 answer = quiz_data.get(quiz_data.get(answer_key))321 if not answer:322 break323 answers_list.append(answer)324 325 score = 0326 327 for item in user_anwser_list:328 if item in answers_list:329 score += 1330 if score>5:331 message = f"### Good ! You got {score} over 10!"332 elif score>7:333 message = f"### Excellent ! You got {score} over 10!"334 else:335 message = f"### You got {score} over 10! Dont worry . You can prepare well and try better next time !"336 337 return message338 339 340 341demo = gr.TabbedInterface([CHATBOT,QUIZBOT], ["AI ChatBot", "AI Nanban-Quizbot"])342 343 344demo.queue()345demo.launch(debug=True)346