app.py

from llama_index.core import Settings, VectorStoreIndex, StorageContext, load_index_from_storage
from llama_index.core.embeddings import BaseEmbedding
from llama_index.llms.mistralai import MistralAI
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core import SimpleDirectoryReader
from llama_index.core import PromptTemplate
# from pydantic import PrivateAttr
# import requests
from typing import List, Optional, Union
# from llama_index.core.embeddings.utils import BaseEmbedding  
from llama_index.embeddings.huggingface import HuggingFaceInferenceAPIEmbedding
# from llama_index.embeddings.huggingface import HuggingFaceEmbedding
import streamlit as st
from functools import lru_cache
import pickle
import os


mistral_api_key = os.getenv("mistral_api_key")

class QASystem:
    def __init__(self, 
                 mistral_api_key: str = mistral_api_key,
                 data_dir: str = "./data",
                 storage_dir: str = "./index_llama_136_multilingual-e5-large",
                 model_temperature: float = 0.002):
        self.data_dir = data_dir
        self.storage_dir = storage_dir
        
        # Initialize embedding model with API
        # api_key = 
        self.embedding_model = HuggingFaceInferenceAPIEmbedding(
                                    model_name="intfloat/multilingual-e5-large",
                                    )
        # self.embedding_model = HuggingFaceEmbedding(model_name="intfloat/multilingual-e5-large",trust_remote_code=True)
        
        self.llm = MistralAI(
            model="mistral-large-latest", 
            api_key=mistral_api_key, 
            temperature=model_temperature,
            max_tokens=1024
        )
        self._configure_settings()

        # self.create_index()

        self.index = self.load_index()  # Define index here 
        
    def _configure_settings(self):
        Settings.llm = self.llm
        Settings.embed_model = self.embedding_model

    def create_index(self):
        print("creating index")
        documents = SimpleDirectoryReader(self.data_dir).load_data()
        node_parser = SentenceSplitter(chunk_size=206, chunk_overlap=0)
        nodes = node_parser.get_nodes_from_documents(documents, show_progress=True)
        
        sentence_index = VectorStoreIndex(nodes, show_progress=True)
        sentence_index.storage_context.persist(self.storage_dir)

        # # Save the index to a pickle file  
        # with open(f"{self.storage_dir}/index.pkl", "wb") as f:  
        #     pickle.dump(sentence_index, f) 
        
        return sentence_index
    
    def load_index(self):
        # with open(f'{self.storage_dir}/index.pkl', 'rb') as f:  
        #     sentence_index = pickle.load(f)

        storage_context = StorageContext.from_defaults(persist_dir=self.storage_dir)
        return load_index_from_storage(storage_context, embed_model=self.embedding_model)
    
    def create_query_engine(self):
        template = """
        استخدم المعلومات التالية للإجابة على السؤال في النهاية. إذا لم تكن تعرف الإجابة، فقل فقط أنك لا تعرف، لا تحاول اختلاق إجابة.

        {context_str}
        السؤال: {query_str}
        الإجابة بالعربية:
        """
        prompt = PromptTemplate(template=template)
        
        return self.index.as_query_engine(
            similarity_top_k=10,
            streaming=True,
            text_qa_template=prompt,
            response_mode="tree_summarize", #tree_summarize, simple_summarize, compact
            embed_model=self.embedding_model
        )
    
    def query(self, question: str):
        query_engine = self.create_query_engine()
        response = query_engine.query(question)
        return  response#.print_response_stream()


# Utilisation de singleton pour éviter les réinitialisations multiples  
# @st.cache_resource  
@lru_cache(maxsize=1000)
def get_qa_system():  
    return QASystem() 

