MatteoScript's picture
Update app.py
0bf87c6 verified
import streamlit as st
from chat_client import chat
import time
import os
from dotenv import load_dotenv
from sentence_transformers import SentenceTransformer
import requests
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
load_dotenv()
URL_APP_SCRIPT = os.getenv('URL_APP_SCRIPT')
CHAT_BOTS = {"Mixtral 8x7B v0.1" :"mistralai/Mixtral-8x7B-Instruct-v0.1",
"Mixtral 7B v0.2" :"mistralai/Mistral-7B-Instruct-v0.2",
"OpenAi GTP4" : "mistralai/Mistral-7B-Instruct-v0.1",
"Antrophic CLAUDE" : "mistralai/Mistral-7B-Instruct-v0.1"}
options_old = {
'Email Genitori': {'systemRole': 'Tu sei un esperto scrittore di email. Attieniti allo stile che ti ho fornito nelle instruction e inserici il contenuto richiesto. Genera il testo di una mail a partire da questo contenuto, con lo stile ricevuto in precedenza: ',
'systemStyle': 'Utilizza lo stile fornito come esempio e parla in ITALIANO e firmati sempre come il Signor Preside',
'instruction': URL_APP_SCRIPT + '1IxE0ic0hsWrxQod2rfh4hnKNqMC-lGT4',
'RAG': False},
'Email Colleghi': {'systemRole': 'Tu sei un esperto scrittore di email. Attieniti allo stile che ti ho fornito nelle instruction e inserici il contenuto richiesto. Genera il testo di una mail a partire da questo contenuto, con lo stile ricevuto in precedenza: ',
'systemStyle': 'Utilizza lo stile fornito come esempio e parla in ITALIANO e firmati sempre come il vostro collega Preside',
'instruction': URL_APP_SCRIPT + '1tEMxG0zJmmyh5PlAofKDkhbi1QGMOwPH',
'RAG': False},
'Decreti': {'systemRole': 'Tu sei il mio assistente per la ricerca documentale! Ti ho fornito una lista di documenti, devi cercare quello che ti chiedo nei documenti',
'systemStyle': 'Sii molto formale, sintetico e parla in ITALIANO',
'instruction': '',
'RAG': True}
}
st.set_page_config(page_title="ZucchettiAI", page_icon="🤖")
def init_state() :
if "messages" not in st.session_state:
st.session_state.messages = []
if "temp" not in st.session_state:
st.session_state.temp = 0.8
if "history" not in st.session_state:
st.session_state.history = []
if "top_k" not in st.session_state:
st.session_state.top_k = 5
if "repetion_penalty" not in st.session_state :
st.session_state.repetion_penalty = 1
if "chat_bot" not in st.session_state :
st.session_state.chat_bot = "Mixtral 8x7B v0.1"
if 'loaded_data' not in st.session_state:
st.session_state.loaded_data = False
if not st.session_state.loaded_data:
with st.spinner('Caricamento in corso...'):
options = requests.get(URL_APP_SCRIPT).json()
st.session_state.options = options
st.session_state.loaded_data = True
def sidebar():
def retrieval_settings() :
st.markdown("# Impostazioni Prompt")
st.session_state.selected_option_key = st.selectbox('Azione', list(st.session_state.options.keys()) + ['Personalizzata'])
st.session_state.selected_option = st.session_state.options.get(st.session_state.selected_option_key, {})
st.session_state.systemRole = st.session_state.selected_option.get('systemRole', '')
st.session_state.systemRole = st.text_area("Descrizione", st.session_state.systemRole, help='Ruolo del chatbot e descrizione dell\'azione che deve svolgere')
st.session_state.systemStyle = st.session_state.selected_option.get('systemStyle', '')
st.session_state.systemStyle = st.text_area("Stile", st.session_state.systemStyle, help='Descrizione dello stile utilizzato per generare il testo')
st.session_state.instruction = st.session_state.selected_option.get('instruction', '')
#st.session_state.instruction = st.text_area("Istruzioni", st.session_state.instruction, help='Testo di riferimento sul quale il modello si basa per generare il testo')
st.session_state.rag_enabled = st.session_state.selected_option.get('tipo', '')=='RAG'
if st.session_state.rag_enabled:
st.session_state.top_k = st.slider(label="Documenti da considerare", min_value=1, max_value=20, value=4, disabled=not st.session_state.rag_enabled)
st.markdown("---")
def model_settings() :
st.markdown("# Impostazioni Modello")
st.session_state.chat_bot = st.sidebar.radio('Seleziona Modello:', [key for key, value in CHAT_BOTS.items() ])
st.session_state.temp = st.slider(label="Creatività", min_value=0.0, max_value=1.0, step=0.1, value=0.9)
st.session_state.max_tokens = st.slider(label="Lunghezza Output", min_value = 64, max_value=2048, step= 32, value=1024)
with st.sidebar:
retrieval_settings()
model_settings()
st.markdown("""> **Creato da [Matteo Bergamelli] 🔗**""")
def header() :
st.title("ZucchettiAI")
with st.expander("Cos'è ZucchettiAI?"):
st.info("""ZucchettiAI Chat è un ChatBot personalizzato basato su un database vettoriale, funziona secondo il principio della Generazione potenziata da Recupero (RAG).
La sua funzione principale ruota attorno alla gestione di un ampio repository di manuali di Zucchetti e fornisce agli utenti risposte in linea con le loro domande.
Questo approccio garantisce una risposta più precisa sulla base della richiesta degli utenti.""")
def chat_box() :
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
def formattaPrompt(prompt, systemRole, systemStyle, instruction):
input_text = f'''
{{
"input": {{
"role": "system",
"content": "{systemRole}",
"style": "{systemStyle}"
}},
"messages": [
{{
"role": "instructions",
"content": "{instruction} ({systemStyle})"
}},
{{
"role": "user",
"content": "{prompt}"
}}
]
}}
'''
return input_text
def gen_augmented_prompt(prompt, top_k) :
links = ""
embedding = HuggingFaceEmbeddings(model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")
db = Chroma(persist_directory='./DB_Manuali', embedding_function=embedding)
docs = db.similarity_search(prompt, k=top_k)
links = []
context = ''
NomeCartellaOriginariaDB = 'Documenti\\'
for doc in docs:
testo = doc.page_content.replace('\n\n', '\n')
testo = testo.replace('***', '')
testo = testo.replace('DOMANDA: ', '')
testo = testo.replace('RISPOSTA:', '')
context += testo + '\n\n\n'
reference = doc.metadata["source"].replace(NomeCartellaOriginariaDB, '') + ' (Pag. ' + str(doc.metadata["page"]+1) + ')'
links.append((reference, testo))
generated_prompt = f"""
A PARTIRE DAL SEGUENTE CONTESTO: {docs},
----
RISPONDI ALLA SEGUENTE RICHIESTA: {prompt}
"""
return context, links
def generate_chat_stream(prompt) :
links = []
prompt_originale = prompt
if st.session_state.rag_enabled :
with st.spinner("Ricerca nei Manuali...."):
time.sleep(1)
st.session_state.instruction, links = gen_augmented_prompt(prompt=prompt_originale, top_k=st.session_state.top_k)
prompt = formattaPrompt(prompt, st.session_state.systemRole, st.session_state.systemStyle, st.session_state.instruction)
print(prompt)
with st.spinner("Generazione in corso...") :
time.sleep(1)
chat_stream = chat(prompt, st.session_state.history,chat_client=CHAT_BOTS[st.session_state.chat_bot] ,
temperature=st.session_state.temp, max_new_tokens=st.session_state.max_tokens)
return chat_stream, links
def stream_handler(chat_stream, placeholder) :
start_time = time.time()
full_response = ''
for chunk in chat_stream :
if chunk.token.text!='</s>' :
full_response += chunk.token.text
placeholder.markdown(full_response + "▌")
placeholder.markdown(full_response)
return full_response
def show_source(links) :
with st.expander("Mostra fonti") :
for link in links:
reference, testo = link
st.info('##### ' + reference.replace('_', ' ') + '\n\n'+ testo)
init_state()
sidebar()
header()
chat_box()
if prompt := st.chat_input("Chatta con ZucchettiAI..."):
st.chat_message("user").markdown(prompt)
st.session_state.messages.append({"role": "user", "content": prompt})
chat_stream, links = generate_chat_stream(prompt)
with st.chat_message("assistant"):
placeholder = st.empty()
full_response = stream_handler(chat_stream, placeholder)
if st.session_state.rag_enabled :
show_source(links)
st.session_state.messages.append({"role": "assistant", "content": full_response})
st.success('Generazione Completata')