Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| from chat_client import chat | |
| from google_function import leggi_gmail | |
| from google_function import scrivi_bozza_gmail | |
| from google_function import leggi_calendario_google | |
| from google_function import connetti_google | |
| 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 | |
| import json | |
| from audio_recorder_streamlit import audio_recorder | |
| import speech_recognition as sr | |
| from googlesearch import search | |
| from bs4 import BeautifulSoup | |
| import PyPDF2 | |
| import pytesseract | |
| from PIL import Image | |
| from youtube_transcript_api import YouTubeTranscriptApi | |
| import webbrowser | |
| from streamlit_javascript import st_javascript | |
| load_dotenv() | |
| URL_APP_SCRIPT = os.getenv('URL_APP_SCRIPT') | |
| URL_PROMPT = URL_APP_SCRIPT + '?IdFoglio=1cLw9q70BsPmxMBj9PIzgXtq6sm3X-GVBVnOB5wE8jr8' | |
| URL_DOCUMENTI = URL_APP_SCRIPT + '?IdSecondoFoglio=1cLw9q70BsPmxMBj9PIzgXtq6sm3X-GVBVnOB5wE8jr8' | |
| SYSTEM_PROMPT = ["Sei BonsiAI e mi aiuterai nelle mie richieste (Parla in ITALIANO)", "Esatto, sono BonsiAI. Di cosa hai bisogno?"] | |
| CHAT_BOTS = {"Mixtral 8x7B v0.1" :"mistralai/Mixtral-8x7B-Instruct-v0.1"} | |
| option_personalizzata = {'Personalizzata': {'systemRole': 'Tu sei BONSI AI, il mio assistente personale della scuola superiore del Bonsignori. Aiutami in base alle mie esigenze', | |
| 'systemStyle': 'Firmati sempre come BONSI AI. (scrivi in italiano)', | |
| 'instruction': '', | |
| 'tipo': '', | |
| 'RAG': False} | |
| } | |
| option_leggiemail = {'Leggi Gmail': {'systemRole': 'Tu sei BONSI AI, il mio assistente personale della scuola superiore del Bonsignori. Effettua l operazione richiesta sulla base delle seguenti email: ', | |
| 'systemStyle': 'Firmati sempre come BONSI AI. (scrivi in italiano)', | |
| 'instruction': '', | |
| 'tipo': 'EMAIL', | |
| 'RAG': False} | |
| } | |
| option_leggicalendar = {'Leggi Calendar': {'systemRole': 'Tu sei BONSI AI, il mio assistente personale della scuola superiore del Bonsignori. Effettua l operazione richiesta sulla base dei seguenti eventi di calendario: ', | |
| 'systemStyle': 'Firmati sempre come BONSI AI. (scrivi in italiano)', | |
| 'instruction': '', | |
| 'tipo': 'CALENDAR', | |
| 'RAG': False} | |
| } | |
| def local_storage_get(key): | |
| return st_javascript(f"localStorage.getItem('{key}');") | |
| def local_storage_set(key, value): | |
| return st_javascript(f"localStorage.setItem('{key}', '{value}');") | |
| # ----------------------------------------------------------- Interfaccia -------------------------------------------------------------------- | |
| st.set_page_config(page_title="Bonsi A.I.", 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 = [SYSTEM_PROMPT] | |
| 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 "split" not in st.session_state: | |
| st.session_state.split = 30 | |
| if "enable_history" not in st.session_state: | |
| st.session_state.enable_history = True | |
| if "audio_bytes" not in st.session_state: | |
| st.session_state.audio_bytes = False | |
| if "cerca_online" not in st.session_state: | |
| st.session_state.cerca_online = False | |
| if "numero_siti" not in st.session_state: | |
| st.session_state.numero_siti = 3 | |
| if "numero_generazioni" not in st.session_state: | |
| st.session_state.numero_generazioni = 1 | |
| if "numero_elementi" not in st.session_state: | |
| st.session_state.numero_elementi = 10 | |
| if "testo_documenti" not in st.session_state: | |
| st.session_state.testo_documenti = '' | |
| if "uploaded_files" not in st.session_state: | |
| st.session_state.uploaded_files = None | |
| if "urls" not in st.session_state: | |
| st.session_state.urls = [""] * 5 | |
| if "creds" not in st.session_state: | |
| st.session_state.creds = None | |
| if "login_effettuato" not in st.session_state: | |
| st.session_state.login_effettuato = False | |
| if "tbs_options" not in st.session_state: | |
| st.session_state.tbs_options = { | |
| "Sempre": "0", | |
| "Ultimo anno": "qdr:y", | |
| "Ultimo mese": "qdr:m", | |
| "Ultima settimana": "qdr:w", | |
| "Ultimo giorno": "qdr:d" | |
| } | |
| if not st.session_state.loaded_data and st.session_state.login_effettuato == True: | |
| place=st.empty() | |
| place=st.empty() | |
| with place: | |
| with st.status("Caricamento in corso...", expanded=True) as status: | |
| st.write("Inizializzazione Ambiente") | |
| time.sleep(1) | |
| st.write("Inizializzazione Prompt") | |
| options = requests.get(URL_PROMPT).json() | |
| st.write("Inizializzazione Documenti") | |
| documenti = requests.get(URL_DOCUMENTI).json() | |
| st.session_state.options = {**option_personalizzata, **option_leggiemail, **option_leggicalendar, **options} | |
| st.session_state.documenti = documenti | |
| st.session_state.loaded_data = True | |
| status.update(label="Caricamento Completato", state="complete", expanded=False) | |
| place.empty() | |
| def read_text_from_file(file): | |
| text = "" | |
| if file.name.endswith(".txt"): | |
| text = file.read().decode("utf-8") | |
| elif file.name.endswith(".pdf"): | |
| pdf_reader = PyPDF2.PdfReader(file) | |
| for page_num in range(len(pdf_reader.pages)): | |
| page = pdf_reader.pages[page_num] | |
| text += page.extract_text() | |
| else: | |
| try: | |
| image = Image.open(file) | |
| text = pytesseract.image_to_string(image) | |
| except: | |
| st.write(f"Non è possibile leggere il testo dal file '{file.name}'.") | |
| return text | |
| def sidebar(): | |
| def retrieval_settings() : | |
| st.markdown("# Impostazioni Prompt") | |
| st.session_state.selected_option_key = st.selectbox('Azione', list(st.session_state.options.keys())) | |
| st.session_state.selected_option = st.session_state.options.get(st.session_state.selected_option_key, {}) | |
| if st.session_state.options.get(st.session_state.selected_option_key, {})["tipo"]=='DOCUMENTO': | |
| st.session_state.selected_documento_key = st.selectbox('Documento', list(st.session_state.documenti.keys())) | |
| st.session_state.selected_documento = st.session_state.documenti.get(st.session_state.selected_documento_key, {}) | |
| st.session_state.instruction = st.session_state.selected_documento.get('instruction', '')['Testo'] | |
| st.session_state.split = st.slider(label="Pagine Suddivisione", min_value=1, max_value=30, value=30, help='Se il documento ha 100 pagine e suddivido per 20 pagine elaborerà la risposta 5 volte. Più alto è il numero e meno volte elaborerà ma la risposta sarà più imprecisa') | |
| else: | |
| st.session_state.instruction = st.session_state.selected_option.get('instruction', '') | |
| 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') | |
| if st.session_state.selected_option["tipo"]=='EMAIL': | |
| st.session_state.numero_elementi = st.slider(label="Numero Email", min_value=1, max_value=100, value=10) | |
| if st.session_state.selected_option["tipo"]=='CALENDAR': | |
| st.session_state.numero_elementi = st.slider(label="Numero Eventi Calendario", min_value=1, max_value=100, value=10) | |
| st.session_state.rag_enabled = st.session_state.selected_option.get('tipo', '')=='RAG' | |
| if st.session_state.selected_option_key == 'Decreti': | |
| st.session_state.top_k = st.slider(label="Documenti da ricercare", min_value=1, max_value=20, value=4, disabled=not st.session_state.rag_enabled) | |
| st.session_state.decreti_escludere = st.multiselect( | |
| 'Decreti da escludere', | |
| ['23.10.2 destinazione risorse residue pnrr DGR 1051-2023_Destinazione risorse PNRR Duale.pdf', '23.10.25 accompagnatoria Circolare Inail assicurazione.pdf', '23.10.26 circolare Inail assicurazione.pdf', '23.10.3 FAQ in attesa di avviso_.pdf', '23.11.2 avviso 24_24 Decreto 17106-2023 Approvazione Avviso IeFP 2023-2024.pdf', '23.5.15 decreto linee inclusione x enti locali.pdf', '23.6.21 Circolare+esplicativa+DGR+312-2023.pdf', '23.7.3 1° Decreto R.L. 23_24 .pdf', '23.9 Regolamento_prevenzione_bullismo_e_cyberbullismo__Centro_Bonsignori.pdf', '23.9.1 FAQ inizio anno formativo.pdf', '23.9.15 DECRETO VERIFICHE AMMINISTR 15-09-23.pdf', '23.9.4 modifica decreto GRS.pdf', '23.9.8 Budget 23_24.pdf', '24.10.2022 DECRETO loghi N.15176.pdf', 'ALLEGATO C_Scheda Supporti al funzionamento.pdf', 'ALLEGATO_ B_ Linee Guida.pdf', 'ALLEGATO_A1_PEI_INFANZIA.pdf', 'ALLEGATO_A2_PEI_PRIMARIA.pdf', 'ALLEGATO_A3_PEI_SEC_1_GRADO.pdf', 'ALLEGATO_A4_PEI_SEC_2_GRADO.pdf', 'ALLEGATO_C_1_Tabella_Fabbisogni.pdf', 'Brand+Guidelines+FSE+.pdf', 'Decreto 20797 del 22-12-2023_Aggiornamento budget PNRR.pdf', 'Decreto 20874 del 29-12-2023 Avviso IeFP PNRR 2023-2024_file unico.pdf'], | |
| []) | |
| st.session_state.uploaded_files = st.file_uploader("Importa file", accept_multiple_files=True) | |
| st.session_state.testo_documenti = '' | |
| for uploaded_file in st.session_state.uploaded_files: | |
| text_doc = read_text_from_file(uploaded_file) | |
| st.session_state.testo_documenti += text_doc | |
| print(st.session_state.testo_documenti) | |
| st.markdown("---") | |
| st.markdown("# Ricerca Online") | |
| st.session_state.cerca_online = st.toggle("Attivata", value=False) | |
| with st.popover("Siti Specifici", disabled=not st.session_state.cerca_online,use_container_width=True): | |
| st.markdown("#### Inserisci Siti Web ") | |
| for i in range(5): | |
| st.session_state.urls[i] = st.text_input(f"URL Sito {i+1}", placeholder='Sito Web...', help='è possibile specificare anche il link di un video Youtube, in tal caso verrà restituita la trascrizione del video') | |
| st.session_state.selected_tbs = st.selectbox("Periodo:", list(st.session_state.tbs_options.keys()), disabled=(not st.session_state.cerca_online) or (st.session_state.urls[0]!="")) | |
| st.session_state.tbs_value = st.session_state.tbs_options[st.session_state.selected_tbs] | |
| st.session_state.numero_siti = st.slider(label="Risultati", min_value = 1, max_value=20, value=3, disabled=(not st.session_state.cerca_online) or (st.session_state.urls[0]!="")) | |
| #st.session_state.suddividi_ricerca = st.toggle("Attivata", value=False) | |
| st.markdown("---") | |
| def model_settings(): | |
| st.markdown("# Impostazioni Modello") | |
| st.session_state.chat_bot = st.sidebar.radio('Modello:', [key for key, value in CHAT_BOTS.items() ]) | |
| st.session_state.numero_generazioni = st.slider(label="Generazioni", min_value = 1, max_value=10, value=1) | |
| st.session_state.enable_history = st.toggle("Storico Messaggi", value=True) | |
| 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 = 2, max_value=2048, step= 32, value=1024) | |
| with st.sidebar: | |
| retrieval_settings() | |
| model_settings() | |
| st.markdown("""> **Creato da Matteo Bergamelli **""") | |
| def audioRec(): | |
| st.session_state.audio_bytes = audio_recorder(text='', icon_size="3x") | |
| if st.session_state.audio_bytes: | |
| with open("./AUDIO.wav", "wb") as file: | |
| file.write(st.session_state.audio_bytes) | |
| wav = sr.AudioFile("./AUDIO.wav") | |
| with wav as source: | |
| recognizer_instance = sr.Recognizer() | |
| recognizer_instance.pause_threshold = 3.0 | |
| audio = recognizer_instance.listen(source) | |
| print("Ok! sto ora elaborando il messaggio!") | |
| try: | |
| text = recognizer_instance.recognize_google(audio, language="it-IT") | |
| print(text) | |
| js = f""" | |
| <script> | |
| var chatInput = parent.document.querySelector('textarea[data-testid="stChatInput"]'); | |
| var nativeInputValueSetter = Object.getOwnPropertyDescriptor(window.HTMLTextAreaElement.prototype, "value").set; | |
| nativeInputValueSetter.call(chatInput, "{text}"); | |
| var event = new Event('input', {{ bubbles: true}}); | |
| chatInput.dispatchEvent(event); | |
| var sendChat = parent.document.getElementsByClassName("st-emotion-cache-1621d17")[0] | |
| sendChat.click(); | |
| var x = parent.document.querySelector('[title="st.iframe"]'); | |
| x.style.display = "none"; | |
| </script> | |
| """ | |
| st.components.v1.html(js) | |
| except Exception as e: | |
| print(e) | |
| def header() : | |
| st.title("Bonsi A.I.", anchor=False) | |
| with st.expander("Cos'è BonsiAI?"): | |
| st.info("""BonsiAI 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 documenti BonsiAI 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): | |
| if st.session_state.cerca_online: | |
| systemRole += '. Ti ho fornito una lista di materiali nelle instruction. Devi rispondere sulla base delle informazioni fonrnite!' | |
| 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_Decreti', embedding_function=embedding) | |
| docs = db.similarity_search(prompt, k=top_k) | |
| links = [] | |
| context = '' | |
| NomeCartellaOriginariaDB = 'Documenti_2\\' | |
| for doc in docs: | |
| testo = doc.page_content.replace('\n', ' ') | |
| context += testo + '\n\n\n' | |
| reference = doc.metadata["source"].replace(NomeCartellaOriginariaDB, '') + ' (Pag. ' + str(doc.metadata["page"]) + ')' | |
| links.append((reference, testo)) | |
| return context, links | |
| def get_search_results_int(url): | |
| result = {'title': '', 'description': '', 'url': '', 'body': ''} | |
| try: | |
| if "www.youtube.com" in url: | |
| video_id = url.split("=")[1] | |
| title = 'Video Youtube' | |
| description = '' | |
| transcript = YouTubeTranscriptApi.get_transcript(video_id) | |
| body_content = " ".join([segment["text"] for segment in transcript]) | |
| print(video_id) | |
| print(body_content) | |
| result = {'title': title, 'description': body_content, 'url': url, 'body': body_content} | |
| else: | |
| response = requests.get(url) | |
| soup = BeautifulSoup(response.text, 'html.parser') | |
| title = soup.title.string if soup.title else "N/A" | |
| description = soup.find('meta', attrs={'name': 'description'})['content'] if soup.find('meta', attrs={'name': 'description'}) else "N/A" | |
| body_content = soup.find('body').get_text() if soup.find('body') else "N/A" | |
| result = {'title': title, 'description': description, 'url': url, 'body': body_content} | |
| except Exception as e: | |
| print(f"Error fetching data from {url}: {e}") | |
| return result | |
| def get_search_results(query, top_k): | |
| results = [] | |
| if st.session_state.urls[0] != "": | |
| for i in range(5): | |
| url = st.session_state.urls[i] | |
| if url != "": | |
| results.append(get_search_results_int(url)) | |
| else: | |
| for url in search(query, num=top_k, stop=top_k, tbs=st.session_state.tbs_value): | |
| results.append(get_search_results_int(url)) | |
| return results | |
| def gen_online_prompt(prompt, top_k) : | |
| links = [] | |
| context = '' | |
| results = get_search_results(prompt, top_k) | |
| for i, result in enumerate(results, start=1): | |
| context += result['title'] + '\n' + result['description'] + '\n' + '\n\n' + result['body'].replace('\n','.') + '\n\n------------------------------------------------------------' | |
| links.append((str(i) + '. ' + result['title'], result['description'] + '\n\n' + result['url'])) | |
| return context, links | |
| def generate_chat_stream(prompt) : | |
| 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 | |
| def inserisci_istruzioni(prompt_originale): | |
| links = [] | |
| if st.session_state.cerca_online: | |
| with st.spinner("Ricerca Online...."): | |
| time.sleep(1) | |
| st.session_state.instruction, links = gen_online_prompt(prompt=prompt_originale, top_k=st.session_state.numero_siti) | |
| if st.session_state.rag_enabled : | |
| with st.spinner("Ricerca nei Decreti...."): | |
| time.sleep(1) | |
| st.session_state.instruction, links = gen_augmented_prompt(prompt=prompt_originale, top_k=st.session_state.top_k) | |
| if st.session_state.selected_option["tipo"]=='EMAIL': | |
| with st.spinner("Ricerca nelle Email...."): | |
| time.sleep(1) | |
| st.session_state.instruction, links = leggi_gmail(max_results=st.session_state.numero_elementi) | |
| if st.session_state.selected_option["tipo"]=='CALENDAR': | |
| with st.spinner("Ricerca nel Calendario...."): | |
| time.sleep(1) | |
| st.session_state.instruction, links = leggi_calendario_google(max_results=st.session_state.numero_elementi) | |
| with st.spinner("Generazione in corso...") : | |
| time.sleep(1) | |
| #st.session_state.instruction = instruction_originale + '\n----------------------------------------------\n' + st.session_state.instruction | |
| return links | |
| def stream_handler(chat_stream, placeholder) : | |
| 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) | |
| def split_text(text, chunk_size): | |
| testo_suddiviso = [] | |
| if text == '': | |
| text = ' ' | |
| if chunk_size < 100: | |
| chunk_size = 60000 | |
| for i in range(0, len(text), chunk_size): | |
| testo_suddiviso.append(text[i:i+chunk_size]) | |
| return testo_suddiviso | |
| init_state() | |
| if st.session_state.login_effettuato == False: | |
| connetti_google() | |
| if st.session_state.login_effettuato == True: | |
| st_javascript("localStorage.removeItem('token');") | |
| init_state() | |
| sidebar() | |
| header() | |
| chat_box() | |
| if prompt := st.chat_input("Chatta con BonsiAI..."): | |
| prompt_originale = prompt | |
| links = inserisci_istruzioni(prompt_originale) | |
| st.session_state.instruction+= ' \n\n' + st.session_state.testo_documenti | |
| instruction_suddivise = split_text(st.session_state.instruction, st.session_state.split*2000) | |
| ruolo_originale = st.session_state.systemRole | |
| ruoli_divisi = ruolo_originale.split("&&") | |
| parte=1 | |
| i=1 | |
| risposta_completa = '' | |
| for ruolo_singolo in ruoli_divisi: | |
| for instruction_singola in instruction_suddivise: | |
| for numgen in range(1, st.session_state.numero_generazioni+1): | |
| if i==1: | |
| st.chat_message("user").markdown(prompt_originale + (': Parte ' + str(parte) if i > 1 else '')) | |
| i+=1 | |
| prompt = formattaPrompt(prompt_originale, ruolo_singolo, st.session_state.systemStyle, instruction_singola) | |
| print('------------------------------------------------------------------------------------') | |
| print(prompt) | |
| st.session_state.messages.append({"role": "user", "content": prompt_originale}) | |
| chat_stream = 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 or st.session_state.cerca_online or st.session_state.selected_option["tipo"]=='EMAIL' or st.session_state.selected_option["tipo"]=='CALENDAR': | |
| show_source(links) | |
| if st.session_state.options.get(st.session_state.selected_option_key, {})["tipo"]=='DOCUMENTO': | |
| with st.expander("Mostra Documento") : | |
| st.info('##### ' + st.session_state.selected_documento_key + ' (Parte ' + str(parte) +')'+ '\n\n\n' + instruction_singola) | |
| parte+=1 | |
| st.session_state.messages.append({"role": "assistant", "content": full_response}) | |
| risposta_completa = risposta_completa + '\n' + full_response | |
| if st.session_state.enable_history: | |
| st.session_state.history.append([prompt_originale, full_response]) | |
| else: | |
| st.session_state.history.append(['', '']) | |
| st.success('Generazione Completata') | |
| payload = {"domanda": prompt_originale, "risposta": risposta_completa} | |
| json_payload = json.dumps(payload) | |
| response = requests.post(URL_APP_SCRIPT, data=json_payload) |