Spaces:
Running
Running
| import streamlit as st | |
| import streamlit_authenticator as stauth | |
| from deta import Deta | |
| import yaml | |
| from yaml.loader import SafeLoader | |
| import os | |
| from cryptography.fernet import Fernet | |
| import streamlit_survey as ss | |
| from free_speech_app.DataLoadDb import * | |
| from free_speech_app.FreeSpeechPromptsResponses import * | |
| from langchain.chat_models import ChatOpenAI | |
| import json as json_module | |
| #connect to/create Deta user database | |
| deta = Deta(st.secrets["deta_key"]) | |
| db = deta.Base("user_data") | |
| # fernet key (generated locally, stored in streamlit secrets) | |
| fernet = Fernet(bytes(st.secrets["fernet_key"], 'utf-8')) | |
| # activeloop_token | |
| os.environ["ACTIVELOOP_TOKEN"] = st.secrets["deeplake_key"] | |
| config_drive = deta.Drive("passwords") | |
| config = config_drive.get("config.yaml").read() | |
| config = yaml.load(config, Loader=SafeLoader) | |
| # Create an authenticator | |
| authenticator = stauth.Authenticate( | |
| config['credentials'], | |
| config['cookie']['name'], | |
| config['cookie']['key'], | |
| config['cookie']['expiry_days'], | |
| config['preauthorized'] | |
| ) | |
| def get_user_data(user): | |
| data = db.fetch().items | |
| for person in data: | |
| if person['key'] == user: | |
| return person | |
| return None | |
| def encrypt(api_key: str, fernet) -> bytes: | |
| """Encrypt the API key.""" | |
| return fernet.encrypt(api_key.encode()) | |
| def decrypt(encrypted_api_key: bytes, fernet) -> str: | |
| """Decrypt the encrypted API key.""" | |
| return fernet.decrypt(encrypted_api_key).decode() | |
| def add_logos(): | |
| st.title("Freequalizer") | |
| left_col_log, right_col_log = st.columns(2) | |
| free_speech_logo = './free_speech_app/logos/Future-of-Free-Speech-logo.png' | |
| vu_logo = './free_speech_app/logos/Vanderbilt-University-Logo.png' | |
| with left_col_log: | |
| st.image(free_speech_logo, caption=None, width=200, use_column_width=None, | |
| clamp=False, channels="RGB", output_format="auto") | |
| with right_col_log: | |
| st.image(vu_logo, caption=None, width=200, use_column_width=None, | |
| clamp=False, channels="RGB", output_format="auto") | |
| authentication_status = None | |
| if not authentication_status: | |
| add_logos() | |
| # Render the login module | |
| name, authentication_status, username = authenticator.login() | |
| # If the user is authenticated | |
| if authentication_status: | |
| authenticator.logout() | |
| st.write(f'Welcome *{name}*') | |
| # Sidebar for navigation | |
| page = st.sidebar.radio("Choose a page", ["Account Setup", "API Key Help", "Respond to Post"]) | |
| # Fetch user data from the database | |
| user_data = get_user_data(username) | |
| # get principles from organization-wide file | |
| with open('org-principles.txt', 'r') as file: | |
| org_principles = str(file.read()) | |
| if page == "Account Setup": | |
| if not user_data or user_data.get('login_count') in [None, 0]: | |
| survey = ss.StreamlitSurvey("Please answer the following questions to help define your principles") | |
| survey.text_input("Who would you identify as a hero or exemplar, fictional or real, whose moral courage reflects what you also believe in? Why?", id="Q1") | |
| survey.text_input("What is an instance of hate that would cross a line for you, making you feel a need to respond?", id="Q2") | |
| survey.text_input("Given your character, how would you seek to address the harms of hate (i.e., to educate, persuade, deescalate, forgive, repair, etc.)?", id="Q3") | |
| json_string = survey.to_json() | |
| json_dict = json_module.loads(json_string) | |
| survey_responses = json_dict["Q1"]["value"] + " " + json_dict["Q2"]["value"] + " " + json_dict["Q3"]["value"] | |
| st.title("Account Setup") | |
| st.markdown("Please use this page to provide your OpenAI API Key, Principles and Writing Style. **Please make sure to press the Save Changes button after providing the information.**") | |
| # Input boxes with existing data | |
| if 'api_key' not in st.session_state: | |
| st.session_state.api_key = "" | |
| api_input = st.text_input("Paste OpenAI API Key", value=decrypt(user_data["api_key"].encode()[ | |
| 2:-1], fernet) if user_data and "api_key" in user_data else "", type="password") | |
| encrypted_api_key = str(encrypt(api_input, fernet)) | |
| st.session_state.api_key = api_input | |
| principles = st.text_area("My Principles (Enter or autopopulates from questions above)", height = 300, value= user_data["principles"] +"\n" + org_principles if user_data and "principles" in user_data and user_data["principles"].strip() != "" else survey_responses + "\n" + org_principles) | |
| writing_style = st.text_area("My Writing Style (Paste Examples)", height = 300, placeholder = "Provide examples of your writing style here", value=user_data["writing_style"] if user_data and "writing_style" in user_data else "") | |
| #sources = st.text_area("Sources (This autopopulates for your reference)", value=st.session_state.sources if 'sources' in st.session_state else '', key = 'sources_key', height = 100) | |
| # Update button | |
| if st.button("Save Changes"): | |
| db.put({"key": username, "principles": principles, "writing_style": writing_style, "api_key": encrypted_api_key, "login_count": 0 if principles.strip() == "" else 1}) | |
| if page == "API Key Help": | |
| st.title("OpenAI API Key Setup") | |
| st.header('What is an API key?') | |
| st.write('An API (Application Programming Interface) key is like a password that allows you to access certain functions or data from a website or service. Many sites use API keys to identify you and control access to their APIs.') | |
| st.header('Why do you need an API key?') | |
| st.write('API keys allow sites to track usage and prevent abuse of their services. They help keep things secure. When you request an API key, the site knows the calls are coming from you.') | |
| image = 'apikeyex.png' | |
| st.header('How to get an OpenAI API key:') | |
| st.write('1. Go to https://platform.openai.com/account/api-keys') | |
| st.write('2. Log in or create an OpenAI account if you do not have one') | |
| st.write('3. Click "Create new secret key" and give your key a name') | |
| st.image(image, caption=None, width=None, use_column_width=None, | |
| clamp=False, channels="RGB", output_format="auto") | |
| st.write('4. Copy the generated API key and keep it private like a password') | |
| st.header('Using your API key') | |
| st.write('When making calls to the OpenAI API, include your API key in the request headers or parameters to authenticate.') | |
| st.code('headers = {"Authorization": f"Bearer {YOUR_API_KEY}"}') | |
| st.warning('Treat your API key like a secret! Do not share it publicly.') | |
| elif page == "Respond to Post": | |
| st.title("Respond to Post") | |
| left_col, right_col = st.columns(2) | |
| # Input boxes | |
| with right_col: | |
| regenerate_background = False | |
| background_info = st.text_area("Background information on original post (This autopopulates for your reference). The Background information is generated by OpenAI's GPT-4, and used by the model to generate a response. Because the Background information is generated by an LLM, it may contain mistakes.", height = 780, value=st.session_state.background_info if 'background_info' in st.session_state else '', key = 'background_info_key') | |
| if st.button("Request New Background Information"): | |
| regenerate_background = True | |
| with left_col: | |
| original_post = st.text_area("Paste Original Post Here \n", height=100) | |
| word_limit = st.text_input("Word Limit for Response", placeholder = "Please provide a word limit for the response") | |
| chat_mdl = None | |
| draft_response = '' | |
| # Check if the "Submit" button is clicked | |
| if st.button("Submit"): | |
| if st.session_state.api_key: | |
| os.environ["OPENAI_API_KEY"] = st.session_state.api_key | |
| # add condition to check for passphrase to allow use of DSI api key stored in secrets | |
| if (os.environ["OPENAI_API_KEY"] == st.secrets["secret_passphrase"]): | |
| #umang key | |
| os.environ["OPENAI_API_KEY"] = st.secrets["dsi_openai_key"] | |
| elif (os.environ["OPENAI_API_KEY"] == st.secrets["secret_passphrase2"]): | |
| #abbie key | |
| os.environ["OPENAI_API_KEY"] = st.secrets["dsi_openai_key2"] | |
| elif (os.environ["OPENAI_API_KEY"] == st.secrets["secret_passphrase3"]): | |
| #myranda key | |
| os.environ["OPENAI_API_KEY"] = st.secrets["dsi_openai_key3"] | |
| elif (os.environ["OPENAI_API_KEY"] == st.secrets["secret_passphrase4"]): | |
| #jasmine key | |
| os.environ["OPENAI_API_KEY"] = st.secrets["dsi_openai_key4"] | |
| chat_mdl = ChatOpenAI(model_name='gpt-4', temperature=0.1) | |
| else: | |
| st.warning('Please enter Open AI API Key') | |
| if chat_mdl is not None: | |
| if user_data is None: | |
| draft_response, background_text = generate_custom_response(original_post, chat_mdl, "", "", word_limit) | |
| response_assessment = assessing_response(chat_mdl, draft_response.content) | |
| st.session_state.draft_response = draft_response.content | |
| st.session_state.background_text = background_text | |
| # st.session_state.sources_text = sources_text | |
| st.session_state.background_info = background_text | |
| # st.session_state.sources = sources_text | |
| st.session_state.response_assessment = response_assessment | |
| st.rerun() | |
| else: | |
| draft_response, background_text = generate_custom_response(original_post, chat_mdl, user_data['principles'], user_data['writing_style'], word_limit) | |
| response_assessment = assessing_response(chat_mdl, draft_response.content) | |
| st.session_state.draft_response = draft_response.content | |
| st.session_state.background_text = background_text | |
| # st.session_state.sources_text = sources_text | |
| st.session_state.background_info = background_text | |
| # st.session_state.sources = sources_text | |
| st.session_state.response_assessment = response_assessment.content | |
| st.rerun() | |
| # Ensure session state variables are initialized | |
| if 'draft_response' not in st.session_state: | |
| st.session_state.draft_response = '' | |
| if 'regenerate_prompt' not in st.session_state: | |
| st.session_state.regenerate_prompt = '' | |
| if 'response_assessment' not in st.session_state: | |
| st.session_state.response_assessment = '' | |
| # Output from function | |
| response_textarea = st.text_area( | |
| label="Draft Response. Please edit here or prompt suggestions in the box below.", | |
| value=st.session_state.draft_response if 'draft_response' in st.session_state else '', | |
| height=350, | |
| key='draft_response_key' | |
| ) | |
| # Add option to save responses | |
| if st.button("Save This Response", help="Save this response to your writing style."): | |
| #user_data["writing_style"] += st.session_state.draft_response | |
| writing_style = user_data["writing_style"] + "\n" + st.session_state.draft_response | |
| principles = user_data["principles"] if user_data and "principles" in user_data else "" | |
| encrypted_api_key = str(encrypt(st.session_state.api_key, fernet)) | |
| db.put({"key": username, "writing_style": writing_style, "principles": principles,"api_key": encrypted_api_key, "login_count": 0 if principles.strip() == "" else 1}) | |
| st.write("Response saved to your writing style.") | |
| # Assessment of the response | |
| response_assessment = st.text_area( | |
| label="Assessment of the Response", | |
| value=st.session_state.response_assessment if 'response_assessment' in st.session_state else '', | |
| height=350, | |
| key='response_assessment_key' | |
| ) | |
| # Initialization of the regeneration flag | |
| if 'is_regenerating' not in st.session_state: | |
| st.session_state.is_regenerating = False | |
| # Check if the app is in the "regeneration" phase | |
| if st.session_state.is_regenerating: | |
| # Display the regenerated response explicitly | |
| regenerate_prompt = st.text_area( | |
| "Request a new draft", | |
| value=st.session_state.regenerate_prompt, | |
| placeholder="You may edit the regenerated draft directly above, or request further changes here.", | |
| height=100, | |
| key='regenerate_prompt_key' | |
| ) | |
| # Reset the regeneration flag | |
| st.session_state.is_regenerating = False | |
| else: | |
| # Normal behavior: display the text area for manual input | |
| regenerate_prompt = st.text_area( | |
| "Request a new draft", | |
| placeholder="You may edit the draft directly above, or request a new draft with additional guidance here.", | |
| height=100, | |
| key='regenerate_prompt_key' | |
| ) | |
| if (draft_response is not None) and (regenerate_prompt is not None): | |
| if st.button("Regenerate"): | |
| if st.session_state.api_key: | |
| os.environ['OPENAI_API_KEY'] = st.session_state.api_key | |
| # add condition to check for passphrase to allow use of DSI api key stored in secrets | |
| if (os.environ["OPENAI_API_KEY"] == st.secrets["secret_passphrase"]): | |
| #umang key | |
| os.environ["OPENAI_API_KEY"] = st.secrets["dsi_openai_key"] | |
| elif (os.environ["OPENAI_API_KEY"] == st.secrets["secret_passphrase2"]): | |
| #abbie key | |
| os.environ["OPENAI_API_KEY"] = st.secrets["dsi_openai_key2"] | |
| elif (os.environ["OPENAI_API_KEY"] == st.secrets["secret_passphrase3"]): | |
| #myranda key | |
| os.environ["OPENAI_API_KEY"] = st.secrets["dsi_openai_key3"] | |
| elif (os.environ["OPENAI_API_KEY"] == st.secrets["secret_passphrase4"]): | |
| #jasmine key | |
| os.environ["OPENAI_API_KEY"] = st.secrets["dsi_openai_key4"] | |
| chat_mdl = ChatOpenAI( | |
| model_name='gpt-4', temperature=0.1) | |
| if chat_mdl is not None: | |
| updated_response, background_info = regenerate_custom_response( | |
| original_post, chat_mdl, regenerate_prompt, st.session_state.draft_response, regenerate_background, st.session_state.background_text) | |
| st.session_state.draft_response = updated_response.content | |
| updated_response_assessment = assessing_response(chat_mdl, updated_response).content | |
| st.session_state.response_assessment = updated_response_assessment | |
| st.session_state.is_regenerating = False | |
| st.rerun() | |
| if regenerate_background: | |
| if st.session_state.api_key: | |
| os.environ['OPENAI_API_KEY'] = st.session_state.api_key | |
| # add condition to check for passphrase to allow use of DSI api key stored in secrets | |
| if (os.environ["OPENAI_API_KEY"] == st.secrets["secret_passphrase"]): | |
| #umang key | |
| os.environ["OPENAI_API_KEY"] = st.secrets["dsi_openai_key"] | |
| elif (os.environ["OPENAI_API_KEY"] == st.secrets["secret_passphrase2"]): | |
| #abbie key | |
| os.environ["OPENAI_API_KEY"] = st.secrets["dsi_openai_key2"] | |
| elif (os.environ["OPENAI_API_KEY"] == st.secrets["secret_passphrase3"]): | |
| #myranda key | |
| os.environ["OPENAI_API_KEY"] = st.secrets["dsi_openai_key3"] | |
| elif (os.environ["OPENAI_API_KEY"] == st.secrets["secret_passphrase4"]): | |
| #jasmine key | |
| os.environ["OPENAI_API_KEY"] = st.secrets["dsi_openai_key4"] | |
| chat_mdl = ChatOpenAI( | |
| model_name='gpt-4', temperature=0.1) | |
| if chat_mdl is not None: | |
| updated_response, background_text = regenerate_custom_response( | |
| original_post, chat_mdl, regenerate_prompt, st.session_state.draft_response, regenerate_background, st.session_state.background_text) | |
| st.session_state.draft_response = updated_response.content | |
| st.session_state.background_text = background_info | |
| st.session_state.background_info = background_info # what is the difference between this and background_text? | |
| regenerate_background = False | |
| st.session_state.is_regenerating = False | |
| st.rerun() | |
| elif authentication_status is False: | |
| st.error('Username/password is incorrect') | |
| # added registration module if username/password is incorrect | |
| try: | |
| if authenticator.register_user('New User Registration', preauthorization=False): | |
| st.success('User Registered Successfully! Please log in above.') | |
| except Exception as e: | |
| st.error(e) | |
| elif authentication_status is None: | |
| st.warning('Please enter your username and password') | |
| try: | |
| if authenticator.register_user(preauthorization=False): | |
| st.success('User Registered Successfully! Please log in above.') | |
| except Exception as e: | |
| st.error(e) | |
| with open('config.yaml', 'w') as file: | |
| yaml.dump(config, file, default_flow_style=False) | |
| config_drive.put("config.yaml", path="config.yaml") | |