app.py

from haystack.nodes import BM25Retriever, FARMReader
from haystack.document_stores import ElasticsearchDocumentStore
from haystack.pipelines import ExtractiveQAPipeline
from abc import ABC, abstractmethod
import certifi
import gradio as gr
import examples

ca_certs=certifi.where()


class Config():
    es_host = "ask2democracy.es.us-central1.gcp.cloud.es.io"
    es_user = "elastic"
    es_password = "siKAHmmk2flwEaKNqQVZwp49"
    proposals_index = "petrolfo"
    #reader_model_name_or_path = "deepset/roberta-base-squad2"
    reader_model_name_or_path = "deepset/xlm-roberta-large-squad2"
    use_gpu = True
    
class DocumentQueries(ABC):
    
    @abstractmethod
    def search_by_query(self, query : str, retriever_top_k: int, reader_top_k: int, es_index: str):
        pass

class ExtractiveProposalQueries(DocumentQueries):
    
    def __init__(self, es_host: str, es_index: str, es_user, es_password, reader_name_or_path: str, use_gpu = True) -> None:
        reader = FARMReader(model_name_or_path = reader_name_or_path, use_gpu = use_gpu, num_processes=1)
        self._initialize_pipeline(es_host, es_index, es_user, es_password, reader = reader)
        
    
    def _initialize_pipeline(self, es_host, es_index, es_user, es_password, reader = None):
        if reader is not None:
            self.reader = reader
        self.es_host = es_host
        self.es_user = es_user
        self.es_password = es_password
        self.document_store = ElasticsearchDocumentStore(host = es_host, username=es_user, password=es_password, index = es_index, port = 443, scheme='https', verify_certs=True, ca_certs=ca_certs)        
        self.retriever = BM25Retriever(document_store = self.document_store)
        self.pipe = ExtractiveQAPipeline(self.reader, self.retriever)

    def search_by_query(self, query : str, retriever_top_k: int, reader_top_k: int, es_index: str = None) :
        if es_index is not None:
            self._initialize_pipeline(self.es_host, es_index, self.es_user, self.es_password)
        params = {"Retriever": {"top_k": retriever_top_k}, "Reader": {"top_k": reader_top_k}}
        prediction = self.pipe.run( query = query, params = params)      
        return prediction["answers"]
    
query = ExtractiveProposalQueries(es_host = Config.es_host, es_index = Config.proposals_index,
                                  es_user = Config.es_user, es_password = Config.es_password,
                                  reader_name_or_path = Config.reader_model_name_or_path,
                                  use_gpu = Config.use_gpu)

def update(query):
    return f"{query}", f"{query}", f"{query}", f"{query}"

def search(question):
  p1_result = query.search_by_query(query = question, retriever_top_k = 5, reader_top_k = 1, es_index = "petro")
  p2_result = query.search_by_query(query = question, retriever_top_k = 5, reader_top_k = 1, es_index = "rodolfo")

  return [p1_result[0].answer, 
          p1_result[0].context,
          p2_result[0].answer,
          p2_result[0].context]

def update(query):
    return f"{query}", f"{query}", f"{query}", f"{query}"

def search(question):
  p1_result = query.search_by_query(query = question, retriever_top_k = 5, reader_top_k = 1, es_index = "petro")
  p2_result = query.search_by_query(query = question, retriever_top_k = 5, reader_top_k = 1, es_index = "rodolfo")

  return [p1_result[0].answer, 
          p1_result[0].context,
          p2_result[0].answer,
          p2_result[0].context]

query = ExtractiveProposalQueries(es_host = Config.es_host, es_index = Config.proposals_index,
                                  es_user = Config.es_user, es_password = Config.es_password,
                                  reader_name_or_path = Config.reader_model_name_or_path,
                                  use_gpu = Config.use_gpu)

def update(query):
    return f"{query}", f"{query}", f"{query}", f"{query}"

def search(question):
  p1_result = query.search_by_query(query = question, retriever_top_k = 5, reader_top_k = 3, es_index = "petro")
  p2_result = query.search_by_query(query = question, retriever_top_k = 5, reader_top_k = 3, es_index = "rodolfo")

  return [p1_result[0].answer, 
          p1_result[0].context,
          p2_result[0].answer,
          p2_result[0].context]

description = """
   <p> 
    Ask2Democracy🇨🇴 - Elecciones precidenciales Colombia 2022
    El debate ha sido un ausente en la segunda vuelta por la presidencia de Colombia del 2022. 
    para ayudar con el conocimiento de las propuestas de los candidatos construí este espacio,
    puede hacerle preguntas a la propuestas de los candidatos.
    Un pequeño aporte a un fin de semana determinante en la democracia Colombiana [Abajo algunos ejemplos]👇
    </p>
    <p>
    <br>Este espacio hace parte de un proyecto open source que busca ayudar con el conocimiento de la historia Colombiana.  <a href= "https://github.com/jorge-henao/ask_to_democracy">repo en github con FastAPI</a></br>
    </p>
    <p>
    Por: Jorge Henao 🇨🇴 <a href="https://twitter.com/jhenaotw" target='_blank'>Twitter</a> <a href="https://www.linkedin.com/in/henaojorge" target='_blank'/>LinkedIn</a>
    </p> 
"""


iface = gr.Interface(fn=search,
                      title= "Ask2Democracy🇨🇴 - Elecciones precidenciales Colombia 2022",
                      description = description,
                      inputs=[
                              gr.inputs.Textbox(lines=2, label="hazle una pregunta a los candidatos", placeholder='que va a hacer con la deuda del ICETEX?')
                              ],
                      outputs=[
                                gr.outputs.Textbox(label="Respuesta de Petro"),
                                gr.outputs.Textbox(label="Contexto de la propuesta"),
                                gr.outputs.Textbox(label="Respuesta de Rodolfo Hernandez"),
                                gr.outputs.Textbox(label="Contexto de la propuesta")                    
                              ],
                      examples = examples,
                      theme = 'peach'
                     )
iface.launch(enable_queue=True)