# Importing required packages import streamlit as st from streamlit_extras.add_vertical_space import add_vertical_space import openai import uuid import time import os from openai import OpenAI import base64 from PIL import Image import io import matplotlib as plt from PIL import Image api_key = os.getenv("OPENAI_API_KEY") assistant_id = os.getenv("assistant_id") def handle_and_display_image(file_id): api_response = client.files.with_raw_response.retrieve_content(file_id) if api_response.status_code == 200: content = api_response.content with open('image.png', 'wb') as f: f.write(content) return Image.open('image.png') else: print('Failed to download file.') return None client = openai.Client() MODEL = "gpt-4-1106-preview" def page1(): try: if "session_id" not in st.session_state: st.session_state.session_id = str(uuid.uuid4()) if "run" not in st.session_state: st.session_state.run = {"status": None} if "messages" not in st.session_state: st.session_state.messages = [] if "retry_error" not in st.session_state: st.session_state.retry_error = 0 if "assistant" not in st.session_state: openai.api_key = api_key # Load the previously created assistant st.session_state.assistant = openai.beta.assistants.retrieve(assistant_id) # Create a new thread for this session st.session_state.thread = client.beta.threads.create( metadata={ 'session_id': st.session_state.session_id, } ) # If the run is completed, display the messages elif hasattr(st.session_state.run, 'status') and st.session_state.run.status == "completed": # Retrieve the list of messages st.session_state.messages = client.beta.threads.messages.list( thread_id=st.session_state.thread.id ) end_time = time.time() # Stop the timer duration = end_time - start_time # Calculate duration st.write(f"Response time: {duration:.2f} seconds") # Display messages for message in reversed(st.session_state.messages.data): if message.role in ["user", "assistant"]: with st.chat_message(message.role): for content_part in message.content: if content_part.type == 'text': # For text responses message_text = content_part.text.value st.markdown(message_text) # New code to handle image files elif hasattr(content_part, 'image_file') and content_part.image_file: image = handle_and_display_image(content_part.image_file.file_id) if image: st.image(image, caption="Generated Image") start_time = time.time() # Start the timer if prompt := st.chat_input("Wie kann ich dir helfen?"): with st.chat_message('user'): st.write(prompt) # Add message to the thread st.session_state.messages = client.beta.threads.messages.create( thread_id=st.session_state.thread.id, role="user", content=prompt ) # Do a run to process the messages in the thread st.session_state.run = client.beta.threads.runs.create( thread_id=st.session_state.thread.id, assistant_id=st.session_state.assistant.id, ) if st.session_state.retry_error < 3: time.sleep(1) # Wait 1 second before checking run status st.rerun() # Check if 'run' object has 'status' attribute if hasattr(st.session_state.run, 'status'): # Handle the 'running' status if st.session_state.run.status == "running": with st.chat_message('assistant'): st.write("Thinking ......") if st.session_state.retry_error < 3: time.sleep(1) # Short delay to prevent immediate rerun, adjust as needed st.rerun() # Handle the 'failed' status elif st.session_state.run.status == "failed": st.session_state.retry_error += 1 with st.chat_message('assistant'): if st.session_state.retry_error < 3: st.write("Run failed, retrying ......") time.sleep(3) # Longer delay before retrying st.rerun() else: st.error("FAILED: The OpenAI API is currently processing too many requests. Please try again later ......") # Handle any status that is not 'completed' elif st.session_state.run.status != "completed": # Attempt to retrieve the run again, possibly redundant if there's no other status but 'running' or 'failed' st.session_state.run = client.beta.threads.runs.retrieve( thread_id=st.session_state.thread.id, run_id=st.session_state.run.id, ) if st.session_state.retry_error < 3: time.sleep(3) st.rerun() except Exception as e: st.error(f"Upsi, an unexpected error occurred: {e}") def page2(): try: hide_streamlit_style = """ """ st.markdown(hide_streamlit_style, unsafe_allow_html=True) # Create columns for layout col1, col2 = st.columns([3, 1]) # Adjust the ratio to your liking with col1: st.title("Test Page with Logo!") with col2: # Load and display the image in the right column, which will be the top-right corner of the page image = Image.open('BinDoc Logo (Quadratisch).png') st.image(image, use_column_width='always') except Exception as e: st.error(f"Upsi, an unexpected error occurred: {e}") def main(): # Sidebar content with st.sidebar: st.title('BinDoc GmbH') st.markdown("Experience revolutionary interaction with BinDocs Chat App, leveraging state-of-the-art AI technology.") add_vertical_space(1) page = st.sidebar.selectbox("Choose a page", ["Analysis Bot", "Test Page"]) add_vertical_space(1) st.write('Made with ❤️ by BinDoc GmbH') # Main area content based on page selection if page == "Analysis Bot": page1() elif page == "Test Page": page2() if __name__ == "__main__": main()