app.py

import streamlit as st
from audio_recorder_streamlit import audio_recorder
from groq_models import create_groq_agent, groq_chatbot, get_tools, summarizer_model
from PIL import Image
from io import BytesIO
import base64
import docx
from streamlit_lottie import st_lottie
import json
from utils import set_safety_settings, about
from streamlit_mic_recorder import speech_to_text
import google.generativeai as genai
import os, random, validators
import time
import tempfile
import asyncio
import edge_tts

st.set_page_config(
    page_title="Super AI Assistant",
    page_icon="⚡",
    layout="wide",
    initial_sidebar_state="auto",
    menu_items={"About": about(), "Get Help":"https://www.linkedin.com/in/aasher-kamal-a227a124b/"},
)

###--- Title ---###
st.markdown("""
    <h1 style='text-align: center;'>
        <span style='color: #F81F6F;'>Super</span> 
        <span style='color: #f5f8fc;'>AI Assistant</span>
    </h1>
""", unsafe_allow_html=True)


google_models = [
    "gemini-1.5-flash",
    "gemini-1.5-pro",
]

groq_models = [
    "llama-3.1-8b-instant",
    "llama-3.1-70b-versatile",
    "llama3-70b-8192",
    "llama3-8b-8192",
    "gemma2-9b-it",
    "mixtral-8x7b-32768"
]

voices = {
    "William":"en-AU-WilliamNeural",
    "James":"en-PH-JamesNeural",
    "Jenny":"en-US-JennyNeural",
    "US Guy":"en-US-GuyNeural",
    "Sawara":"hi-IN-SwaraNeural",
}

@st.cache_data
def load_lottie_file(filepath: str):
    with open(filepath, "r") as f:
        return json.load(f)

###--- FUNCTION FOR SPEECH GENERATION---###
async def generate_speech(text, voice):
    communicate = edge_tts.Communicate(text, voice)
    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as temp_file:
        await communicate.save(temp_file.name)
        temp_file_path = temp_file.name
    return temp_file_path

###---AUTOPLAY SPEECH---###
def get_audio_player(file_path):
    with open(file_path, "rb") as f:
        data = f.read()
        b64 = base64.b64encode(data).decode()
        return f'<audio autoplay="true" src="data:audio/mp3;base64,{b64}">'
        
###---SPEECH FILE TO AUTOPLAY AUDIO---###
def generate_voice(text, voice):
    text_to_speak = (text).translate(str.maketrans('', '', '#-*_😊👋😄😁🥳👍🤩😂😎')) # Removing special chars and emojis
    with st.spinner("Generating voice response..."):
        temp_file_path = asyncio.run(generate_speech(text_to_speak, voice)) 
        audio_player_html = get_audio_player(temp_file_path)  # Create an audio player
        st.markdown(audio_player_html, unsafe_allow_html=True)
        os.unlink(temp_file_path)  # Clean up the temporary audio file

###---LLM & PARAMETERS---###
def get_llm_info(available_models):
    with st.sidebar:
        tip =tip = "Select Gemini models if you require multi-modal capabilities (text, image, audio and video inputs)"
        model = st.selectbox("Choose LLM:", available_models, help=tip)

        model_type = None
        if model.startswith(("llama", "gemma", "mixtral")): model_type = "groq"
        elif model.startswith("gemini"): model_type = "google"

        with st.popover("⚙️Model Parameters", use_container_width=True):
            temp = st.slider("Temperature:", min_value=0.0,
                                            max_value=2.0, value=0.5, step=0.5)
            
            max_tokens = st.slider("Maximum Tokens:", min_value=100,
                                        max_value=2000, value=400, step=200)
    return model, model_type, temp, max_tokens


###--- Function to convert base64 to temp file ---###
def base64_to_temp_file(base64_string, unique_name, file_extension):
    file_bytes = BytesIO(base64.b64decode(base64_string.split(",")[1]))
    temp_file_path = f"{unique_name}.{file_extension}"
    with open(temp_file_path, "wb") as temp_file:
        temp_file.write(file_bytes.read())
    return temp_file_path

