app.py

import gradio as gr
from langchain_mistralai.chat_models import ChatMistralAI
from langchain.prompts import ChatPromptTemplate
from langchain_deepseek import ChatDeepSeek
from langchain_google_genai import ChatGoogleGenerativeAI
import os
from pathlib import Path
import json
import faiss
import numpy as np
from langchain.schema import Document
import pickle
import re
import requests
from functools import lru_cache
import torch
from sentence_transformers import SentenceTransformer
from sentence_transformers.cross_encoder import CrossEncoder
import threading
from queue import Queue
import concurrent.futures
from typing import Generator, Tuple
import time

class OptimizedRAGLoader:
    def __init__(self,
                 docs_folder: str = "./docs",
                 splits_folder: str = "./splits",
                 index_folder: str = "./index"):
        
        self.docs_folder = Path(docs_folder)
        self.splits_folder = Path(splits_folder)
        self.index_folder = Path(index_folder)
        
        # Create folders if they don't exist
        for folder in [self.splits_folder, self.index_folder]:
            folder.mkdir(parents=True, exist_ok=True)
            
        # File paths
        self.splits_path = self.splits_folder / "splits.json"
        self.index_path = self.index_folder / "faiss.index"
        self.documents_path = self.index_folder / "documents.pkl"
        
        # Initialize components
        self.index = None
        self.indexed_documents = None
        
        # Initialize encoder model
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.encoder = SentenceTransformer("intfloat/multilingual-e5-large")
        self.encoder.to(self.device)
        self.reranker = model = CrossEncoder("cross-encoder/mmarco-mMiniLMv2-L12-H384-v1",trust_remote_code=True)
        
        # Initialize thread pool
        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
        
        # Initialize response cache
        self.response_cache = {}
        
    @lru_cache(maxsize=1000)
    def encode(self, text: str):
        """Cached encoding function"""
        with torch.no_grad():
            embeddings = self.encoder.encode(
                text,
                convert_to_numpy=True,
                normalize_embeddings=True
            )
        return embeddings
    
    def batch_encode(self, texts: list):
        """Batch encoding for multiple texts"""
        with torch.no_grad():
            embeddings = self.encoder.encode(
                texts,
                batch_size=32,
                convert_to_numpy=True,
                normalize_embeddings=True,
                show_progress_bar=False
            )
        return embeddings

    def load_and_split_texts(self):
        if self._splits_exist():
            return self._load_existing_splits()
            
        documents = []
        futures = []
        
        for file_path in self.docs_folder.glob("*.txt"):
            future = self.executor.submit(self._process_file, file_path)
            futures.append(future)
            
        for future in concurrent.futures.as_completed(futures):
            documents.extend(future.result())
            
        self._save_splits(documents)
        return documents
    
    def _process_file(self, file_path):
        with open(file_path, 'r', encoding='utf-8') as file:
            text = file.read()
            chunks = [s.strip() for s in re.split(r'(?<=[.!?])\s+', text) if s.strip()]
            
            return [
                Document(
                    page_content=chunk,
                    metadata={
                        'source': file_path.name,
                        'chunk_id': i,
                        'total_chunks': len(chunks)
                    }
                )
                for i, chunk in enumerate(chunks)
            ]

    def load_index(self) -> bool:
        """
        Charge l'index FAISS et les documents associés s'ils existent

        Returns:
            bool: True si l'index a été chargé, False sinon
        """
        if not self._index_exists():
            print("Aucun index trouvé.")
            return False

        print("Chargement de l'index existant...")
        try:
            # Charger l'index FAISS
            self.index = faiss.read_index(str(self.index_path))

            # Charger les documents associés
            with open(self.documents_path, 'rb') as f:
                self.indexed_documents = pickle.load(f)

            print(f"Index chargé avec {self.index.ntotal} vecteurs")
            return True

        except Exception as e:
            print(f"Erreur lors du chargement de l'index: {e}")
            return False

    def create_index(self, documents=None):
        if documents is None:
            documents = self.load_and_split_texts()
            
        if not documents:
            return False
            
        texts = [doc.page_content for doc in documents]
        embeddings = self.batch_encode(texts)
        
        dimension = embeddings.shape[1]
        self.index = faiss.IndexFlatL2(dimension)
        
        if torch.cuda.is_available():
            # Use GPU for FAISS if available
            res = faiss.StandardGpuResources()
            self.index = faiss.index_cpu_to_gpu(res, 0, self.index)
            
        self.index.add(np.array(embeddings).astype('float32'))
        self.indexed_documents = documents
        
        # Save index and documents
        cpu_index = faiss.index_gpu_to_cpu(self.index) if torch.cuda.is_available() else self.index
        faiss.write_index(cpu_index, str(self.index_path))
        
        with open(self.documents_path, 'wb') as f:
            pickle.dump(documents, f)
            
        return True

    def _index_exists(self) -> bool:
        """Vérifie si l'index et les documents associés existent"""
        return self.index_path.exists() and self.documents_path.exists()

    def get_retriever(self, k: int = 10):
        if self.index is None:
            if not self.load_index():
                if not self.create_index():
                    raise ValueError("Unable to load or create index")

        def retriever_function(query: str) -> list:
            # Check cache first
            cache_key = f"{query}_{k}"
            if cache_key in self.response_cache:
                return self.response_cache[cache_key]

            query_embedding = self.encode(query)
            
            distances, indices = self.index.search(
                np.array([query_embedding]).astype('float32'),
                k
            )
            
            results = [
                self.indexed_documents[idx]
                for idx in indices[0]
                if idx != -1
            ]
            
            # Cache the results
            self.response_cache[cache_key] = results
            return results
            
        return retriever_function

