app.py

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="😆")