import streamlit as st import random, string import os import os.path import requests from os import listdir from os.path import isfile, join from groq import Groq from pinecone import Pinecone from sentence_transformers import SentenceTransformer from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.chains.conversation.memory import ConversationBufferWindowMemory from langchain_groq import ChatGroq from langchain.chains import LLMChain from langchain_core.prompts import ( ChatPromptTemplate, HumanMessagePromptTemplate, MessagesPlaceholder, ) from transformers import pipeline GROQ_API_KEY = st.secrets['GROQ_API_KEY'] PINECONE_API_KEY = st.secrets['PINECONE_API_KEY'] # Initialize Groq client client = Groq(api_key = GROQ_API_KEY) # Initialize Pinecone pc = Pinecone(api_key = PINECONE_API_KEY) # Create or connect to an existing index index = pc.Index("audio-sample") # Load the pre-trained sentiment analysis model sentiment_analysis = pipeline( "sentiment-analysis", model="distilbert-base-uncased-finetuned-sst-2-english") if 'sentiment_st' not in st.session_state: st.session_state.sentiment_st = '' if 'chat_list' not in st.session_state: st.session_state.chat_list = [] if 'body' not in st.session_state: st.session_state.body = '' if 'processing' not in st.session_state: st.session_state.processing = "processing..." if 'memory' not in st.session_state: st.session_state.memory = ConversationBufferWindowMemory(k=5, memory_key="chat_history", return_messages=True) if 'namespace' not in st.session_state: st.session_state.namespace = "Ans_"+''.join(random.choice(string.ascii_uppercase + string.ascii_lowercase + string.digits) for _ in range(5)) # em_model = SentenceTransformer("all-MiniLM-L6-v2") model = SentenceTransformer('all-mpnet-base-v2') def randomIdGenerate(): ran = ''.join(random.choices(string.ascii_uppercase + string.digits, k = 5)) return ran def readFiles(audio): st.session_state.processing = "Processing files..." translation = client.audio.translations.create( file=audio, model="whisper-large-v3", ) st.session_state.body = translation.text splits = get_text_chunks(translation.text) # st.write("splits:") # st.write(splits) # print(splits) emb = embedThetext(splits) # st.write("emb:") # st.write(emb) saveInPinecone(emb) return splits def get_text_chunks(text): st.session_state.processing = "Text to chunks..." text_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=200) chunks = text_splitter.split_text(text) return chunks def embedThetext(text): st.session_state.processing = "Embedding text..." vectors = [] if text: embeddings = model.encode(text) metadata_list = [{"text": s} for s in text] ids = [f'id-{randomIdGenerate()}' for i in range(len(text))] vectors = [ {'id': id_, 'values': embedding, 'metadata': metadata} for id_, embedding, metadata in zip(ids, embeddings, metadata_list) ] return vectors def saveInPinecone(vector): st.session_state.processing = "Inserting to prinecone vector..." if vector: index.upsert( vectors = vector, namespace=st.session_state.namespace ) def get_query_embdedding(embed): query_embedding = model.encode([embed]).tolist() return query_embedding def chk_sentiment_result(result): pos = 0 neg = 0 nut = 0 resp = '' for x in result: if(x['label'] == "POSITIVE"): pos = pos + 1 elif(x['label'] == "NEGATIVE"): neg = neg + 1 else: nut = nut + 1 # if (pos >= neg and pos >= nut): # resp = 1 # elif (neg >= pos and neg >= nut): # resp = 2 resp = 'POSITIVE = ' + str(pos) + ', NEGATIVE = ' + str(neg) + ', NEUTRAL = ' + str(nut) return resp st.title('Create Summary From Audio Files') uploaded_files = st.file_uploader("Choose a Audio file") button = st.button("Upload file to Process", key="process_but") st.divider() audio_txt = '' if button: if uploaded_files: with st.spinner(st.session_state.processing): audio_txt = readFiles(uploaded_files) st.success('Audio Processed Successfully') else: st.error('No files selected') if audio_txt: result = sentiment_analysis(audio_txt) rl = chk_sentiment_result(result) st.session_state.sentiment_st = rl if st.session_state.body: st.title('Text From Audio Files:') with st.chat_message("machine"): st.write(st.session_state.body) st.divider() if st.session_state.sentiment_st: st.title('Sentiment of Audio Files:') st.info(st.session_state.sentiment_st) # st.write('Sentiment not analysed') st.divider() # Define the query query = st.chat_input("Enter Your Summarize Query?") # query = "Who is Bhagat singh?" docs = '' if query: # Get the query embedding question_embedding = get_query_embdedding(query) # Query the Pinecone index query_result = index.query(namespace = st.session_state.namespace, vector=question_embedding, top_k=5, include_metadata=True) # Extract metadata from query result docs = {x["metadata"]['text']: i for i, x in enumerate(query_result["matches"])} # print (docs) # Create a template for the summary Template = f"Based on the following context: {docs} generate a precise summary related to the question: {query}" # print(Template) # Generate the summary chat_completion = client.chat.completions.create( messages=[ { "role": "user", "content": Template, } ], model="llama3-70b-8192", ) # Print the summary response = chat_completion.choices[0].message.content # print(response) result = {"ques":query, "ans":response} st.session_state.chat_list.append(result) for c_list in st.session_state.chat_list: with st.chat_message("user"): st.write(c_list["ques"]) with st.chat_message("AI"): st.write(c_list["ans"]) if c_list["ans"]: resl = sentiment_analysis(c_list["ans"]) if (resl[0]['label'] == "POSITIVE"): st.info('POSITIVE', icon="😄") elif (resl[0]['label'] == "NEGATIVE"): st.info('NEGATIVE', icon="😞") else: st.info('NEUTRAL', icon="😆")