import re import openai import pandas as pd import pinecone import spacy import streamlit_scrollable_textbox as stx import torch from sentence_transformers import SentenceTransformer from tqdm import tqdm from transformers import ( AutoModelForMaskedLM, AutoModelForSeq2SeqLM, AutoTokenizer, pipeline, ) import streamlit as st @st.experimental_singleton def get_data(): data = pd.read_csv("earnings_calls_cleaned_metadata.csv") return data # Initialize Spacy Model @st.experimental_singleton def get_spacy_model(): return spacy.load("en_core_web_sm") # Initialize models from HuggingFace @st.experimental_singleton def get_t5_model(): return pipeline("summarization", model="t5-small", tokenizer="t5-small") @st.experimental_singleton def get_flan_t5_model(): return pipeline( "summarization", model="google/flan-t5-small", tokenizer="google/flan-t5-small", max_length=512, # length_penalty = 0 ) @st.experimental_singleton def get_mpnet_embedding_model(): device = "cuda" if torch.cuda.is_available() else "cpu" model = SentenceTransformer( "sentence-transformers/all-mpnet-base-v2", device=device ) model.max_seq_length = 512 return model @st.experimental_singleton def get_splade_sparse_embedding_model(): model_sparse = "naver/splade-cocondenser-ensembledistil" # check device device = "cuda" if torch.cuda.is_available() else "cpu" tokenizer = AutoTokenizer.from_pretrained(model_sparse) model_sparse = AutoModelForMaskedLM.from_pretrained(model_sparse) # move to gpu if available model_sparse.to(device) return model_sparse, tokenizer @st.experimental_singleton def get_sgpt_embedding_model(): device = "cuda" if torch.cuda.is_available() else "cpu" model = SentenceTransformer( "Muennighoff/SGPT-125M-weightedmean-nli-bitfit", device=device ) model.max_seq_length = 512 return model @st.experimental_memo def save_key(api_key): return api_key def create_dense_embeddings(query, model): dense_emb = model.encode([query]).tolist() return dense_emb def create_sparse_embeddings(query, model, tokenizer): device = "cuda" if torch.cuda.is_available() else "cpu" inputs = tokenizer(query, return_tensors="pt").to(device) with torch.no_grad(): logits = model(**inputs).logits inter = torch.log1p(torch.relu(logits[0])) token_max = torch.max(inter, dim=0) # sum over input tokens nz_tokens = torch.where(token_max.values > 0)[0] nz_weights = token_max.values[nz_tokens] order = torch.sort(nz_weights, descending=True) nz_weights = nz_weights[order[1]] nz_tokens = nz_tokens[order[1]] return { "indices": nz_tokens.cpu().numpy().tolist(), "values": nz_weights.cpu().numpy().tolist(), } def hybrid_score_norm(dense, sparse, alpha: float): """Hybrid score using a convex combination alpha * dense + (1 - alpha) * sparse Args: dense: Array of floats representing sparse: a dict of `indices` and `values` alpha: scale between 0 and 1 """ if alpha < 0 or alpha > 1: raise ValueError("Alpha must be between 0 and 1") hs = { "indices": sparse["indices"], "values": [v * (1 - alpha) for v in sparse["values"]], } return [v * alpha for v in dense], hs def query_pinecone_sparse( dense_vec, sparse_vec, top_k, index, year, quarter, ticker, participant_type, threshold=0.25, ): if participant_type == "Company Speaker": participant = "Answer" else: participant = "Question" if year == "All": if quarter == "All": xc = index.query( vector=dense_vec, sparse_vector=sparse_vec, top_k=top_k, filter={ "Year": { "$in": [ int("2020"), int("2019"), int("2018"), int("2017"), int("2016"), ] }, "Quarter": {"$in": ["Q1", "Q2", "Q3", "Q4"]}, "Ticker": {"$eq": ticker}, "QA_Flag": {"$eq": participant}, }, include_metadata=True, ) else: xc = index.query( vector=dense_vec, sparse_vector=sparse_vec, top_k=top_k, filter={ "Year": { "$in": [ int("2020"), int("2019"), int("2018"), int("2017"), int("2016"), ] }, "Quarter": {"$eq": quarter}, "Ticker": {"$eq": ticker}, "QA_Flag": {"$eq": participant}, }, include_metadata=True, ) else: # search pinecone index for context passage with the answer xc = index.query( vector=dense_vec, sparse_vector=sparse_vec, top_k=top_k, filter={ "Year": int(year), "Quarter": {"$eq": quarter}, "Ticker": {"$eq": ticker}, "QA_Flag": {"$eq": participant}, }, include_metadata=True, ) # filter the context passages based on the score threshold filtered_matches = [] for match in xc["matches"]: if match["score"] >= threshold: filtered_matches.append(match) xc["matches"] = filtered_matches return xc def query_pinecone( dense_vec, top_k, index, year, quarter, ticker, participant_type, threshold=0.25, ): if participant_type == "Company Speaker": participant = "Answer" else: participant = "Question" if year == "All": if quarter == "All": xc = index.query( vector=dense_vec, top_k=top_k, filter={ "Year": { "$in": [ int("2020"), int("2019"), int("2018"), int("2017"), int("2016"), ] }, "Quarter": {"$in": ["Q1", "Q2", "Q3", "Q4"]}, "Ticker": {"$eq": ticker}, "QA_Flag": {"$eq": participant}, }, include_metadata=True, ) else: xc = index.query( vector=dense_vec, top_k=top_k, filter={ "Year": { "$in": [ int("2020"), int("2019"), int("2018"), int("2017"), int("2016"), ] }, "Quarter": {"$eq": quarter}, "Ticker": {"$eq": ticker}, "QA_Flag": {"$eq": participant}, }, include_metadata=True, ) else: # search pinecone index for context passage with the answer xc = index.query( vector=dense_vec, top_k=top_k, filter={ "Year": int(year), "Quarter": {"$eq": quarter}, "Ticker": {"$eq": ticker}, "QA_Flag": {"$eq": participant}, }, include_metadata=True, ) # filter the context passages based on the score threshold filtered_matches = [] for match in xc["matches"]: if match["score"] >= threshold: filtered_matches.append(match) xc["matches"] = filtered_matches return xc def format_query(query_results): # extract passage_text from Pinecone search result context = [ result["metadata"]["Text"] for result in query_results["matches"] ] return context def sentence_id_combine(data, query_results, lag=1): # Extract sentence IDs from query results ids = [ result["metadata"]["Sentence_id"] for result in query_results["matches"] ] # Generate new IDs by adding a lag value to the original IDs new_ids = [id + i for id in ids for i in range(-lag, lag + 1)] # Remove duplicates and sort the new IDs new_ids = sorted(set(new_ids)) # Create a list of lookup IDs by grouping the new IDs in groups of lag*2+1 lookup_ids = [ new_ids[i : i + (lag * 2 + 1)] for i in range(0, len(new_ids), lag * 2 + 1) ] # Create a list of context sentences by joining the sentences corresponding to the lookup IDs context_list = [ " ".join( data.loc[data["Sentence_id"].isin(lookup_id), "Text"].to_list() ) for lookup_id in lookup_ids ] return context_list def text_lookup(data, sentence_ids): context = ". ".join(data.iloc[sentence_ids].to_list()) return context def generate_gpt_prompt(query_text, context_list): context = " ".join(context_list) prompt = f"""Answer the question in 6 long detailed points as accurately as possible using the provided context. Include as many key details as possible. Context: {context} Question: {query_text} Answer:""" return prompt def generate_gpt_prompt_2(query_text, context_list): context = " ".join(context_list) prompt = f""" Context information is below: --------------------- {context} --------------------- Given the context information and prior knowledge, answer this question: {query_text} Try to include as many key details as possible and format the answer in points.""" return prompt def generate_flant5_prompt(query_text, context_list): context = " \n".join(context_list) prompt = f"""Given the context information and prior knowledge, answer this question: {query_text} Context information is below: --------------------- {context} ---------------------""" return prompt def get_context_list_prompt(prompt): prompt_list = prompt.split("---------------------") context = prompt_list[-2].strip() context_list = context.split(" \n") return context_list def gpt_model(prompt): response = openai.Completion.create( model="text-davinci-003", prompt=prompt, temperature=0, max_tokens=1024, ) return response.choices[0].text # Entity Extraction def extract_quarter_year(string): # Extract year from string year_match = re.search(r"\d{4}", string) if year_match: year = year_match.group() else: return None, None # Extract quarter from string quarter_match = re.search(r"Q\d", string) if quarter_match: quarter = "Q" + quarter_match.group()[1] else: return None, None return quarter, year def extract_entities(query, model): doc = model(query) entities = {ent.label_: ent.text for ent in doc.ents} if "ORG" in entities.keys(): company = entities["ORG"].lower() if "DATE" in entities.keys(): quarter, year = extract_quarter_year(entities["DATE"]) return company, quarter, year else: return company, None, None else: if "DATE" in entities.keys(): quarter, year = extract_quarter_year(entities["DATE"]) return None, quarter, year else: return None, None, None def clean_entities(company, quarter, year): company_ticker_map = { "apple": "AAPL", "amd": "AMD", "amazon": "AMZN", "cisco": "CSCO", "google": "GOOGL", "microsoft": "MSFT", "nvidia": "NVDA", "asml": "ASML", "intel": "INTC", "micron": "MU", } ticker_choice = [ "AAPL", "CSCO", "MSFT", "ASML", "NVDA", "GOOGL", "MU", "INTC", "AMZN", "AMD", ] year_choice = ["2020", "2019", "2018", "2017", "2016", "All"] quarter_choice = ["Q1", "Q2", "Q3", "Q4", "All"] if company is not None: if company in company_ticker_map.keys(): ticker = company_ticker_map[company] ticker_index = ticker_choice.index(ticker) else: ticker_index = 0 else: ticker_index = 0 if quarter is not None: if quarter in quarter_choice: quarter_index = quarter_choice.index(quarter) else: quarter_index = len(quarter_choice) - 1 else: quarter_index = len(quarter_choice) - 1 if year is not None: if year in year_choice: year_index = year_choice.index(year) else: year_index = len(year_choice) - 1 else: year_index = len(year_choice) - 1 return ticker_index, quarter_index, year_index # Transcript Retrieval def retrieve_transcript(data, year, quarter, ticker): if year == "All" or quarter == "All": row = ( data.loc[ (data.Ticker == ticker), ["File_Name"], ] .drop_duplicates() .iloc[0, 0] ) else: row = ( data.loc[ (data.Year == int(year)) & (data.Quarter == quarter) & (data.Ticker == ticker), ["File_Name"], ] .drop_duplicates() .iloc[0, 0] ) # convert row to a string and join values with "-" # row_str = "-".join(row.astype(str)) + ".txt" open_file = open( f"Transcripts/{ticker}/{row}", "r", ) file_text = open_file.read() return file_text