##----Preparing messages for Gemini----##
def messages_to_gemini(messages):
    gemini_messages = []
    prev_role = None
    uploaded_files = set([file.display_name.split(".")[0] for file in genai.list_files()])

    for message in messages:
        if prev_role and (prev_role == message["role"]):
            gemini_message = gemini_messages[-1]
        else:
            gemini_message = {
                "role": "model" if message["role"] == "assistant" else "user",
                "parts": [],
            }

        for content in message["content"]:
            content_type = content["type"]

            if content_type in ["text","docx_file"]:
                gemini_message["parts"].append(content[content_type])

            elif content_type == "image_url":
                gemini_message["parts"].append(base64_to_image(content["image_url"]["url"]))

            elif content_type in ["video_file", "audio_file", "speech_input"]:
                file_name = content['unique_name']

                if file_name not in uploaded_files:
                    temp_file_path = base64_to_temp_file(content[content_type], file_name, "mp4" if content_type == "video_file" else "wav")

                    with st.spinner(f"Sending {content_type.replace('_', ' ')} to Gemini..."):
                        file = genai.upload_file(path=temp_file_path)

                        while file.state.name == "PROCESSING":
                            st.write(':green[One moment, please.]')
                            time.sleep(10)
                            file = genai.get_file(file.name)

                        if file.state.name == "FAILED":
                            raise ValueError(file.state.name)

                        file = genai.get_file(name=file.name)
                        gemini_message["parts"].append(file)
                    os.remove(temp_file_path)

            elif content_type == "pdf_file":
                if content['pdf_file'].split(".")[0] not in uploaded_files:
                    with st.spinner("Sending your PDF to Gemini..."):
                        gemini_message["parts"].append(genai.upload_file(path=content['pdf_file']))
                    os.remove(content['pdf_file'])

        if prev_role != message["role"]:
            gemini_messages.append(gemini_message)

        prev_role = message["role"]

    return gemini_messages


##-- Converting base64 to image ---##
def base64_to_image(base64_string):
    
    return Image.open(BytesIO(base64.b64decode(base64_string.split(",")[1])))

##-- Handle PDF and Docx files ---##
def add_pdf_docx_file_to_messages():
    if st.session_state.pdf_docx_uploaded:
        file_type = st.session_state.pdf_docx_uploaded.type
        if file_type == "application/pdf":
        # Save the PDF file
            pdf_id = random.randint(1000, 9999)
            pdf_filename = f"pdf_{pdf_id}.pdf"
            with open(pdf_filename, "wb") as f:
                f.write(st.session_state.pdf_docx_uploaded.read())
            
            # Add the PDF file to session_state messages
            st.session_state.messages.append(
                {
                    "role": "user", 
                    "content": [{
                        "type": "pdf_file",
                        "pdf_file": pdf_filename,
                    }]
                }
            )
        else:
            file_content = st.session_state.pdf_docx_uploaded
            doc = docx.Document(file_content)
            text = " ".join(para.text for para in doc.paragraphs)

            # Add the DOCX file to session_state messages
            st.session_state.messages.append(
                {
                    "role": "user", 
                    "content": [{
                        "type": "docx_file",
                        "docx_file": text,
                    }]
                }
            )

##--- Function for adding media files to session_state messages ---###
def add_media_files_to_messages():
    if st.session_state.uploaded_file:
        file_type = st.session_state.uploaded_file.type
        file_content = st.session_state.uploaded_file.getvalue()
        
        encoded_file = base64.b64encode(file_content).decode()
        content_type = None
        
        if file_type.startswith("image"):
            content_type = "image_url"
            data_url = f"data:{file_type};base64,{encoded_file}"

        elif file_type == "video/mp4":
            content_type = "video_file"
            unique_id = f"temp_{random.randint(1000, 9999)}"
            data_url = f"data:{file_type};base64,{encoded_file}"

        elif file_type.startswith("audio"):
            content_type = "audio_file"
            unique_id = f"temp_{random.randint(1000, 9999)}"
            data_url = f"data:{file_type};base64,{encoded_file}"

        # Only append if the content type is recognized
        if content_type:
            message_content = {"type": content_type}
            if content_type == "image_url":
                message_content["image_url"] = {"url": data_url}
            else:
                message_content[content_type] = data_url
                message_content["unique_name"] = unique_id

            st.session_state.messages.append(
                {
                    "role": "user", 
                    "content": [message_content]
                }
            )