# # Initialize components
# mistral_api_key = os.getenv("mistral_api_key")
# llm = ChatMistralAI(
#     model="mistral-large-latest",
#     mistral_api_key=mistral_api_key,
#     temperature=0.01,
#     streaming=True,
# )

# deepseek_api_key = os.getenv("DEEPSEEK_KEY")
# llm = ChatDeepSeek(
#     model="deepseek-chat",
#     temperature=0,
#     api_key=deepseek_api_key,
#     streaming=True,
# )


gemini_api_key = os.getenv("GEMINI_KEY")
llm = ChatGoogleGenerativeAI(
    model="gemini-1.5-pro",
    temperature=0,
    google_api_key=gemini_api_key,
    disable_streaming=True,
)


rag_loader = OptimizedRAGLoader()
retriever = rag_loader.get_retriever(k=5)  # Reduced k for faster retrieval

# Cache for processed questions
question_cache = {}

prompt_template = ChatPromptTemplate.from_messages([
    ("system", """Vous êtes un assistant juridique expert qualifié. Analysez et répondez aux questions juridiques avec précision.
    
    PROCESSUS D'ANALYSE :
    1. Analysez le contexte fourni : {context}
    2. Utilisez la recherche web si la reponse n'existe pas dans le contexte
    3. Privilégiez les sources officielles et la jurisprudence récente

    Question à traiter : {question}
    """),
    ("human", "{question}")
])



import gradio as gr
from typing import Iterator

# Ajouter du CSS pour personnaliser l'apparence
css = """
/* Reset RTL global */
*, *::before, *::after {
    direction: rtl !important;
    text-align: right !important;
}

body {
    font-family: 'Amiri', sans-serif;  /* Utilisation de la police Arabe andalouse */
    background-color: black;  /* Fond blanc */
    color: black !important;  /* Texte noir */
    direction: rtl !important;  /* Texte en arabe aligné à droite */
}

.gradio-container {
    direction: rtl !important;  /* Alignement RTL pour toute l'interface */
}

/* Éléments de formulaire */
input[type="text"], 
.gradio-textbox input,
textarea {
    border-radius: 20px;
    padding: 10px 15px;
    border: 2px solid #000;
    font-size: 16px;
    width: 80%;
    margin: 0 auto;
    text-align: right !important;
}

/* Surcharge des styles de placeholder */
input::placeholder,
textarea::placeholder {
    text-align: right !important;
    direction: rtl !important;
}

/* Boutons */
.gradio-button {
    border-radius: 20px;
    font-size: 16px;
    background-color: #007BFF;
    color: white;
    padding: 10px 20px;
    margin: 10px auto;
    border: none;
    width: 80%;
    display: block;
}

.gradio-button:hover {
    background-color: #0056b3;
}

.gradio-chatbot .message {
    border-radius: 20px;
    padding: 10px;
    margin: 10px 0;
    background-color: #f1f1f1;
    border: 1px solid #ddd;
    width: 80%;
    text-align: right !important;
    direction: rtl !important;
}

/* Messages utilisateur alignés à gauche */
.gradio-chatbot .user-message {
    margin-right: auto;
    background-color: #e3f2fd;
    text-align: right !important;
    direction: rtl !important;
}

/* Messages assistant alignés à droite */
.gradio-chatbot .assistant-message {
    margin-right: auto;
    background-color: #f1f1f1;
    text-align: right 
}

/* Corrections RTL pour les éléments spécifiques */
.gradio-textbox textarea {
    text-align: right !important;
}

.gradio-dropdown div {
    text-align: right !important;
}
"""

