import streamlit as st from streamlit_feedback import streamlit_feedback import os import pandas as pd import base64 from io import BytesIO import nest_asyncio from llama_index.llms import OpenAI from llama_index import SimpleDirectoryReader from llama_index import Document from llama_index import VectorStoreIndex from llama_index import ServiceContext from llama_index.embeddings import HuggingFaceEmbedding from llama_index.memory import ChatMemoryBuffer from vision_api import get_transcribed_text nest_asyncio.apply() # App title st.set_page_config(page_title="💬 Open AI Chatbot") openai_api = os.getenv("OPENAI_API_KEY") # "./raw_documents/HI_Knowledge_Base.pdf" input_files = ["./raw_documents/HI Chapter Summary Version 1.3.pdf", "./raw_documents/qna.txt"] embedding_model = "BAAI/bge-small-en-v1.5" system_content = ("You are a helpful study assistant. " "You do not respond as 'User' or pretend to be 'User'. " "You only respond once as 'Assistant'." ) data_df = pd.DataFrame( { "Completion": [30, 40, 100, 10], } ) data_df.index = ["Chapter 1", "Chapter 2", "Chapter 3", "Chapter 4"] # Replicate Credentials with st.sidebar: st.title("💬 Open AI Chatbot") st.write("This chatbot is created using the GPT model from Open AI.") if openai_api: pass elif "OPENAI_API_KEY" in st.secrets: st.success("API key already provided!", icon="✅") openai_api = st.secrets["OPENAI_API_KEY"] else: openai_api = st.text_input("Enter OpenAI API token:", type="password") if not (openai_api.startswith("sk-") and len(openai_api)==51): st.warning("Please enter your credentials!", icon="⚠️") else: st.success("Proceed to entering your prompt message!", icon="👉") ### for streamlit purpose os.environ["OPENAI_API_KEY"] = openai_api st.subheader("Models and parameters") selected_model = st.sidebar.selectbox("Choose an OpenAI model", ["gpt-3.5-turbo-1106", "gpt-4-1106-preview"], key="selected_model") temperature = st.sidebar.slider("temperature", min_value=0.0, max_value=2.0, value=0.0, step=0.01) st.data_editor( data_df, column_config={ "Completion": st.column_config.ProgressColumn( "Completion %", help="Percentage of content covered", format="%.1f%%", min_value=0, max_value=100, ), }, hide_index=False, ) st.markdown("📖 Reach out to SakiMilo to learn how to create this app!") if "init" not in st.session_state.keys(): st.session_state.init = {"warm_started": "No"} st.session_state.feedback = False # Store LLM generated responses if "messages" not in st.session_state.keys(): st.session_state.messages = [{"role": "assistant", "content": "How may I assist you today?", "type": "text"}] if "feedback_key" not in st.session_state: st.session_state.feedback_key = 0 if "release_file" not in st.session_state: st.session_state.release_file = "false" def clear_chat_history(): st.session_state.messages = [{"role": "assistant", "content": "How may I assist you today?", "type": "text"}] chat_engine = get_query_engine(input_files=input_files, llm_model=selected_model, temperature=temperature, embedding_model=embedding_model, system_content=system_content) chat_engine.reset() st.sidebar.button("Clear Chat History", on_click=clear_chat_history) if st.sidebar.button("I want to submit a feedback!"): st.session_state.feedback = True st.session_state.feedback_key += 1 # overwrite feedback component @st.cache_resource def get_document_object(input_files): documents = SimpleDirectoryReader(input_files=input_files).load_data() document = Document(text="\n\n".join([doc.text for doc in documents])) return document @st.cache_resource def get_llm_object(selected_model, temperature): llm = OpenAI(model=selected_model, temperature=temperature) return llm @st.cache_resource def get_embedding_model(model_name): embed_model = HuggingFaceEmbedding(model_name=model_name) return embed_model @st.cache_resource def get_query_engine(input_files, llm_model, temperature, embedding_model, system_content): document = get_document_object(input_files) llm = get_llm_object(llm_model, temperature) embedded_model = get_embedding_model(embedding_model) service_context = ServiceContext.from_defaults(llm=llm, embed_model=embedded_model) index = VectorStoreIndex.from_documents([document], service_context=service_context) memory = ChatMemoryBuffer.from_defaults(token_limit=15000) # chat_engine = index.as_query_engine(streaming=True) chat_engine = index.as_chat_engine( chat_mode="context", memory=memory, system_prompt=system_content ) return chat_engine def generate_llm_response(prompt_input): chat_engine = get_query_engine(input_files=input_files, llm_model=selected_model, temperature=temperature, embedding_model=embedding_model, system_content=system_content) # st.session_state.messages response = chat_engine.stream_chat(prompt_input) return response def handle_feedback(user_response): st.toast("✔️ Feedback received!") st.session_state.feedback = False def handle_image_upload(): st.session_state.release_file = "true" # Warm start if st.session_state.init["warm_started"] == "No": clear_chat_history() st.session_state.init["warm_started"] = "Yes" # Image upload option with st.sidebar: image_file = st.file_uploader("Upload your image here...", type=["png", "jpeg", "jpg"], on_change=handle_image_upload) if st.session_state.release_file == "true" and image_file: with st.spinner("Uploading..."): b64string = base64.b64encode(image_file.read()).decode('utf-8') message = { "role": "user", "content": b64string, "type": "image"} st.session_state.messages.append(message) transcribed_msg = get_transcribed_text(b64string) message = { "role": "admin", "content": transcribed_msg, "type": "text"} st.session_state.messages.append(message) st.session_state.release_file = "false" # Display or clear chat messages for message in st.session_state.messages: if message["role"] == "admin": continue with st.chat_message(message["role"]): if message["type"] == "text": st.write(message["content"]) elif message["type"] == "image": img_io = BytesIO(base64.b64decode(message["content"].encode("utf-8"))) st.image(img_io) # User-provided prompt if prompt := st.chat_input(disabled=not openai_api): client = OpenAI() st.session_state.messages.append({"role": "user", "content": prompt, "type": "text"}) with st.chat_message("user"): st.write(prompt) # Retrieve text prompt from image submission if prompt is None and \ st.session_state.messages[-1]["role"] == "admin": prompt = st.session_state.messages[-1]["content"] # Generate a new response if last message is not from assistant if st.session_state.messages[-1]["role"] != "assistant": with st.chat_message("assistant"): with st.spinner("Thinking..."): response = generate_llm_response(prompt) placeholder = st.empty() full_response = "" for token in response.response_gen: full_response += token placeholder.markdown(full_response) placeholder.markdown(full_response) message = {"role": "assistant", "content": full_response, "type": "text"} st.session_state.messages.append(message) # Trigger feedback if st.session_state.feedback: result = streamlit_feedback( feedback_type="thumbs", optional_text_label="[Optional] Please provide an explanation", on_submit=handle_feedback, key=f"feedback_{st.session_state.feedback_key}" )