###--- FUNCTION TO ADD CAMERA IMAGE TO MESSAGES ---##
def add_camera_img_to_messages():
    if "camera_img" in st.session_state and st.session_state.camera_img:
        img = base64.b64encode(st.session_state.camera_img.getvalue()).decode()
        st.session_state.messages.append(
            {
                "role": "user", 
                "content": [{
                    "type": "image_url",
                    "image_url": {"url": f"data:image/jpeg;base64,{img}"}
                }]
            }
        )

##--- FUNCTION TO RESET CONVERSATION ---##
def reset_conversation():
    keys_to_reset = ["messages", "groq_chat_history", "uploaded_files", "pdf_docx_uploaded"]

    for key in keys_to_reset:
        if key in st.session_state:
            st.session_state.pop(key, None)

    # for file in genai.list_files():
    #     genai.delete_file(file.name)


##--- FUNCTION TO STREAM GEMINI RESPONSE ---##
def stream_gemini_response(model_params, api_key):
    response_message = ""

    genai.configure(api_key=api_key)
    model = genai.GenerativeModel(
            model_name = model_params["model"],
            generation_config={"temperature": model_params["temperature"], "max_output_tokens": model_params["max_tokens"]},
            safety_settings=set_safety_settings(),
            system_instruction="""You are a helpful assistant who asnwers user's questions professionally and politely."""
        )
    gemini_messages = messages_to_gemini(st.session_state.messages)

    for chunk in model.generate_content(contents=gemini_messages, stream=True):
        chunk_text = chunk.text or ""
        response_message += chunk_text
        yield chunk_text

    st.session_state.messages.append({
    "role": "assistant", 
    "content": [
        {
            "type": "text",
            "text": response_message,
        }
    ]})


if "summarize" not in st.session_state:
    st.session_state.summarize = False

##--API KEYS VALIDATION--##
def validate_api_keys(groq_api_key=None, google_api_key=None):

    return ("gsk" in groq_api_key if groq_api_key else False) or ("AIza" in google_api_key if google_api_key else False)

###--- DISPLAYING CHAT HISTORY---###
def render_content(content):
    content_type = content["type"]
    if content_type == "text":
        st.markdown(content["text"])
    elif content_type == "image_url":
        st.image(content["image_url"]["url"])
    elif content_type == "video_file":
        st.video(content["video_file"])
    elif content_type in ["audio_file", "speech_input"]:
        st.audio(content[content_type], autoplay=content_type == "audio_file")

###--VALIDATING CONTENT TO BE DISPLAYED--###
def is_valid_content(content):
    return not (
        (content["type"] == "text" and content["text"] == "Listen attentively to the audio. If there is a question in the audio, answer it professionally.") or
        content["type"] in ["pdf_file", "docx_file"]
    )


###---CHAT HISTORY UPDATE---###
def update_chat_history(role, content, history):
    history.append({"role": role, "content": content})

###---GROQ MODELS RESPONSE---###
def handle_groq_response(model_params, api_key, question, chat_history, llm_type, tools):
    if llm_type == "Chatbot":
        return st.write_stream(groq_chatbot(model_params=model_params, api_key=api_key,
                                            question=question, chat_history=chat_history))
    elif llm_type == "Agent":
        response = create_groq_agent(model_params=model_params, api_key=api_key,
                                     question=question, tools=get_tools(tools),
                                     chat_history=chat_history)
        st.markdown(response)
        return response

