app.py

import torch
import transformers
import gradio as gr
from ragatouille import RAGPretrainedModel
from huggingface_hub import InferenceClient
import re
from datetime import datetime
import json

import arxiv
from helper import *

retrieve_results = 20 
show_examples = True
llm_models_to_choose = ['mistralai/Mixtral-8x7B-Instruct-v0.1','mistralai/Mistral-7B-Instruct-v0.2', 'google/gemma-7b-it', 'None']

generate_kwargs = dict(
    temperature = None,
    max_new_tokens = 512,
    top_p = None,
    do_sample = False,
    )

## RAG Model
RAG = RAGPretrainedModel.from_index("colbert/indexes/arxiv_colbert")

try:
  gr.Info("Setting up retriever, please wait...")
  rag_initial_output = RAG.search("What is Generative AI in Healthcare?", k = 1)
  gr.Info("Retriever working successfully!")
    
except:
  gr.Warning("Retriever not working!")

## Header
mark_text = '# 🩺🔍 Search Results\n'
header_text = "## Arxiv Paper Summary With QA Retrieval Augmented Generation \n"

try:
  with open("README.md", "r") as f:
      mdfile = f.read()
  date_pattern = r'Index Last Updated : \d{4}-\d{2}-\d{2}'
  match = re.search(date_pattern, mdfile)
  date = match.group().split(': ')[1]
  formatted_date = datetime.strptime(date, '%Y-%m-%d').strftime('%d %b %Y')
  header_text += f'Index Last Updated: {formatted_date}\n'
  index_info = f"Semantic Search - up to {formatted_date}"  
except:
  index_info = "Semantic Search"

database_choices = [index_info,'Arxiv Search - Latest - (EXPERIMENTAL)']

## Arxiv API
arx_client = arxiv.Client()
is_arxiv_available = True
check_arxiv_result = get_arxiv_live_search("What is Self Rewarding AI and how can it be used in Multi-Agent Systems?", arx_client, retrieve_results)
if len(check_arxiv_result) == 0:
  is_arxiv_available = False
  print("Arxiv search not working, switching to default search ...")
  database_choices = [index_info]



## Show examples
sample_outputs = {
    'output_placeholder': 'The LLM will provide an answer to your question here...',
    'search_placeholder': '''
1. What is MoE?
2. What are Multi Agent Systems? 
3. What is Self Rewarding AI?
4. What is Semantic and Episodic memory?
5. What is AutoGen?
6. What is ChatDev?
7. What is Omniverse?
8. What is Lumiere?
9. What is SORA?
'''
}

output_placeholder = sample_outputs['output_placeholder']
md_text_initial = sample_outputs['search_placeholder']


with gr.Blocks(theme = gr.themes.Soft()) as demo:
    header = gr.Markdown(header_text)
    
    with gr.Group():
      msg = gr.Textbox(label = 'Search', placeholder = 'What is Generative AI in Healthcare?')
        
      with gr.Accordion("Advanced Settings", open=False):
        with gr.Row(equal_height = True):
          llm_model = gr.Dropdown(choices = llm_models_to_choose, value = 'mistralai/Mistral-7B-Instruct-v0.2', label = 'LLM Model')
          llm_results = gr.Slider(minimum=4, maximum=10, value=5, step=1, interactive=True, label="Top n results as context")
          database_src = gr.Dropdown(choices = database_choices, value = index_info, label = 'Search Source')
          stream_results = gr.Checkbox(value = True, label = "Stream output", visible = False)

    output_text = gr.Textbox(show_label = True, container = True, label = 'LLM Answer', visible = True, placeholder = output_placeholder)
    input = gr.Textbox(show_label = False, visible = False)
    gr_md = gr.Markdown(mark_text + md_text_initial)

    def update_with_rag_md(message, llm_results_use = 5, database_choice = index_info, llm_model_picked = 'mistralai/Mistral-7B-Instruct-v0.2'):
        prompt_text_from_data = ""
        database_to_use = database_choice
        if database_choice == index_info:
          rag_out = get_rag(message)
        else:
          arxiv_search_success = True
          try:
            rag_out = get_arxiv_live_search(message, arx_client, retrieve_results)
            if len(rag_out) == 0:
              arxiv_search_success = False 
          except:
            arxiv_search_success = False
 

          if not arxiv_search_success:
            gr.Warning("Arxiv Search not working, switching to semantic search ...")
            rag_out = get_rag(message)
            database_to_use = index_info 

        md_text_updated = mark_text
        for i in range(retrieve_results):
          rag_answer = rag_out[i]
          if i < llm_results_use:
            md_text_paper, prompt_text = get_md_text_abstract(rag_answer, source = database_to_use, return_prompt_formatting = True)
            prompt_text_from_data += f"{i+1}. {prompt_text}"
          else:
            md_text_paper = get_md_text_abstract(rag_answer, source = database_to_use)
          md_text_updated += md_text_paper
        prompt = get_prompt_text(message, prompt_text_from_data, llm_model_picked = llm_model_picked)
        return md_text_updated, prompt

    def ask_llm(prompt, llm_model_picked = 'mistralai/Mistral-7B-Instruct-v0.2', stream_outputs = False):
       model_disabled_text = "LLM Model is disabled"
       output = ""
        
       if llm_model_picked == 'None':
          if stream_outputs:
              for out in model_disabled_text:
                output += out
                yield output
              return output 
          else:
              return model_disabled_text
              
       client = InferenceClient(llm_model_picked)
       try:
           stream = client.text_generation(prompt, **generate_kwargs,  stream=stream_outputs, details=False, return_full_text=False)
           
       except:
           gr.Warning("LLM Inference rate limit reached, try again later!")
           return ""
       
       if stream_outputs:
           for response in stream:
              output += response
              SaveResponseAndRead(response)
              yield output
           return output
       else:
           return stream


    msg.submit(update_with_rag_md, [msg, llm_results,  database_src, llm_model], [gr_md, input]).success(ask_llm, [input, llm_model, stream_results], output_text)

demo.queue().launch()