import streamlit as st import os import json import fitz import re from transformers import GPT2Tokenizer, GPT2LMHeadModel, AutoModelForSequenceClassification, BertTokenizer, BertModel,T5Tokenizer, T5ForConditionalGeneration,AutoTokenizer, AutoModelForSeq2SeqLM import torch from sklearn.metrics.pairwise import cosine_similarity import numpy as np import nltk from nltk.tokenize import sent_tokenize from nltk.corpus import stopwords def is_new_file_upload(uploaded_file): if 'last_uploaded_file' in st.session_state: # Check if the newly uploaded file is different from the last one if (uploaded_file.name != st.session_state.last_uploaded_file['name'] or uploaded_file.size != st.session_state.last_uploaded_file['size']): st.session_state.last_uploaded_file = {'name': uploaded_file.name, 'size': uploaded_file.size} # st.write("A new src image file has been uploaded.") return True else: # st.write("The same src image file has been re-uploaded.") return False else: # st.write("This is the first file upload detected.") st.session_state.last_uploaded_file = {'name': uploaded_file.name, 'size': uploaded_file.size} return True def add_commonality_to_similarity_score(similarity, sentence, query): # Tokenize both the sentence and the query # sentence_words = set(sentence.split()) # query_words = set(query.split()) sentence_words = set(word for word in sentence.split() if word.lower() not in st.session_state.stop_words) query_words = set(word for word in query.split() if word.lower() not in st.session_state.stop_words) # Calculate the number of common words common_words = len(sentence_words.intersection(query_words)) # Adjust the similarity score with the common words count combined_score = similarity + (common_words / max(len(query_words), 1)) # Normalize by the length of the query to keep the score between -1 and 1 return combined_score,similarity,(common_words / max(len(query_words), 1)) def contradiction_detection(premise,hypothesis): inputs = st.session_state.roberta_tokenizer.encode_plus(premise, hypothesis, return_tensors="pt", truncation=True) # Get model predictions outputs = st.session_state.roberta_model(**inputs) # Get the logits (raw predictions before softmax) logits = outputs.logits # Apply softmax to get probabilities for each class probabilities = torch.softmax(logits, dim=1) # Class labels: 0 = entailment, 1 = neutral, 2 = contradiction predicted_class = torch.argmax(probabilities, dim=1).item() # Class labels labels = ["Contradiction", "Neutral", "Entailment"] # Output the result print(f"Prediction: {labels[predicted_class]}") return {labels[predicted_class]} if 'is_initialized' not in st.session_state: st.session_state['is_initialized'] = True nltk.download('punkt') nltk.download('stopwords') # print("stop words start") # print(stopwords.words('english')) # print("stop words end") stop_words_list = stopwords.words('english') st.session_state.stop_words = set(stop_words_list) st.session_state.bert_tokenizer = BertTokenizer.from_pretrained("bert-base-uncased", ) st.session_state.bert_model = BertModel.from_pretrained("bert-base-uncased", ).to('cuda') st.session_state.roberta_tokenizer = AutoTokenizer.from_pretrained("roberta-large-mnli") st.session_state.roberta_model = AutoModelForSequenceClassification.from_pretrained("roberta-large-mnli") def encode_sentence(sentence): if len(sentence.strip()) < 4: return None sentence_tokens = st.session_state.bert_tokenizer(sentence, return_tensors="pt", padding=True, truncation=True).to( 'cuda') with torch.no_grad(): sentence_encoding = st.session_state.bert_model(**sentence_tokens).last_hidden_state[:, 0, :].cpu().numpy() return sentence_encoding def encode_paragraph(paragraph): sentence_encodings = [] paragraph_without_newline = paragraph.replace("\n", "") sentences = sent_tokenize(paragraph_without_newline) for sentence in sentences: # if sentence.strip().endswith('?'): # sentence_encodings.append(None) # continue sentence_encoding = encode_sentence(sentence) sentence_encodings.append([sentence, sentence_encoding]) return sentence_encodings if 'list_count' in st.session_state: st.write(f'The number of elements at the top level of the hierarchy: {st.session_state.list_count }') if 'paragraph_sentence_encodings' not in st.session_state: print("start embedding paragarphs") read_progress_bar = st.progress(0) st.session_state.paragraph_sentence_encodings = [] for index,paragraph in enumerate(st.session_state.restored_paragraphs): #print(paragraph) progress_percentage = (index) / (st.session_state.list_count - 1) # print(progress_percentage) read_progress_bar.progress(progress_percentage) # sentence_encodings.append([sentence,bert_model(**sentence_tokens).last_hidden_state[:, 0, :].detach().numpy()]) sentence_encodings=encode_paragraph(paragraph['paragraph']) st.session_state.paragraph_sentence_encodings.append([paragraph, sentence_encodings]) st.rerun() big_text = """

