Spaces:

nickmuchi
/

Earnings-Call-Analysis-Whisperer

Running

App Files Files Community

Earnings-Call-Analysis-Whisperer / pages /3_Earnings_Semantic_Search_🔎_.py

nickmuchi

Update pages/3_Earnings_Semantic_Search_🔎_.py

446f9c9 over 1 year ago

raw history blame

No virus

4.63 kB

	import streamlit as st
	from functions import *

	st.set_page_config(page_title="Earnings Semantic Search", page_icon="🔎")
	st.sidebar.header("Semantic Search")
	st.markdown("## Earnings Semantic Search with SBert")

	def gen_sentiment(text):
	'''Generate sentiment of given text'''
	return sent_pipe(text)[0]['label']

	bi_enc_dict = {'mpnet-base-v2':"all-mpnet-base-v2",
	'e5-base':'intfloat/e5-base',
	'instructor-base': 'hkunlp/instructor-base',
	'mpnet-base-dot-v1':'multi-qa-mpnet-base-dot-v1',
	'setfit-finance': 'nickmuchi/setfit-finetuned-financial-text-classification'}

	search_input = st.text_input(
	label='Enter Your Search Query',value= "What key challenges did the business face?", key='search')

	sbert_model_name = st.sidebar.selectbox("Embedding Model", options=list(bi_enc_dict.keys()), key='sbox')

	top_k = 2

	window_size = st.sidebar.slider("Number of Sentences Generated in Search Response",min_value=1,max_value=7,value=3)

	try:

	if search_input:

	if "sen_df" in st.session_state and "earnings_passages" in st.session_state:

	## Save to a dataframe for ease of visualization
	sen_df = st.session_state['sen_df']

	passages = chunk_long_text(st.session_state['earnings_passages'],150,window_size=window_size)

	with st.spinner(
	text=f"Loading {bi_enc_dict[sbert_model_name]} encoder model..."
	):
	sbert = load_sbert(bi_enc_dict[sbert_model_name])


	##### Sematic Search #####
	# Encode the query using the bi-encoder and find potentially relevant passages
	# corpus_embeddings = sbert.encode(passages, convert_to_tensor=True, show_progress_bar=True)
	# question_embedding = sbert.encode(search_input, convert_to_tensor=True)
	# question_embedding = question_embedding.cpu()
	# hits = util.semantic_search(question_embedding, corpus_embeddings, top_k=top_k)
	# hits = hits[0] # Get the hits for the first query

	# ##### Re-Ranking #####
	# # Now, score all retrieved passages with the cross_encoder
	# cross_inp = [[search_input, passages[hit['corpus_id']]] for hit in hits]
	# cross_scores = cross_encoder.predict(cross_inp)

	# # Sort results by the cross-encoder scores
	# for idx in range(len(cross_scores)):
	# hits[idx]['cross-score'] = cross_scores[idx]

	# # Output of top-3 hits from re-ranker
	# hits = sorted(hits, key=lambda x: x['cross-score'], reverse=True)
	embedding_model = bi_enc_dict[sbert_model_name]

	hits = embed_text(search_input,passages,embedding_model)

	score='cross-score'
	df = pd.DataFrame([(hit[score],passages[hit['corpus_id']]) for hit in hits[0:int(top_k)]],columns=['Score','Text'])
	df['Score'] = round(df['Score'],2)
	df['Sentiment'] = df.Text.apply(gen_sentiment)

	def gen_annotated_text(df):
	'''Generate annotated text'''

	tag_list=[]
	for row in df.itertuples():
	label = row[3]
	text = row[2]
	if label == 'Positive':
	tag_list.append((text,label,'#8fce00'))
	elif label == 'Negative':
	tag_list.append((text,label,'#f44336'))
	else:
	tag_list.append((text,label,'#000000'))

	return tag_list

	text_annotations = gen_annotated_text(df)

	first, second = text_annotations[0], text_annotations[1]


	with st.expander(label='Best Search Query Result', expanded=True):
	annotated_text(first)

	with st.expander(label='Alternative Search Query Result'):
	annotated_text(second)

	else:

	st.write('Please ensure you have entered the YouTube URL or uploaded the Earnings Call file')

	else:

	st.write('Please ensure you have entered the YouTube URL or uploaded the Earnings Call file')

	except RuntimeError:

	st.write('Please ensure you have entered the YouTube URL or uploaded the Earnings Call file')