def main():  
    st.markdown("""  
    <style>  
        @import url('https://fonts.googleapis.com/css2?family=Noto+Kufi+Arabic:wght@100;200;300;400;500;600;700;800;900&display=swap');

        /* Application globale de la police */
        * {
            font-family: 'Noto Kufi Arabic', sans-serif !important;
        }
        
        body {  
            text-align: right;
            font-family: 'Noto Kufi Arabic', sans-serif !important;
        }  
        
        /* Style pour tous les textes */
        p, div, span, button, input, label, h1, h2, h3, h4, h5, h6 {
            font-family: 'Noto Kufi Arabic', sans-serif !important;
        }

        /* Titre principal avec taille réduite */
        h1 {
            font-size: 1.2em !important;
            margin-bottom: 0.5em !important;
            text-align: center;
            padding: 0.3px;
        }

        .css-1h9b9rq.e1tzin5v0 {
            direction: rtl; 
            text-align: right;
            font-family: 'Noto Kufi Arabic', sans-serif !important;
        }  
        
        /* Style pour l'expandeur */
        .streamlit-expanderContent, div[data-testid="stExpander"] {
            direction: rtl !important;
            text-align: right !important;
            font-family: 'Noto Kufi Arabic', sans-serif !important;
        }
        
        /* Style pour les boutons de l'expandeur */
        button[kind="secondary"] {
            direction: rtl !important;
            text-align: right !important;
            width: 100% !important;
            font-family: 'Noto Kufi Arabic', sans-serif !important;
            font-weight: 30 !important;
        }
        
        /* Style pour tous les éléments de texte */
        p, div {
            direction: rtl !important;
            text-align: right !important;
            font-family: 'Noto Kufi Arabic', sans-serif !important;
        }
        
        /* Style pour les bullet points */
        ul, li {
            direction: rtl !important;
            text-align: right !important;
            margin-right: 20px !important;
            margin-left: 0 !important;
            font-family: 'Noto Kufi Arabic', sans-serif !important;
        }
        
        .stTextInput, .stButton {  
            margin-left: auto;  
            margin-right: 0;  
            font-family: 'Noto Kufi Arabic', sans-serif !important;
        }  
        
        .stTextInput {  
            width: 100% !important;   
            direction: rtl; 
            text-align: right;
            font-family: 'Noto Kufi Arabic', sans-serif !important;
        }
        
        /* Force RTL sur tous les conteneurs */
        .element-container, .stMarkdown {
            direction: rtl !important;
            text-align: right !important;
            font-family: 'Noto Kufi Arabic', sans-serif !important;
        }
        
        /* Style spécifique pour l'expandeur des sources */
        .css-1fcdlhc, .css-1629p8f {
            direction: rtl !important;
            text-align: right !important;
            font-family: 'Noto Kufi Arabic', sans-serif !important;
        }

        /* Style pour le titre */
        .stTitle {
            font-family: 'Noto Kufi Arabic', sans-serif !important;
            font-weight: 700 !important;
        }

        /* Style pour les boutons */
        .stButton>button {
            font-family: 'Noto Kufi Arabic', sans-serif !important;
            font-weight: 500 !important;
        }

        /* Style pour les champs de texte */
        .stTextInput>div>div>input {
            font-family: 'Noto Kufi Arabic', sans-serif !important;
        }
    </style>  
    """, unsafe_allow_html=True)  
    
    st.title("هذا تطبيق للاجابة عن الاسئلة المتعلقة بالقانون المغربي ")  
    st.title("حاليا يضم 136 قانونا")
    
    qa_system = get_qa_system()  
    
    question = st.text_input("اطرح سؤالك :",placeholder=None)  
    
    if st.button("بحث"):  
        if question:  
            response_container = st.empty()  
            def stream_response(token):  
                if 'current_response' not in st.session_state:  
                    st.session_state.current_response = ""  
                st.session_state.current_response += token  
                response_container.markdown(st.session_state.current_response, unsafe_allow_html=True)  
            
            try:  
                query_engine = qa_system.create_query_engine()  
                st.session_state.current_response = ""  
                response = query_engine.query(question)  
                
                full_response = ""  
                for token in response.response_gen:  
                    full_response += token  
                    stream_response(token)  
                
                if hasattr(response, 'source_nodes'):  
                    st.markdown("""
                        <div style="direction: rtl !important; text-align: right !important; font-family: 'Noto Kufi Arabic', sans-serif !important;">
                            <div class="streamlit-expanderHeader">
                                المصادر
                            </div>
                        </div>
                    """, unsafe_allow_html=True)
                    with st.expander(""):  
                        for node in response.source_nodes:  
                            st.markdown(f"""
                                <div style="direction: rtl !important; text-align: right !important; font-family: 'Noto Kufi Arabic', sans-serif !important;">
                                    <p style="text-align: right !important;">مصادر الجواب : {node.metadata.get('file_name', 'Unknown')}</p>
                                    <p style="text-align: right !important;">Extrait: {node.text[:]}</p>
                                </div>
                            """, unsafe_allow_html=True)
            except Exception as e:  
                st.error(f"Une erreur s'est produite : {e}")  

if __name__ == "__main__":  
    main()