Contradiction Dectection

""" # Display the styled text st.markdown(big_text, unsafe_allow_html=True) def convert_pdf_to_paragraph_list(doc): paragraphs = [] sentence_endings = ('.', '!', '?') start_page = 1 for page_num in range(start_page - 1, len(doc)): # start_page - 1 to adjust for 0-based index page = doc.load_page(page_num) blocks = page.get_text("blocks") block_index = 1 for block in blocks: x0, y0, x1, y1, text, block_type, flags = block if text.strip() != "": text = text.strip() text = re.sub(r'\n\s+\n', '\n\n', text) list_pattern = re.compile(r'^\s*((?:\d+\.|[a-zA-Z]\.|[*-])\s+.+)', re.MULTILINE) match = list_pattern.search(text) containsList = False if match: containsList = True # print ("list detected") paragraph = "" if bool(re.search(r'\n{2,}', text)): substrings = re.split(r'\n{2,}', text) for substring in substrings: if substring.strip() != "": paragraph = substring paragraphs.append( {"paragraph": paragraph, "containsList": containsList, "page_num": page_num, "text": text}); # print(f" {substring} ") else: paragraph = text paragraphs.append( {"paragraph": paragraph, "containsList": containsList, "page_num": page_num, "text": None}); return paragraphs uploaded_pdf_file = st.file_uploader("Upload a PDF file", type=['pdf']) st.markdown( f'Sample Master PDF download and then upload to above', unsafe_allow_html=True) st.markdown("sample queries to invoke contradiction:
A Member shall be deemed disabled under this provision if, due to illness or injury, the Member is unable to safely and fully carry out two or more Activities of Daily Living without the assistance or verbal prompting of another individual.",unsafe_allow_html=True) st.markdown( f'Sample Secondary txt download and then upload to above', unsafe_allow_html=True) if uploaded_pdf_file is not None: if is_new_file_upload(uploaded_pdf_file): print("is new file uploaded") if 'prev_query' in st.session_state: del st.session_state['prev_query'] if 'paragraph_sentence_encodings' in st.session_state: del st.session_state['paragraph_sentence_encodings'] save_path = './uploaded_files' if not os.path.exists(save_path): os.makedirs(save_path) with open(os.path.join(save_path, uploaded_pdf_file.name), "wb") as f: f.write(uploaded_pdf_file.getbuffer()) # Write the file to the specified location st.success(f'Saved file temp_{uploaded_pdf_file.name} in {save_path}') st.session_state.uploaded_path=os.path.join(save_path, uploaded_pdf_file.name) # st.session_state.page_count = utils.get_pdf_page_count(st.session_state.uploaded_pdf_path) # print("page_count=",st.session_state.page_count) doc = fitz.open(st.session_state.uploaded_path) st.session_state.restored_paragraphs=convert_pdf_to_paragraph_list(doc) if isinstance(st.session_state.restored_paragraphs, list): # Count the restored_paragraphs of top-level elements st.session_state.list_count = len(st.session_state.restored_paragraphs) st.write(f'The number of elements at the top level of the hierarchy: {st.session_state.list_count}') st.rerun() def find_sentences_scores(paragraph_sentence_encodings, query_encoding, processing_progress_bar,total_count): sentence_scores = [] for index, paragraph_sentence_encoding in enumerate(paragraph_sentence_encodings): progress_percentage = index / (total_count - 1) processing_progress_bar.progress(progress_percentage) sentence_similarities = [] for sentence_encoding in paragraph_sentence_encoding[1]: if sentence_encoding: similarity = cosine_similarity(query_encoding, sentence_encoding[1])[0][0] combined_score, similarity_score, commonality_score = add_commonality_to_similarity_score(similarity, sentence_encoding[0], query) sentence_similarities.append((combined_score, sentence_encoding[0], commonality_score)) sentence_scores.append((combined_score, sentence_encoding[0])) sentence_similarities.sort(reverse=True, key=lambda x: x[0]) # print(sentence_similarities) if len(sentence_similarities) >= 3: top_three_avg_similarity = np.mean([s[0] for s in sentence_similarities[:3]]) top_three_avg_commonality = np.mean([s[2] for s in sentence_similarities[:3]]) top_three_sentences = sentence_similarities[:3] elif sentence_similarities: top_three_avg_similarity = np.mean([s[0] for s in sentence_similarities]) top_three_avg_commonality = np.mean([s[2] for s in sentence_similarities]) top_three_sentences = sentence_similarities else: top_three_avg_similarity = 0 top_three_avg_commonality = 0 top_three_sentences = [] # print(f"top_three_sentences={top_three_sentences}") # top_three_texts = [s[1] for s in top_three_sentences] # remaining_texts = [s[0] for s in paragraph_sentence_encoding[1] if s and s[0] not in top_three_texts] # reordered_paragraph = top_three_texts + remaining_texts # # original_paragraph = ' '.join([s[0] for s in paragraph_sentence_encoding[1] if s]) # modified_paragraph = ' '.join(reordered_paragraph) paragraph_scores.append( (top_three_avg_similarity, top_three_avg_commonality, {'top_three_sentences': top_three_sentences, 'original_text': paragraph_sentence_encoding[0]}) ) sentence_scores = sorted(sentence_scores, key=lambda x: x[0], reverse=True) if 'paragraph_sentence_encodings' in st.session_state: query = st.text_input("Enter your query") if query: if 'prev_query' not in st.session_state or st.session_state.prev_query != query: st.session_state.prev_query = query st.session_state.premise = query query_encoding = encode_sentence(query) paragraph_scores = [] total_count = len(st.session_state.paragraph_sentence_encodings) processing_progress_bar = st.progress(0) sentence_scores = find_sentences_scores( st.session_state.paragraph_sentence_encodings, query_encoding, processing_progress_bar,total_count) st.session_state.paragraph_scores = sorted(paragraph_scores, key=lambda x: x[0], reverse=True) if 'paragraph_scores' in st.session_state: st.write("Top scored paragraphs and their scores:") for i, (similarity_score, commonality_score, paragraph) in enumerate( st.session_state.paragraph_scores[:3]): #number of paragraphs to consider #st.write("top_three_sentences: ", paragraph['top_three_sentences']) st.write("paragarph number ***", i) prev_contradiction_detected =True for top_sentence in paragraph['top_three_sentences']: if prev_contradiction_detected: contradiction_detection_result =contradiction_detection(st.session_state.premise,top_sentence[1]) if contradiction_detection_result == {"Contradiction"}: st.write("master document page number ", paragraph['original_text']['page_num']) st.write("master document sentence: ", top_sentence[1]) st.write("secondary document sentence: ", st.session_state.premise) st.write(contradiction_detection_result) # st.write(contradiction_detection(st.session_state.premise, top_sentence[1])) else: prev_contradiction_detected = False else: break #print(top_sentence[1]) # st.write(f"Similarity Score: {similarity_score}, Commonality Score: {commonality_score}") # st.write("top_three_sentences: ", paragraph['top_three_sentences']) #st.write("Original Paragraph: ", paragraph['original_text']) #A Member will be considered Actively at Work if he or she is able and available for active performance of all of his or her regular duties # A Member will be considered as inactive at Work if he or she is able and available for active performance of all of his or her regular duties #A Member shall be deemed inactive at Work if he or she is capable and available to perform all of his or her regular responsibilities. # st.write("Modified Paragraph: ", paragraph['modified_text'])