Spaces:
Runtime error
Runtime error
| from haystack.nodes import TfidfRetriever | |
| from haystack.document_stores import InMemoryDocumentStore | |
| import spacy | |
| import re | |
| from spacy.matcher import Matcher | |
| from markdown import markdown | |
| from annotated_text import annotation | |
| from haystack.schema import Document | |
| from typing import List, Text, Tuple | |
| from typing_extensions import Literal | |
| from utils.preprocessing import processingpipeline | |
| from utils.streamlitcheck import check_streamlit | |
| import logging | |
| try: | |
| from termcolor import colored | |
| except: | |
| pass | |
| try: | |
| import streamlit as st | |
| except ImportError: | |
| logging.info("Streamlit not installed") | |
| def runLexicalPreprocessingPipeline(file_name:str,file_path:str, | |
| split_by: Literal["sentence", "word"] = 'word', | |
| split_length:int = 80, split_overlap:int = 0, | |
| remove_punc:bool = False,)->List[Document]: | |
| """ | |
| creates the pipeline and runs the preprocessing pipeline, | |
| the params for pipeline are fetched from paramconfig. As lexical doesnt gets | |
| affected by overlap, threfore split_overlap = 0 in default paramconfig and | |
| split_by = word. | |
| Params | |
| ------------ | |
| file_name: filename, in case of streamlit application use | |
| st.session_state['filename'] | |
| file_path: filepath, in case of streamlit application use | |
| st.session_state['filepath'] | |
| split_by: document splitting strategy either as word or sentence | |
| split_length: when synthetically creating the paragrpahs from document, | |
| it defines the length of paragraph. | |
| split_overlap: Number of words or sentences that overlap when creating | |
| the paragraphs. This is done as one sentence or 'some words' make sense | |
| when read in together with others. Therefore the overlap is used. | |
| splititng of text. | |
| removePunc: to remove all Punctuation including ',' and '.' or not | |
| Return | |
| -------------- | |
| List[Document]: When preprocessing pipeline is run, the output dictionary | |
| has four objects. For the lexicaal search using TFIDFRetriever we | |
| need to use the List of Haystack Document, which can be fetched by | |
| key = 'documents' on output. | |
| """ | |
| lexical_processing_pipeline = processingpipeline() | |
| output_lexical_pre = lexical_processing_pipeline.run(file_paths = file_path, | |
| params= {"FileConverter": {"file_path": file_path, \ | |
| "file_name": file_name}, | |
| "UdfPreProcessor": {"remove_punc": remove_punc, \ | |
| "split_by": split_by, \ | |
| "split_length":split_length,\ | |
| "split_overlap": split_overlap}}) | |
| return output_lexical_pre | |
| def tokenize_lexical_query(query:str)-> List[str]: | |
| """ | |
| Removes the stop words from query and returns the list of important keywords | |
| in query. For the lexical search the relevent paragraphs in document are | |
| retreived using TfIDFretreiver from Haystack. However to highlight these | |
| keywords we need the tokenized form of query. | |
| Params | |
| -------- | |
| query: string which represents either list of keywords user is looking for | |
| or a query in form of Question. | |
| Return | |
| ----------- | |
| token_list: list of important keywords in the query. | |
| """ | |
| nlp = spacy.load("en_core_web_sm") | |
| token_list = [token.text.lower() for token in nlp(query) | |
| if not (token.is_stop or token.is_punct)] | |
| return token_list | |
| def runSpacyMatcher(token_list:List[str], document:Text | |
| )->Tuple[List[List[int]],spacy.tokens.doc.Doc]: | |
| """ | |
| Using the spacy in backend finds the keywords in the document using the | |
| Matcher class from spacy. We can alternatively use the regex, but spacy | |
| finds all keywords in serialized manner which helps in annotation of answers. | |
| Params | |
| ------- | |
| token_list: this is token list which tokenize_lexical_query function returns | |
| document: text in which we need to find the tokens | |
| Return | |
| -------- | |
| matches: List of [start_index, end_index] in the spacydoc(at word level not | |
| character) for the keywords in token list. | |
| spacydoc: the keyword index in the spacydoc are at word level and not character, | |
| therefore to allow the annotator to work seamlessly we return the spacydoc. | |
| """ | |
| nlp = spacy.load("en_core_web_sm") | |
| spacydoc = nlp(document) | |
| matcher = Matcher(nlp.vocab) | |
| token_pattern = [[{"LOWER":token}] for token in token_list] | |
| matcher.add(",".join(token_list), token_pattern) | |
| spacymatches = matcher(spacydoc) | |
| # getting start and end index in spacydoc so that annotator can work seamlessly | |
| matches = [] | |
| for match_id, start, end in spacymatches: | |
| matches = matches + [[start, end]] | |
| return matches, spacydoc | |
| def runRegexMatcher(token_list:List[str], document:Text): | |
| """ | |
| Using the regex in backend finds the keywords in the document. | |
| Params | |
| ------- | |
| token_list: this is token list which tokenize_lexical_query function returns | |
| document: text in which we need to find the tokens | |
| Return | |
| -------- | |
| matches: List of [start_index, end_index] in the document for the keywords | |
| in token list at character level. | |
| document: the keyword index returned by regex are at character level, | |
| therefore to allow the annotator to work seamlessly we return the text back. | |
| """ | |
| matches = [] | |
| for token in token_list: | |
| matches = (matches + | |
| [[val.start(), val.start() + | |
| len(token)] for val in re.finditer(token, document)]) | |
| return matches, document | |
| def spacyAnnotator(matches: List[List[int]], document:spacy.tokens.doc.Doc): | |
| """ | |
| This is spacy Annotator and needs spacy.doc | |
| Annotates the text in the document defined by list of [start index, end index] | |
| Example: "How are you today", if document type is text, matches = [[0,3]] | |
| will give answer = "How", however in case we used the spacy matcher then the | |
| matches = [[0,3]] will give answer = "How are you". However if spacy is used | |
| to find "How" then the matches = [[0,1]] for the string defined above. | |
| Params | |
| ----------- | |
| matches: As mentioned its list of list. Example [[0,1],[10,13]] | |
| document: document which needs to be indexed. | |
| Return | |
| -------- | |
| will send the output to either app front end using streamlit or | |
| write directly to output screen. | |
| """ | |
| start = 0 | |
| annotated_text = "" | |
| for match in matches: | |
| start_idx = match[0] | |
| end_idx = match[1] | |
| if check_streamlit(): | |
| annotated_text = (annotated_text + document[start:start_idx].text | |
| + str(annotation(body=document[start_idx:end_idx].text, | |
| label="ANSWER", background="#964448", color='#ffffff'))) | |
| else: | |
| annotated_text = (annotated_text + document[start:start_idx].text | |
| + colored(document[start_idx:end_idx].text, | |
| "green", attrs = ['bold'])) | |
| start = end_idx | |
| annotated_text = annotated_text + document[end_idx:].text | |
| if check_streamlit(): | |
| st.write( | |
| markdown(annotated_text), | |
| unsafe_allow_html=True, | |
| ) | |
| else: | |
| print(annotated_text) | |
| def lexical_search(query:Text, documents:List[Document],top_k:int): | |
| """ | |
| Performs the Lexical search on the List of haystack documents which is | |
| returned by preprocessing Pipeline. | |
| Params | |
| ------- | |
| query: Keywords that need to be searche in documents. | |
| documents: List of Haystack documents returned by preprocessing pipeline. | |
| top_k: Number of Top results to be fetched. | |
| """ | |
| document_store = InMemoryDocumentStore() | |
| document_store.write_documents(documents) | |
| # Haystack Retriever works with document stores only. | |
| retriever = TfidfRetriever(document_store) | |
| results = retriever.retrieve(query=query, top_k = top_k) | |
| query_tokens = tokenize_lexical_query(query) | |
| flag = True | |
| for count, result in enumerate(results): | |
| matches, doc = runSpacyMatcher(query_tokens,result.content) | |
| if len(matches) != 0: | |
| if flag: | |
| flag = False | |
| if check_streamlit(): | |
| st.markdown("##### Top few lexical search (TFIDF) hits #####") | |
| else: | |
| print("Top few lexical search (TFIDF) hits") | |
| if check_streamlit(): | |
| st.write("Result {}".format(count+1)) | |
| else: | |
| print("Results {}".format(count +1)) | |
| spacyAnnotator(matches, doc) | |
| if flag: | |
| if check_streamlit(): | |
| st.info("🤔 No relevant result found. Please try another keyword.") | |
| else: | |
| print("No relevant result found. Please try another keyword.") |