###---- MAIN FUNCTION FOR ALL MODELS CONVERSATION HANDLING---###
def process_user_input(message_container, trasncribed_text):
    prompt = st.chat_input("Type your question", key="question") or st.session_state.speech_file_added

    if not prompt:
        return

    if model_type == "groq":
        question = trasncribed_text if st.session_state.speech_file_added else prompt

        if question is None:
            message_container.error("Couldn't recognize your speech.", icon="❌")
            return

        message_container.chat_message("user", avatar="assets/user.png").markdown(question)
        update_chat_history("user", question, st.session_state.groq_chat_history)

        with message_container.chat_message("assistant", avatar="assets/assistant.png"):
            try:
                final_response = handle_groq_response(model_params, groq_api_key, question, 
                                                      st.session_state.groq_chat_history, 
                                                      groq_llm_type, st.session_state.selected_tools)
                update_chat_history("assistant", final_response, st.session_state.groq_chat_history)
                if "voice_response" in st.session_state and st.session_state.voice_response:
                    response_voice = st.session_state.voice_response
                    generate_voice(final_response, voices[response_voice])

            except Exception as e:
                st.error(f"An error occurred: {e}", icon="❌")

    else:  # Gemini models
        if not st.session_state.speech_file_added:
            message_container.chat_message("user", avatar="assets/user.png").markdown(prompt)
            content = [{"type": "text", "text": prompt}]
        else:
            content = [{"type": "text", "text": "Listen attentively to the audio. If there is a question in the audio, answer it professionally."}]

        update_chat_history("user", content, st.session_state.messages)

        with message_container.chat_message("assistant", avatar="assets/assistant.png"):
            try:
                final_response = st.write_stream(stream_gemini_response(model_params=model_params, api_key=google_api_key))

                if "voice_response" in st.session_state and st.session_state.voice_response:
                    response_voice = st.session_state.voice_response
                    generate_voice(final_response, voices[response_voice])

            except Exception as e:
                st.error(f"An error occurred: {e}", icon="❌")



##--- API KEYS ---##
with st.sidebar:
    st.logo(image="assets/logo.png")
    api_cols = st.columns(2)
    with api_cols[0]:
        with st.popover("🔐 Groq", use_container_width=True):
            groq_api_key = st.text_input("Click [here](https://console.groq.com/keys) to get your Groq API key", type="password")
    
    with api_cols[1]:
        with st.popover("🔐 Google", use_container_width=True):
            google_api_key = st.text_input("Click [here](https://aistudio.google.com/app/apikey) to get your Google API key", type="password")
 
##--- API KEY CHECK ---##
if not validate_api_keys(groq_api_key, google_api_key):
    st.info("Please Add a valid API Key in the sidebar to proceed.")

####--- SIDEBAR AFTER API KEYS VALIDATION ---###
else:
    with st.sidebar:
        st.divider()
        columns = st.columns(2)
        # animation
        with columns[0]:
            lottie_animation = load_lottie_file("assets/animation.json")
            if lottie_animation:
                st_lottie(lottie_animation, height=100, width=100, quality="high", key="lottie_anim")

        with columns[1]:
            if st.toggle("Voice Response"):
                response_voice = st.selectbox("Available Voices:", options=voices.keys(), key="voice_response")
    
        available_models = (google_models if google_api_key else []) + (groq_models if groq_api_key else [])
        model, model_type, temperature, max_tokens = get_llm_info(available_models)

        model_params = {
                "model": model,
                "temperature": temperature,
                "max_tokens": max_tokens
            }
        st.divider()

        ###---- Google Gemini Sidebar Customization----###
        if model_type == "google":
            st.write("Upload a file or take a picture")

            media_cols = st.columns(2)

            with media_cols[0]:
                with st.popover("📁 Upload", use_container_width=True):
                    st.file_uploader(
                        "Upload an image, audio or a video", 
                        type=["png", "jpg", "jpeg", "wav", "mp3", "mp4"], 
                        accept_multiple_files=False,
                        key="uploaded_file",
                        on_change=add_media_files_to_messages,
                    )
            with media_cols[1]:                    
                with st.popover("📷 Camera", use_container_width=True):
                    activate_camera = st.checkbox("Activate camera")
                    if activate_camera:
                        st.camera_input(
                            "Take a picture", 
                            key="camera_img",
                            on_change=add_camera_img_to_messages,
                        )
            st.divider()
            tip = "If you upload a PDF or DOCX file, it will be sent to LLM."
            pdf_upload = st.file_uploader("Upload a PDF or Docx file", type=["pdf", "docx"], key="pdf_docx_uploaded", on_change=add_pdf_docx_file_to_messages, help=tip)
        
        ###---- Groq Models Sidebar Customization----###
        else:
            if "selected_tools" not in st.session_state:
                st.session_state.selected_tools = None

            groq_llm_type = st.radio(label="Select the LLM type:", key="groq_llm_type",options=["Agent", "Chatbot", "Summarizer"], horizontal=True)
            if groq_llm_type == "Summarizer":
                url = st.text_input("Enter YT video or Webpage URL:", key="url_to_summarize",
                                    help="Only Youtube videos having captions can be summarized.")
                
                summarize_button = st.button("Summarize", type="primary", use_container_width=True, key="summarize")

            elif groq_llm_type == "Agent":
                st.session_state.selected_tools = st.multiselect("Select Tools for Agent", default=["Wikipedia", "ArXiv", "DuckDuckGo Search"],
                                       options=["Wikipedia", "ArXiv", "DuckDuckGo Search"])
                

