app.py

#region# import libs
import streamlit as st
import os
from mistralai import Mistral
import numpy as np
# import fitz  # PyMuPDF pour extraction PDF
import faiss
import pickle
from mistralai import Mistral
# from sklearn.manifold import TSNE
# from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
# from dotenv import load_dotenv
 
MISTRAL_API_KEY = os.getenv("api_mistral")
model = "mistral-large-latest" #"ministral-8b-latest" # ancien model : 'mistral-large-latest'
mistral_client = Mistral(api_key=MISTRAL_API_KEY)
MAX_TOKENS = 1500
#endregion

#region# rag
model_embedding = "mistral-embed"
# 📌 Paramètres de segmentation
chunk_size = 256  # Réduction du chunk size pour un meilleur contrôle du contexte
chunk_overlap = 15
# 📌 Définition des chemins de stockage
index_path = "files/faiss_index.bin"
chunks_path = "files/chunked_docs.pkl"
metadata_path = "files/metadata.pkl"
embeddings_path = "files/embeddings.npy"

print("🔄 Chargement des données existantes...")
index = faiss.read_index(index_path)
with open(chunks_path, "rb") as f:
    chunked_docs = pickle.load(f)
with open(metadata_path, "rb") as f:
    metadata_list = pickle.load(f)
embeddings = np.load(embeddings_path)
print("✅ Index, chunks, embeddings et métadonnées chargés avec succès !")

# 📌 Récupération des chunks les plus pertinents
def retrieve_relevant_chunks(question, k=5):
    question_embedding_response = mistral_client.embeddings.create(
        model=model_embedding,
        inputs=[question],
    )
    question_embedding = np.array(question_embedding_response.data[0].embedding).astype('float32').reshape(1, -1)
    distances, indices = index.search(question_embedding, k)
    if len(indices[0]) == 0:
        print("⚠️ Avertissement : Aucun chunk pertinent trouvé, réponse possible moins précise.")
        return [], []
    return [chunked_docs[i] for i in indices[0]]
#endregion

#region# Définition des prompts
def generate_prompts(score:str, type: str, annee_min: str, annee_max:str, context) -> dict:
    """
    Genere les prefixes et suffixes des prompts pour Mistral en fonction du score de vulgarisation, du type d'espece, et les années des documents
    Args:
        score (str): 1 = vulgarisé, 2 = intermédiaire, 3 = technique
        type (str): 'Ponte' ou 'Chair'
        annee_min (str): annee min de publication
        annee_max (str): annee max de publication
    
    """

    if type == "Ponte":
        type_description = "volailles pondeuses"
    elif type == "Chair":
        type_description = "volailles de chair"
    else:
        raise ValueError("Type must be either 'Ponte' or 'Chair'")
 
    if score == "1":
        prefix_prompt = f"""Tu es un assistant IA spécialisé en nutrition de la volaille. Ton utilisateur est un chercheur travaillant sur
        l'amélioration des régimes alimentaires pour optimiser la santé et la croissance des {type_description}. Voici les informations extraites des documents à utiliser avec priorité : {context}.
        Réponds en vulgarisant les informations.
        Pour fournir la réponse, tu dois te baser sur des publications/articles qui ont une date de publication entre {annee_min} et {annee_max}."""
        suffix_prompt = """Réponds en français et donne une réponse directe et claire.
        N'inclus pas de bibliographie dans ta réponse. Intègre les numéros de tes sources dans le texte."""
    elif score == "2":
        prefix_prompt = f"""Tu es un assistant IA spécialisé en nutrition de la volaille. Ton utilisateur est un chercheur travaillant sur
        l'amélioration des régimes alimentaires pour optimiser la santé et la croissance des {type_description}. Voici les informations extraites des documents à utiliser avec priorité : {context}.
        Réponds en fournissant des explications claires et concises, adaptées à la question posée.
        Pour fournir la réponse, tu dois te baser sur des publications/articles qui ont une date de publication entre {annee_min} et {annee_max}.
        Tes réponses doivent être structurées, complètes et adaptées aux professionnels du secteur."""
        suffix_prompt = """Présente une réponse claire et concise. Utilise un ton
        professionnel, clair et rigoureux. Réponds en français.
        N'inclus pas de bibliographie dans ta réponse. Intègre les numéros de tes sources dans le texte."""
    elif score == "3":
        prefix_prompt = f"""Tu es un assistant IA spécialisé en nutrition de la volaille. Ton utilisateur est un chercheur travaillant sur
        l'amélioration des régimes alimentaires pour optimiser la santé et la croissance des {type_description}. Voici les informations extraites des documents à utiliser avec priorité : {context}.
        Réponds en fournissant des explications détaillées et précises, adaptées à la complexité de la question posée.
        N'oublie pas de citer à la fin de ta réponse les références sur lesquelles tu t'es basé avec son année (entre {annee_min} et {annee_max}).
        Tes réponses doivent être structurées, complètes et adaptées aux professionnels du secteur."""
        suffix_prompt = """Présente une réponse détaillée et complète. Utilise un ton
        professionnel, clair et rigoureux. Si possible, inclue des chiffres, des études ou des références pertinentes pour renforcer
        la crédibilité de la réponse. Réponds en français.
        N'inclus pas de bibliographie dans ta réponse. Intègre les numéros de tes sources dans le texte."""
    else:
        raise ValueError("Score must be 1, 2, or 3")
 
    return prefix_prompt, suffix_prompt

