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!='' : 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')