def process_question(question: str) -> Iterator[str]:
    if question in question_cache:
        response, docs = question_cache[question]
        yield response + "\nSources:\n" + "\n".join([doc.page_content for doc in docs])
        return
        
    relevant_docs = retriever(question)
    # context = "\n".join([doc.page_content for doc in relevant_docs])


    context = [doc.page_content for doc in relevant_docs]
    text_pairs = [[question, text] for text in context]
    scores = rag_loader.reranker.predict(text_pairs)
    
    scored_docs = list(zip(scores, context, relevant_docs))
    # scored_docs.sort(reverse=True)
    scored_docs.sort(key=lambda x: x[0], reverse=True)
    reranked_docs = [d[2].page_content for d in scored_docs][:10]

    
    prompt = prompt_template.format_messages(
        context=reranked_docs,
        question=question
    )
    full_response = ""
    try:
        for chunk in llm.stream(prompt):
            if isinstance(chunk, str):
                current_chunk = chunk
            else:
                current_chunk = chunk.content
            full_response += current_chunk
            # sources = "\n".join(set([doc.metadata.get("source") for doc in relevant_docs]))
            # sources = [os.path.splitext(source[1])[0] for source in sources] 
            # yield full_response + "\n\n\nالمصادر المحتملة :\n" + "".join(sources)
            sources = [doc.metadata.get("source") for doc in relevant_docs]
            sources = list(set([os.path.splitext(source)[0] for source in sources]))


            sources = [d[2].metadata['source'] for d in scored_docs][:10]
            sources = list(set([os.path.splitext(source)[0] for source in sources]))

            
            yield full_response + "\n\n\nالمصادر المحتملة :\n" + "\n".join(sources)
            # yield full_response + "\n\n\nالمصادر المحتملة:\n" + "\n".join([doc.metadata.get("source") for doc in relevant_docs])
        question_cache[question] = (full_response, relevant_docs)
    except Exception as e:
        yield f"Erreur lors du traitement : {str(e)}"


def gradio_stream(question: str, chat_history: list) -> Iterator[list]:
    """
    Format the output for Gradio Chatbot component with streaming.
    """
    full_response = ""
    try:
        for partial_response in process_question(question):
            full_response = partial_response
            # Append the latest assistant response to chat history
            updated_chat = chat_history + [[question, partial_response]]
            yield updated_chat
    except Exception as e:
        # Handle errors during streaming
        updated_chat = chat_history + [[question, f"Erreur : {str(e)}"]]
        yield updated_chat


# Gradio interface
with gr.Blocks(css=css) as demo:

    gr.Markdown("<h2 style='text-align: center !important;'>هذا تطبيق للاجابة على الأسئلة المتعلقة بالقوانين المغربية</h2>")

    # Organisation en 3 lignes
    with gr.Row():  # Première ligne: Question
        message = gr.Textbox(label="أدخل سؤالك", placeholder="اكتب سؤالك هنا", elem_id="question_input")
        
    with gr.Row():  # Deuxième ligne: Bouton de recherche
        send = gr.Button("بحث", elem_id="search_button")

    with gr.Row():  # Troisième ligne: Affichage de la réponse
        chatbot = gr.Chatbot(label="")

    # Fonction de mise à jour pour l'utilisateur
    def user_input(user_message, chat_history):
        return "", chat_history + [[user_message, None]]

    send.click(user_input, [message, chatbot], [message, chatbot], queue=False)
    send.click(gradio_stream, [message, chatbot], chatbot)

demo.launch(share=True)