def send_prompt_to_mistral(type_reponse: str, user_prompt: str, temperature:int, n_comp:int, prefix_prompt: str, suffix_prompt:str, verbose=True) -> str:
    """
    Envoie un prompt à Mistral pour obtenir une réponse
    Args:
        type_reponse (str): Le rôle de l'utilisateur, peut être 'technicien' ou 'chercheur'.
                    Si le rôle ne correspond pas à l'un de ces deux, une exception sera levée.
        user_prompt (str): Le texte du prompt à envoyer à Mistral.
        temperature (int): Lower values make the model more deterministic, focusing on likely responses for accuracy
        verbose (bool): Print la reponse avant le return
        prefixe (str): Prefixe du prompt
        suffixe (str): Suffixe du prompt
    Returns:
        dict: La réponse du modèle Mistral à partir du prompt fourni.
    Raises:
        ValueError: Si le rôle spécifié n'est pas 'technicien' ou 'chercheur'.
    """


    # Création du message à envoyer à Mistral
    messages = [{"role": type_reponse, "content": suffix_prompt}]
    
    # Envoi de la requête à Mistral et récupération de la réponse
    chat_response = mistral_client.chat.complete(
        model = model,
        messages=[
            {"role": "system", "content": prefix_prompt},
            {"role": "user", "content": user_prompt + suffix_prompt},
        ],

        #response_format={
        #    'type': 'json_object'
        #},
        temperature=temperature,
        n=n_comp,
        max_tokens=MAX_TOKENS,
        stream=False
        #stop='\n'
    )

    if verbose:
        print(chat_response)

    return chat_response

def is_valid_mistral_response(response: dict) -> bool:
    """
    Vérifie si la réponse de l'API Mistral est valide.
    
    Critères de validité :
    - La clé "choices" existe et contient au moins un élément.
    - Le texte généré n'est pas vide et ne contient pas uniquement des tabulations ou espaces.
    - La génération ne s'est pas arrêtée pour une raison autre que 'stop'.
    - La réponse ne contient pas de texte générique inutile.
    
    :param response: Dictionnaire représentant la réponse de l'API Mistral
    :return: True si la réponse est valide, False sinon
    """
    if not isinstance(response, dict):
        return False
    
    choices = response.get("choices")
    if not choices or not isinstance(choices, list):
        return False
    
    first_choice = choices[0]
    if not isinstance(first_choice, dict):
        return False
    
    text = first_choice.get("message", {}).get("content", "").strip()
    if not text or text in ["\t\t", "", "N/A"]:
        return False
    
    finish_reason = first_choice.get("finish_reason", "")
    if finish_reason in ["error", "tool_calls"]:
        return False
    
    # Vérification du contenu inutile
    invalid_responses = [
        "Je suis une IA", "Désolé, je ne peux pas répondre", "Je ne sais pas"
    ]
    if any(invalid in text for invalid in invalid_responses):
        return False
    
    return True

def print_pretty_response(response: dict, verbose=True):

    pretty = response.choices[0].message.content

    if verbose:
        print(pretty)
    
    return pretty

def response_details(response, verbose=True):
    """
    Envoie les details techniques du prompt
    """

    details = {}
    details["id"] = response.id
    details["total_tokens"] = response.usage.total_tokens
    details["prefix"] = response.choices[0].message.prefix
    details["role"] = response.choices[0].message.role

    if verbose:
        print(details)

    return details

def prompt_pipeline(user_prompt: str, niveau_detail: str, type_reponse: str, souche: str, annee_publication_min: str, annee_publication_max: str, context) -> dict:
    """
    Fonction visible de l'application pour appeler un prompt et obtenir sa reponse
    Args: 
        prompt (str): Prompt utilisateur
        niveau_detail (str): Niveau de detail de la requete : 1, 2, 3. Plus haut = plus d'infos
        type_reponse (str): 'Ponte', 'Chair'
    Returns:
        Dict
    """

    prefix_prompt, suffix_prompt = generate_prompts(score=niveau_detail, type=type_reponse, annee_min=annee_publication_min, annee_max=annee_publication_max, context= context)

    reponse_mistral = send_prompt_to_mistral(
        type_reponse=type_reponse,
        user_prompt=user_prompt,
        temperature=0.05,
        n_comp=1,
        verbose=False,
        prefix_prompt=prefix_prompt,
        suffix_prompt=suffix_prompt
    )

    to_return = {}
    to_return["reponse_propre"] = print_pretty_response(reponse_mistral, verbose=True)
    to_return["details"] = response_details(reponse_mistral, verbose=False)
    to_return["prefix"] = str(prefix_prompt)

    return to_return

