Spaces:
Sleeping
Sleeping
| # 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 = """ | |
| <style> | |
| #MainMenu {visibility: hidden;} | |
| footer {visibility: hidden;} | |
| </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() | |