app.py

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_CARTELLA = os.getenv('URL_CARTELLA')

CHAT_BOTS = {"Mixtral 8x7B v0.1" :"mistralai/Mixtral-8x7B-Instruct-v0.1"}
SYSTEM_PROMPT = ["Sei BonsiAI e mi aiuterai nelle mie richieste (Parla in ITALIANO)", "Esatto, sono BonsiAI. Di cosa hai bisogno?"]
IDENTITY_CHANGE = ["Sei BonsiAI da ora in poi!", "Certo farò del mio meglio"]
options = {
    '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_CARTELLA + '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_CARTELLA + '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="BonsiAI", page_icon="🤖")

def gen_augmented_prompt(prompt, top_k) :   
    links = ""
    persist_directory1 = './DB_Decreti'
    embedding = HuggingFaceEmbeddings(model_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2")
    db = Chroma(persist_directory=persist_directory1, embedding_function=embedding) 
    context = db.similarity_search(prompt, k=top_k)
    print(context)
    generated_prompt = f"""
    A PARTIRE DAL SEGUENTE CONTESTO: {context},

    ----
    RISPONDI ALLA SEGUENTE RICHIESTA: {prompt}
    """
    return generated_prompt, links

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"

def sidebar() :
    def retrieval_settings() :
        st.markdown("# Impostazioni")
        st.session_state.selected_option_key = st.selectbox('Azione', list(options.keys()) + ['Personalizzata'])
        st.session_state.selected_option = 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)
        st.session_state.systemStyle = st.session_state.selected_option.get('systemStyle', '')
        st.session_state.systemStyle = st.text_area("Stile", st.session_state.systemStyle)
        st.session_state.instruction = st.session_state.selected_option.get('instruction', '')
        st.session_state.rag_enabled = st.session_state.selected_option.get('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.markdown("---")
    
    def model_settings() :
        #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("BonsiAI")
    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 instruction.startswith("http"):
        try:
            with st.spinner("Ricerca in Drive...") :
                resp = requests.get(instruction)
                resp.raise_for_status() 
                instruction = resp.text
        except requests.exceptions.RequestException as e:
            instruction = ""
    input_text = f'''
    {{
      "input": {{
          "role": "system",
          "content": "{systemRole}", 
          "style": "{systemStyle} Attenzione! Il testo generato deve essere lungo {st.session_state.max_tokens*2} CARATTERI"
      }},
      "messages": [
          {{
              "role": "instructions",
              "content": "{instruction} ({systemStyle})"
          }},
          {{
              "role": "user",
              "content": "{prompt}"
          }}
      ]
    }}
    '''
    return input_text

def generate_chat_stream(prompt) :
    links = []
    prompt = formattaPrompt(prompt, st.session_state.systemRole, st.session_state.systemStyle, st.session_state.instruction)
    print(prompt)
    if st.session_state.rag_enabled :
        with st.spinner("Ricerca nei documenti...."):
            time.sleep(1)
            prompt, links = gen_augmented_prompt(prompt=prompt, top_k=st.session_state.top_k)        
    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 i, link in enumerate(links) :
            st.info(f"{link}")

init_state()
sidebar()
header()
chat_box()

if prompt := st.chat_input("Chatta con BonsiAI..."):
    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.history.append([prompt, full_response])
    st.session_state.messages.append({"role": "assistant", "content": full_response})
    st.success('Generazione Completata')