import os import re import time from pathlib import Path from relik.retriever import GoldenRetriever from relik.retriever.indexers.inmemory import InMemoryDocumentIndex from relik.retriever.indexers.document import DocumentStore from relik.retriever import GoldenRetriever from relik.reader.pytorch_modules.span import RelikReaderForSpanExtraction import requests import streamlit as st from spacy import displacy from streamlit_extras.badges import badge from streamlit_extras.stylable_container import stylable_container # RELIK = os.getenv("RELIK", "localhost:8000/api/entities") import random from relik.inference.annotator import Relik from relik.inference.data.objects import ( AnnotationType, RelikOutput, Span, TaskType, Triples, ) def get_random_color(ents): colors = {} random_colors = generate_pastel_colors(len(ents)) for ent in ents: colors[ent] = random_colors.pop(random.randint(0, len(random_colors) - 1)) return colors def floatrange(start, stop, steps): if int(steps) == 1: return [stop] return [ start + float(i) * (stop - start) / (float(steps) - 1) for i in range(steps) ] def hsl_to_rgb(h, s, l): def hue_2_rgb(v1, v2, v_h): while v_h < 0.0: v_h += 1.0 while v_h > 1.0: v_h -= 1.0 if 6 * v_h < 1.0: return v1 + (v2 - v1) * 6.0 * v_h if 2 * v_h < 1.0: return v2 if 3 * v_h < 2.0: return v1 + (v2 - v1) * ((2.0 / 3.0) - v_h) * 6.0 return v1 # if not (0 <= s <= 1): raise ValueError, "s (saturation) parameter must be between 0 and 1." # if not (0 <= l <= 1): raise ValueError, "l (lightness) parameter must be between 0 and 1." r, b, g = (l * 255,) * 3 if s != 0.0: if l < 0.5: var_2 = l * (1.0 + s) else: var_2 = (l + s) - (s * l) var_1 = 2.0 * l - var_2 r = 255 * hue_2_rgb(var_1, var_2, h + (1.0 / 3.0)) g = 255 * hue_2_rgb(var_1, var_2, h) b = 255 * hue_2_rgb(var_1, var_2, h - (1.0 / 3.0)) return int(round(r)), int(round(g)), int(round(b)) def generate_pastel_colors(n): """Return different pastel colours. Input: n (integer) : The number of colors to return Output: A list of colors in HTML notation (eg.['#cce0ff', '#ffcccc', '#ccffe0', '#f5ccff', '#f5ffcc']) Example: >>> print generate_pastel_colors(5) ['#cce0ff', '#f5ccff', '#ffcccc', '#f5ffcc', '#ccffe0'] """ if n == 0: return [] # To generate colors, we use the HSL colorspace (see http://en.wikipedia.org/wiki/HSL_color_space) start_hue = 0.0 # 0=red 1/3=0.333=green 2/3=0.666=blue saturation = 1.0 lightness = 0.9 # We take points around the chromatic circle (hue): # (Note: we generate n+1 colors, then drop the last one ([:-1]) because # it equals the first one (hue 0 = hue 1)) return [ "#%02x%02x%02x" % hsl_to_rgb(hue, saturation, lightness) for hue in floatrange(start_hue, start_hue + 1, n + 1) ][:-1] def set_sidebar(css): with st.sidebar: st.markdown(f"", unsafe_allow_html=True) st.image( "https://upload.wikimedia.org/wikipedia/commons/8/87/The_World_Bank_logo.svg", use_column_width=True, ) st.markdown("### World Bank") st.markdown("### DIME") def get_el_annotations(response): i_link_wrapper = " Intervention {}" o_link_wrapper = " Outcome: {}" # swap labels key with ents ents = [ { "start": l.start, "end": l.end, "label": i_link_wrapper.format(l.label[0].upper() + l.label[1:].replace("/", "%2").replace(" ", "%20").replace("&","%26"), l.label), } if io_map[l.label] == "intervention" else { "start": l.start, "end": l.end, "label": o_link_wrapper.format(l.label[0].upper() + l.label[1:].replace("/", "%2").replace(" ", "%20").replace("&","%26"), l.label), } for l in response.spans ] dict_of_ents = {"text": response.text, "ents": ents} label_in_text = set(l["label"] for l in dict_of_ents["ents"]) options = {"ents": label_in_text, "colors": get_random_color(label_in_text)} return dict_of_ents, options def get_retriever_annotations(response): el_link_wrapper = " {}" # swap labels key with ents ents = [l.text for l in response.candidates[TaskType.SPAN] ] dict_of_ents = {"text": response.text, "ents": ents} label_in_text = set(l for l in dict_of_ents["ents"]) options = {"ents": label_in_text, "colors": get_random_color(label_in_text)} return dict_of_ents, options def get_retriever_annotations_candidates(text, ents): el_link_wrapper = " {}" # swap labels key with ents dict_of_ents = {"text": text, "ents": ents} label_in_text = set(l for l in dict_of_ents["ents"]) options = {"ents": label_in_text, "colors": get_random_color(label_in_text)} return dict_of_ents, options import json io_map = {} with open("/home/user/app/models/retriever/document_index/documents.jsonl", "r") as r: for line in r: element = json.loads(line) io_map[element["text"]] = element["metadata"]["type"] import json db_set = set() with open("models/retriever/intervention/gpt/db/document_index/documents.jsonl", "r") as r: for line in r: element = json.loads(line) db_set.add(element["text"]) with open("models/retriever/outcome/gpt/db/document_index/documents.jsonl", "r") as r: for line in r: element = json.loads(line) db_set.add(element["text"]) @st.cache_resource() def load_model(): retriever_question = GoldenRetriever( question_encoder="/home/user/app/models/retriever/question_encoder", document_index="/home/user/app/models/retriever/document_index/questions" ) retriever_intervention_gpt_taxonomy = GoldenRetriever( question_encoder="models/retriever/intervention/gpt/taxonomy/question_encoder", document_index="models/retriever/intervention/gpt/taxonomy/document_index" ) retriever_intervention_gpt_db = GoldenRetriever( question_encoder="models/retriever/intervention/gpt/db/question_encoder", document_index="models/retriever/intervention/gpt/db/document_index" ) retriever_outcome_gpt_taxonomy = GoldenRetriever( question_encoder="models/retriever/outcome/gpt/taxonomy/question_encoder", document_index="models/retriever/outcome/gpt/taxonomy/document_index" ) retriever_outcome_gpt_db = GoldenRetriever( question_encoder="models/retriever/outcome/gpt/db/question_encoder", document_index="models/retriever/outcome/gpt/db/document_index" ) reader = RelikReaderForSpanExtraction("/home/user/app/models/small-extended-large-batch", dataset_kwargs={"use_nme": True}) relik_question = Relik(reader=reader, retriever=retriever_question, window_size="none", top_k=100, task="span", device="cpu", document_index_device="cpu") return [relik_question, retriever_intervention_gpt_db, retriever_outcome_gpt_db, retriever_intervention_gpt_taxonomy, retriever_outcome_gpt_taxonomy] def set_intro(css): # intro st.markdown("# ImpactAI") st.image( "http://35.237.102.64/public/logo.png", ) st.markdown( "### 3ie taxonomy level 4 Intervention/Outcome candidate retriever with Entity Linking" ) # st.markdown( # "This is a front-end for the paper [Universal Semantic Annotator: the First Unified API " # "for WSD, SRL and Semantic Parsing](https://www.researchgate.net/publication/360671045_Universal_Semantic_Annotator_the_First_Unified_API_for_WSD_SRL_and_Semantic_Parsing), which will be presented at LREC 2022 by " # "[Riccardo Orlando](https://riccorl.github.io), [Simone Conia](https://c-simone.github.io/), " # "[Stefano Faralli](https://corsidilaurea.uniroma1.it/it/users/stefanofaralliuniroma1it), and [Roberto Navigli](https://www.diag.uniroma1.it/navigli/)." # ) def run_client(): with open(Path(__file__).parent / "style.css") as f: css = f.read() st.set_page_config( page_title="ImpactAI", page_icon="🦮", layout="wide", ) set_sidebar(css) set_intro(css) # Radio button selection analysis_type = st.radio( "Choose analysis type:", options=["Retriever", "Entity Linking"], index=0 # Default to 'question' ) selection_options = ["DB Intervention", "DB Outcome", "Top-k DB Intervention", "Top-k Soft DB Outcome", "Taxonomy Intervention", "Taxonomy Outcome"] if analysis_type == "Retriever": # Selection list using selectbox selection_list = st.selectbox( "Select an option:", options=selection_options ) # text input text = st.text_area( "Enter Text Below:", value="How does unconditional cash transfer affect to reduce poverty?", height=200, max_chars=1500, ) with stylable_container( key="annotate_button", css_styles=""" button { background-color: #a8ebff; color: black; border-radius: 25px; } """, ): submit = st.button("Annotate") # submit = st.button("Run") if "relik_model" not in st.session_state.keys(): st.session_state["relik_model"] = load_model() relik_model = st.session_state["relik_model"][0] # ReLik API call if submit: entity_linking_bool = False if analysis_type == "Entity Linking": relik_model = st.session_state["relik_model"][0] entity_linking_bool = True else: model_idx = selection_options.index(selection_list) relik_model = st.session_state["relik_model"][model_idx+1] text = text.strip() if text: st.markdown("####") with st.spinner(text="In progress"): if entity_linking_bool: response = relik_model(text) dict_of_ents, options = get_el_annotations(response=response) dict_of_ents_candidates, options_candidates = get_retriever_annotations(response=response) st.markdown("#### Entity Linking") display = displacy.render( dict_of_ents, manual=True, style="ent", options=options ) display = display.replace("\n", " ") # heurstic, prevents split of annotation decorations display = display.replace("border-radius: 0.35em;", "border-radius: 0.35em; white-space: nowrap;") with st.container(): st.write(display, unsafe_allow_html=True) candidate_text = "".join(f"
  • Intervention: {candidate}
  • " if io_map[candidate] == "intervention" else f"
  • Outcome: {candidate}
  • " for candidate in dict_of_ents_candidates["ents"][0:10]) text = """

    Possible Candidates:

    " st.markdown(text, unsafe_allow_html=True) else: if "Top-k" in selection_options: response = relik_model.retrieve(text, k=100, batch_size=400, progress_bar=False) candidates_text = [] for pred in response[0]: if pred.document.text in db_set: candidates_text.append(pred.document.text) else: response = relik_model.retrieve(text, k=10, batch_size=400, progress_bar=False) candidates_text = [] for pred in response[0]: candidates_text.append(pred.document.text) if candidates_text: candidates_text = candidates_text[:5] dict_of_ents_candidates, options_candidates = get_retriever_annotations_candidates(text, candidates_text) text = """

    Possible Candidates:

    " st.markdown(text, unsafe_allow_html=True) else: text = "

    No Candidates Found

    " else: st.error("Please enter some text.") if __name__ == "__main__": run_client()