#endregion

#region# Titre de l'application et mise en page des éléments graphiques
st.set_page_config(page_title="VolAI", page_icon="🤖")
st.title("VolAI, le chatbot expert en nutrition de volailles")
st.sidebar.image("img/avril_logo_rvb.jpg")
st.sidebar.header("")


#Choix production
choix_prod = st.sidebar.pills(
    "Sur quelle espèce voulez-vous avoir des renseignements ?",
    ("Ponte", "Chair"),)
 
#Niveau vulgarisation
choix_vulgarisation = st.sidebar.pills(
    "Quel niveau de vulgarisation souhaitez-vous ?   (1- Vulgarisé  2-Intermédiaire  3-Technique)",
    ("1", "2", "3"),)

#Années de publication
choix_annee = st.sidebar.slider("Années de publication",
                                min_value=1980,
                                max_value=2025,
                                value=(2010,2025))
#endregion
 
#region# Interface utilisateur
user_input = st.text_area("Entrez votre question:", placeholder="E.g.: Quels sont les impact et la toxicité spécifique sur les volaille aux doses d’alkaloides tropaniques ?")
 
if st.button("Envoyer la question..."):
    if user_input and choix_prod and choix_vulgarisation and choix_annee :
        with st.spinner("Veuillez patienter quelques instants..."):
            # Génération de la réponse

            #todo mettre relevant chunks et context = 
            relevant_chunks= retrieve_relevant_chunks(user_input)
            # context = "\n".join([chunk["text"] for chunk in relevant_chunks])
            chunk_references = [f"[{i+1}]" for i in range(len(relevant_chunks))]
            context = "\n\n".join([f"{chunk_references[i]} (Source: {src['metadata']['source']}) :\n{src['text']}" for i, src in enumerate(relevant_chunks)])
           
            response0 = prompt_pipeline(
            user_prompt = user_input,
            niveau_detail=choix_vulgarisation,
            type_reponse=choix_prod,
            souche=None,
            context=context,
            annee_publication_max=max(choix_annee),
            annee_publication_min=min(choix_annee)
            )

            # st.markdown(f"""
            # <div style="border: 2px solid #453103; padding: 15px; border-radius: 10px;">
            #     {response0["prefix"]}
            # </div>
            # """, unsafe_allow_html=True)

            # print("prefix = ",response0["prefix"])
            
            # st.markdown("**Bot :** \\t " + bot_response)
            # Afficher un titre
            st.subheader("Réponse :")
 
            # Ajouter du texte Markdown avec un cadre
            st.markdown(f"""
            <div style="border: 2px solid #453103; padding: 15px; border-radius: 10px;">
                {response0["reponse_propre"]}
            </div>
            """, unsafe_allow_html=True)

            #print du contexte
            st.subheader("Sources :")
            st.markdown(f"""
            <div style="border: 2px solid #453103; padding: 15px; border-radius: 10px;">
                {context}
            </div>
            """, unsafe_allow_html=True)

            # #réponse sans contexte
            # response1 = prompt_pipeline(
            # user_prompt = user_input,
            # niveau_detail=choix_vulgarisation,
            # type_reponse=choix_prod,
            # souche=None,
            # context='',
            # annee_publication_max=max(choix_annee),
            # annee_publication_min=min(choix_annee)
            # )
            # st.subheader("Réponse sans contexte :")
            # st.markdown(f"""
            # <div style="border: 2px solid #453103; padding: 15px; border-radius: 10px;">
            #     {response1['reponse_propre']}
            # </div>
            # """, unsafe_allow_html=True)

            #encadré sources
            # # Afficher un titre
            # st.subheader("Sources :")
 
            # # Ajouter du texte Markdown avec un cadre
            # st.markdown(f"""
            # <div style="border: 2px solid #453103; padding: 15px; border-radius: 10px;">
            # {bot_response[1]}
            # </div>
            # """, unsafe_allow_html=True)

            #encadré Reviews
            # st.markdown("""
            # <div style="border: 2px solid #453103; padding: 15px; border-radius: 10px;">
            #     Reviews
            # </div>
            # """, unsafe_allow_html=True)

    else:
        if not user_input:
            st.warning("Veuillez entrer un message!")
        if not choix_prod or not choix_vulgarisation or not choix_annee:
            st.warning("Veuillez compléter les paramètres dans le bandeau latéral de gauche!")
#endregion