import time import uuid import streamlit as st from PIL import Image from fpdf import FPDF from streamlit_float import * from langchain_core.messages import HumanMessage from langgraph.graph import START, StateGraph, MessagesState from langgraph.checkpoint.memory import MemorySaver from langchain_google_genai import ChatGoogleGenerativeAI # [STREAMLIT] PAGE CONFIGURATION icon = Image.open("icon.png") st.set_page_config(page_title="ChatGPT Clone", page_icon=icon) st.logo("logo.svg") # [STREAMLIT] HIDE MENU hide_menu = """ """ st.markdown(hide_menu, unsafe_allow_html=True) # [STREAMLIT] CENTER TOAST cntr_toast = """ """ st.markdown(cntr_toast, unsafe_allow_html=True) # [STREAMLIT] ADJUST ICON SIZE icon = """ """ st.markdown(icon, unsafe_allow_html=True) # [STREAMLIT] ADJUST LOGO SIZE logo = """ """ st.markdown(logo, unsafe_allow_html=True) # [STREAMLIT] ADJUST BUTTON BORDER btn_border = """ """ st.markdown(btn_border, unsafe_allow_html=True) # [STREAMLIT] ADJUST SETTINGS BUTTON set_btn = """ """ st.markdown(set_btn, unsafe_allow_html=True) # [STREAMLIT] ADJUST TOP PADDING top = """ """ st.markdown(top, unsafe_allow_html=True) # [STREAMLIT] ADJUST HEADER header = """ """ st.markdown(header, unsafe_allow_html=True) # [STREAMLIT] ADJUST USER CHAT ALIGNMENT reverse = """ """ st.markdown(reverse, unsafe_allow_html=True) # [STREAMLIT] HIDE USER ICON hide_icon = """ """ st.markdown(hide_icon, unsafe_allow_html=True) # [STREAMLIT] ADJUST CHAT INPUT PADDING bottom = """ """ st.markdown(bottom, unsafe_allow_html=True) # [STREAMLIT] CHAT INPUT BORDER chat_border = """ """ st.markdown(chat_border, unsafe_allow_html=True) # [STREAMLIT] TEXT AREA BORDER text_border = """ """ st.markdown(text_border, unsafe_allow_html=True) # [STREAMLIT] INITIALIZING SESSION STATES if 'conversation' not in st.session_state: st.session_state['conversation'] = None if 'thread_id' not in st.session_state: st.session_state['thread_id'] = str(uuid.uuid4()) if 'messages' not in st.session_state: st.session_state['messages'] = [] if 'API_Key' not in st.session_state: st.session_state['API_Key'] = "" # [STREAMLIT] STREAM CHAT RESPONSE def stream(content): for word in content.split(" "): yield word + " " time.sleep(0.03) # [FPDF] GENERATE A PDF FILE FROM CONVERSATION def generate_pdf(): pdf = FPDF() pdf.set_auto_page_break(auto=True, margin=15) pdf.add_page() pdf.set_font("Arial", size=12) pdf.set_left_margin(10) pdf.set_right_margin(10) pdf.set_font("Arial", size=16, style="B") pdf.cell(200, 10, txt="Chat Conversation", ln=True, align="C") pdf.ln(10) pdf.set_font("Arial", size=12) for message in st.session_state['messages']: if message['role'] == "assistant": pdf.set_text_color(0, 100, 0) pdf.multi_cell(0, 10, txt=f"Assistant: {message['content']}") else: pdf.set_text_color(0, 0, 255) pdf.multi_cell(0, 10, txt=f"User: {message['content']}") pdf.ln(1) pdf_output = pdf.output(dest="S") return bytes(pdf_output) # [STREAMLIT] CHAT BOT GREETINGS with st.chat_message("assistant", avatar=":material/token:"): st.markdown("How can I assist you? 👋") # [STREAMLIT] DISPLAY THE EXISTING CHAT HISTORY for message in st.session_state['messages']: if message["role"] == "assistant": with st.chat_message(message["role"], avatar=":material/token:"): st.markdown(message["content"]) else: with st.chat_message(message["role"]): st.markdown(message["content"]) # [STREAMLIT] USER CHAT INPUT user_input = st.chat_input("Say something.") # [STREAMLIT] IF SEND BUTTON IS CLICKED if user_input: # [STREAMLIT] CHECK IF API KEY IS INPUTTED if st.session_state['API_Key'] != "": # [STREAMLIT] SHOW USER MESSAGE with st.chat_message("user"): st.markdown(user_input) # [STREAMLIT] STORE USER MESSAGE IN SESSION STATE st.session_state['messages'].append({"role": "user", "content": user_input}) # [LANGGRAPH] INITIALIZE LANGGRAPH AND MEMORY if st.session_state['conversation'] is None: # [LANGGRAPH] DEFINE THE STATE GRAPH workflow = StateGraph(state_schema=MessagesState) model = ChatGoogleGenerativeAI(model="gemini-1.5-flash", google_api_key=st.session_state['API_Key']) # [LANGGRAPH] DEFINE THE LLM NODE def call_model(state: MessagesState): response = model.invoke(state["messages"]) return {"messages": response} workflow.add_edge(START, "model") workflow.add_node("model", call_model) # [LANGGRAPH] ADD MEMORY SAVER memory = MemorySaver() app = workflow.compile(checkpointer=memory) st.session_state['conversation'] = app # [LANGGRAPH] SEND THE INPUT TO LANGGRAPH app = st.session_state['conversation'] thread_id = st.session_state['thread_id'] config = {"configurable": {"thread_id": thread_id}} # [STREAMLIT] SHOW AI RESPONSE input_message = HumanMessage(content=user_input) response_text = "" for event in app.stream({"messages": [input_message]}, config, stream_mode="values"): response_text = event["messages"][-1].content with st.chat_message("assistant", avatar=":material/token:"): st.write(stream(response_text)) # [STREAMLIT] STORE AI RESPONSE IN SESSION STATE st.session_state['messages'].append({"role": "assistant", "content": response_text}) else: st.toast("**API key not found**. Please set your API key in the chat options.", icon="🗝️") # [STREAMLIT] CHAT HISTORY OPTIONS float_init() @st.dialog("Chat Options") def open_options(): col1, col2 = st.columns(2) # [STREAMLIT] CLEAR SESSION STATES with col1: st.write("Clear Conversation") if len(st.session_state['messages']) != 0: clear = st.button("**CLEAR**", type="primary", key="clear", use_container_width=True) if clear: st.session_state['messages'] = [] st.session_state['conversation'] = None st.rerun() else: st.button("**CLEAR**", type="primary", key="clear", disabled=True, use_container_width=True) # [STREAMLIT] DOWNLOAD CONVERSATION SUMMARY with col2: st.write("Summarize Conversation") if len(st.session_state['messages']) != 0: download = st.download_button(label="**SUMMARIZE**", type="primary", key="summarize", data=generate_pdf(), file_name="conversation.pdf", mime="application/pdf", use_container_width=True) if download: st.rerun() else: st.button("**SUMMARIZE**", type="primary", key="summarize", disabled=True, use_container_width=True) # [STREAMLIT] INPUT API KEY if st.session_state['API_Key'] == '': user_API = st.text_input("Enter Your Gemini API Key", type="password") if user_API: if st.button("**SAVE KEY**", type="secondary", key="save", use_container_width=True): st.session_state['API_Key'] = user_API st.rerun() else: st.button("**SAVE KEY**", type="secondary", key="save", disabled=True, use_container_width=True) else: user_API = st.text_input("Enter Your Gemini API Key", value=st.session_state['API_Key'], type="password") if user_API: if st.button("**SAVE KEY**", type="secondary", key="save", use_container_width=True): st.session_state['API_Key'] = user_API st.rerun() else: st.button("**SAVE KEY**", type="secondary", key="save", disabled=True, use_container_width=True) button_container = st.container() with button_container: if st.button("⚙️", type="secondary"): open_options() button_css = float_css_helper(width="1.8rem", height="2rem", right="3rem", top="2rem", transition=0) button_container.float(button_css)