nh_chatbot / app.py
nshern's picture
.
07645e7
import logging
import uuid
import streamlit as st
from st_pages import Page, show_pages
from nh_chatbot import create_engine
from nh_chatbot.etl import utils
from nh_chatbot.utils import filter_no_source_response, get_urls_from_sources
logging.basicConfig(
level=logging.INFO,
filename="app.log",
filemode="a",
format="%(asctime)s - %(name)s - %(levelname)s - Session ID: %(session_id)s - %(message)s",
)
def reset_chat():
st.session_state.chat_engine.reset()
init_chat()
def show_info_text():
st.header("NH Chatbot")
st.markdown("Indskriv dit spørgsmål i feltet placeret i bunden af siden 👇")
st.markdown("")
def init_chat():
st.session_state.messages = []
st.session_state.chat_engine.reset()
with open("app.log", "r") as file:
log_content = file.read()
show_info_text()
show_pages(
[
Page("app.py", "Hjem", "🏠"),
Page("other_pages/settings.py", "Indstillinger", "⚙️"),
Page("other_pages/developer.py", "Udvikler", "🔨"),
]
)
css = """
<style>
[data-testid="stSidebar"]{
min-width: 400px;
max-width: 800px;
}
</style>
"""
st.markdown(css, unsafe_allow_html=True)
@st.cache_resource(show_spinner="Starting engine..")
def load_engine_from_cache():
return create_engine.create_engine()
if "chat_engine" not in st.session_state:
engine = load_engine_from_cache()
st.session_state.chat_engine = engine
# Initialize chat history
if "messages" not in st.session_state:
init_chat()
if "session_id" not in st.session_state:
st.session_state.session_id = uuid.uuid4()
# NOTE: Display chat messages from history on app rerun
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
dummy_input = "Kan jeg tilføje oplysninger, dokumenter eller bilag til klagen ved at sende dem via mail efterfølgende?"
dummy_input = "..."
prompt = st.chat_input(dummy_input)
if prompt:
with st.chat_message(name="User"):
st.write(prompt)
with st.spinner("Tænker..."):
response = st.session_state.chat_engine.chat(prompt)
with st.chat_message(name="LLM"):
enhanced_response = f"{response.response}\n"
if len(response.source_nodes) > 0:
# NOTE: Extract references from the source node metadata
urls = get_urls_from_sources(response)
enhanced_response = enhanced_response + "\n*Kildehenvisning:*\n"
for url in set(urls):
enhanced_response = enhanced_response + f"\n- *{url}*"
else:
enhanced_response = filter_no_source_response(response)
st.markdown(enhanced_response)
st.session_state.messages.append(
{"role": "user", "content": prompt},
)
st.session_state.messages.append(
{"role": "llm", "content": enhanced_response},
)
enhanced_response = str(enhanced_response).replace("\n", "\\n")
logging.info(f"user: {prompt}", extra={"session_id": st.session_state.session_id})
logging.info(
f"llm: {enhanced_response}",
extra={"session_id": st.session_state.session_id},
)