""" Credit to Derek Thomas, derek@huggingface.co """ from ragatouille import RAGPretrainedModel import subprocess import logging from pathlib import Path from time import perf_counter import gradio as gr from jinja2 import Environment, FileSystemLoader import numpy as np from sentence_transformers import CrossEncoder from backend.query_llm import generate_hf, generate_openai from backend.semantic_search import table, retriever VECTOR_COLUMN_NAME = "vector" TEXT_COLUMN_NAME = "text" proj_dir = Path(__file__).parent # Setting up the logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # 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') # crossEncoder #cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') #cross_encoder = CrossEncoder('BAAI/bge-reranker-base') # Examples examples = ['what is social media and what are rules related to it for expenditure monitoring ', 'how many reports to be submitted by Expenditure observer with annexure names ?','what is expenditure limits for parlimentary constituency and assembly constituency' ] 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 = 15 top_k_rank = 10 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=='ColBERT': gr.Warning('Retrieving using ColBERT') RAG= RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0") RAG_db=RAG.from_index('.ragatouille/colbert/indexes/mockingbird') documents_full=RAG_db.search(query) 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 print('Final history is ',history) yield history, prompt_html 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=='MiniLM-L6v2' : cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') elif cross_encoder=='BGE reranker': cross_encoder = CrossEncoder('BAAI/bge-reranker-base') cross_scores = cross_encoder.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 print('Final history is ',history) yield history, prompt_html with gr.Blocks(theme='Insuz/SimpleIndigo') as demo: gr.HTML(value="""

NIRVACHANA - Expenditure Observer AI Assistant

Chatbot
""",elem_id='heading') gr.HTML(value="""

A free chat bot assistant for Expenditure Observers on Compendium on Election Expenditure Monitoring using Open source LLMs.
The bot can answer questions in natural language, taking relevant extracts from the ECI document which can be accessed here.

""",elem_id='Sub-heading') 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=['MiniLM-L6v2','BGE reranker','ColBERT'], value='BGE reranker',label="Embeddings", info="Choose MiniLM for Speed, BGE reranker for accuracy,ColBERT for both") 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]) # 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]) # Turn it back on txt_msg.then(lambda: gr.Textbox(interactive=True), None, [txt], queue=False) # Examples gr.Examples(examples, txt) demo.queue() demo.launch(debug=True)