###--- Session state variables ---###
    session_keys = ["messages", "uploaded_files", "groq_chat_history"]
    for key in session_keys:
        if key not in st.session_state:
            st.session_state[key] = []

    if "transcribed_text" not in st.session_state:
        st.session_state.transcribed_text = None

    if "speech_file_added" not in st.session_state:
        st.session_state.speech_file_added = False
        
######-----  Main Interface -----#######
    chat_col1, chat_col2 = st.columns([1,4])

    with chat_col1:
        ###--- Audio Recording ---###
        if model_type == "google":
            audio_bytes = audio_recorder("Speak",
                                        pause_threshold=3,
                                        neutral_color="#f5f8fc",
                                        recording_color="#f81f6f",
                                        icon_name="microphone-lines",
                                        icon_size="3x")
                
            if "prev_speech_hash" not in st.session_state:
                st.session_state.prev_speech_hash = None

            if audio_bytes and st.session_state.prev_speech_hash != hash(audio_bytes):
                st.session_state.prev_speech_hash = hash(audio_bytes)

                speech_base64 = base64.b64encode(audio_bytes).decode()
                unique_id = random.randint(1000, 9999)

                st.session_state.messages.append(
                    {
                        "role": "user",
                        "content": [{
                            "type": "speech_input",
                            "speech_input": f"data:audio/wav;base64,{speech_base64}",
                            "unique_name": f"temp_{unique_id}"
                        }]
                    }
                )
                st.session_state.speech_file_added = True

        else:
            with st.spinner("Transcribing..."):
                st.session_state.transcribed_text = speech_to_text(language="en", just_once=True, use_container_width=True)
        
            if st.session_state.transcribed_text:
                st.session_state.speech_file_added = True

        ###--- Reset Conversation ---###
        st.button(
                "🗑 Reset",
                use_container_width=True,
                on_click=reset_conversation,
                help="If clicked, conversation will be reset.",
            )
    ###--- Session state variables ---###
        if "pdf_docx_uploaded" not in st.session_state:
            st.session_state.pdf_docx_uploaded = None

        if st.session_state.pdf_docx_uploaded:
            if model_type == "google":
                file_name = st.session_state.pdf_docx_uploaded.name
                st.info(f"Your file :green['{file_name}'] has been uploaded!")


    ####---DISPLAY CONVERSATION---###
    with chat_col2:
        message_container = st.container(height=400, border=False)
        if model_type == "google":
            for message in st.session_state.messages:
                valid_contents = [content for content in message["content"] if is_valid_content(content)]

                if valid_contents:
                    avatar = "assets/assistant.png" if message["role"] == "assistant" else "assets/user.png"

                    with message_container.chat_message(message["role"], avatar=avatar):
                        for content in valid_contents:
                            render_content(content)
                            
        if model_type == "groq":
            for msg in st.session_state.groq_chat_history:
                avatar = "assets/assistant.png" if msg["role"] == "assistant" else "assets/user.png"
                with message_container.chat_message(msg["role"], avatar=avatar):
                    st.markdown(msg['content'])

 ###---- Summarizer model------###
    if model_type == "groq" and groq_llm_type == "Summarizer":
        if st.session_state.summarize:
            with message_container.chat_message("assistant", avatar="assets/assistant.png"):
                if not url.strip():
                    st.error("Please enter a URL")
                elif not validators.url(url):
                    st.error("Please enter a valid URL")
                else:
                    try:
                        with st.spinner("Summarizing..."):
                            final_response = summarizer_model(model_params=model_params, api_key=groq_api_key, url=url)
                        st.markdown(final_response)
                        st.session_state.groq_chat_history.append({"role": "assistant", "content": final_response})
                    except Exception as e:
                        st.error(f"An error occurred: {e}", icon="❌")

###----- User Question -----###
    else:
        process_user_input(message_container, st.session_state.transcribed_text)
        st.session_state.speech_file_added = False