File size: 4,141 Bytes
e5b28bf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | import streamlit as st
import time
import json
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, HumanMessagePromptTemplate, MessagesPlaceholder
chat_model = ChatGoogleGenerativeAI(api_key="AIzaSyC1B3zDW4G19olwgTz368YgS-ZARqzsEFE", model="gemini-2.0-flash-exp")
output_parser = StrOutputParser()
chat_template = ChatPromptTemplate(
[
SystemMessage("""you act as an data science instructor. so you should answer only data science related questions.
# if anyone ask you other questions rather then data science then simply tell them to ask data science related question."""),
MessagesPlaceholder(variable_name="chat_history"),
HumanMessagePromptTemplate.from_template("""{Que}""")
]
)
with st.sidebar:
st.title("Data Science Tutor App")
with st.spinner("Loading..."):
time.sleep(1)
st.success("Done!")
st.title(":tophat: Data Science Tutor")
memory_buffer = {"history": []}
def load_history():
try:
with open("history.json", "r") as file:
data = json.load(file)
history = []
for message in data["history"]:
if message["type"] == "HumanMessage":
history.append(HumanMessage(content=message["content"]))
elif message["type"] == "AIMessage":
history.append(AIMessage(content=message["content"]))
return {"history": history}
except (FileNotFoundError, json.JSONDecodeError):
return {"history": []}
def save_history(history):
with open("history.json", "w") as file:
data = {"history": []}
for message in history["history"]:
if isinstance(message, HumanMessage):
data["history"].append({"type": "HumanMessage", "content": message.content})
elif isinstance(message, AIMessage):
data["history"].append({"type": "AIMessage", "content": message.content})
json.dump(data, file, indent=4)
memory_buffer = load_history()
def get_history_from_buffer(human_input):
return memory_buffer["history"]
def my_fragment(source):
qu = {"Que": source}
response = chain.invoke(qu)
memory_buffer["history"].append(HumanMessage(content=qu["Que"]))
memory_buffer["history"].append(AIMessage(content=response))
save_history(memory_buffer)
return memory_buffer["history"]
runnable_get_history_from_buffer = RunnableLambda(get_history_from_buffer)
chain = RunnablePassthrough.assign(chat_history=runnable_get_history_from_buffer) | chat_template | chat_model | output_parser
conversation_container = st.container()
st.markdown(
"""
<style>
.stTextArea textarea {
position: fixed;
bottom: 80px;
width: 50%;
background-color: #f0f0f0;
}
.stButton button {
position: fixed;
bottom: 10px;
}
#history-container {
max-height: 70vh;
overflow-y: auto;
}
</style>
""", unsafe_allow_html=True
)
input_container = st.container()
with input_container:
source = st.text_area(label="Enter your data science question", placeholder="Enter Your Data Science Question...")
if st.button("Generate", type="primary"):
if source:
my_fragment(source)
source = ""
st.subheader("Your Chat")
for message in memory_buffer["history"]:
if isinstance(message, HumanMessage):
st.write(f":speech_balloon:: {message.content}")
elif isinstance(message, AIMessage):
st.write(f":point_right:: {message.content}")
st.markdown(
"""
<script>
const chatHistory = document.querySelector('#history-container');
chatHistory.scrollTop = chatHistory.scrollHeight;
</script>
""", unsafe_allow_html=True
) |