# ============================================================================= # BACKEND API POUR COACH PÉDAGOGIQUE IA (VERSION ROBUSTE AVEC SURVEILLANCE) # ============================================================================= # Ce script conserve le modèle Llama-3-8B original comme demandé. # Il ajoute une surveillance de la mémoire (RAM) pour diagnostiquer les # problèmes de ressources sur les plateformes comme Hugging Face Spaces. # # ATTENTION : Il est très probable que ce script échoue sur le Free Tier de # Hugging Face en raison d'une consommation de RAM trop élevée par le modèle. # ============================================================================= # --- Imports --- import os import torch import random import json import logging import psutil # Bibliothèque pour surveiller l'utilisation des ressources système from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException from pydantic import BaseModel from llama_cpp import Llama from langchain_community.vectorstores import FAISS from langchain_community.embeddings import HuggingFaceEmbeddings from huggingface_hub import hf_hub_download, login # --- Configuration du Logging --- logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # --- Fonction pour afficher l'utilisation de la mémoire --- def log_memory_usage(stage=""): """Affiche l'utilisation actuelle de la RAM et du disque.""" process = psutil.Process(os.getpid()) mem_info = process.memory_info() ram_used_gb = mem_info.rss / (1024 ** 3) # Convertir en Go # Obtenir l'utilisation du disque pour le répertoire courant disk_usage = psutil.disk_usage('/') disk_total_gb = disk_usage.total / (1024 ** 3) disk_used_gb = disk_usage.used / (1024 ** 3) logger.info(f"--- Utilisation des ressources ({stage}) ---") logger.info(f"RAM utilisée par l'application : {ram_used_gb:.2f} Go") logger.info(f"Disque : {disk_used_gb:.2f}/{disk_total_gb:.2f} Go utilisés") logger.info("-------------------------------------------") # --- Classe Singleton pour charger les modèles une seule fois --- class ModelSingleton: llm = None vectorstore = None embeddings = None exemples_quiz = [] def load_all_models(self): if self.llm is not None: return try: log_memory_usage("Avant chargement des modèles") hf_token = os.environ.get("HF_TOKEN") if hf_token: logger.info("Authentification à Hugging Face via le token du Space.") login(token=hf_token) base_dir = os.path.dirname(os.path.abspath(__file__)) faiss_index_path = os.path.join(base_dir, "faiss_index_wize") embedding_model_path = os.path.join(base_dir, "embedding_model_saved") if not os.path.exists(faiss_index_path) or not os.path.exists(embedding_model_path): raise FileNotFoundError("Les dossiers 'faiss_index_wize' ou 'embedding_model_saved' sont introuvables.") logger.info(f"Chargement du modèle d'embedding depuis : {embedding_model_path}") self.embeddings = HuggingFaceEmbeddings(model_name=embedding_model_path, model_kwargs={'device': 'cpu'}) logger.info("Chargement de l'index FAISS...") self.vectorstore = FAISS.load_local(faiss_index_path, self.embeddings, allow_dangerous_deserialization=True) logger.info("✅ Artefacts locaux (Embeddings et FAISS) chargés.") log_memory_usage("Après chargement FAISS/Embeddings") # --- MODÈLE CONSERVÉ COMME DEMANDÉ --- model_repo_id = "QuantFactory/Meta-Llama-3-8B-Instruct-GGUF" model_filename = "Meta-Llama-3-8B-Instruct.Q4_K_M.gguf" logger.info(f"Téléchargement du LLM '{model_filename}'...") model_path = hf_hub_download(repo_id=model_repo_id, filename=model_filename) log_memory_usage("Après téléchargement du LLM") logger.info("!!! POINT CRITIQUE : Chargement du LLM 8B en mémoire. C'est ici que le manque de RAM peut causer un crash.") self.llm = Llama( model_path=model_path, n_gpu_layers=-1, n_ctx=4096, verbose=False, chat_format="llama-3" ) logger.info("✅ Modèle LLM chargé avec succès.") log_memory_usage("Après chargement du LLM") exemples_path = os.path.join(base_dir, "exemples_quiz.json") if os.path.exists(exemples_path): with open(exemples_path, 'r', encoding='utf-8') as f: self.exemples_quiz = json.load(f) logger.info(f"✅ {len(self.exemples_quiz)} exemples de quiz chargés.") except Exception as e: logger.error(f"ERREUR CRITIQUE PENDANT LE CHARGEMENT DES MODELES: {e}", exc_info=True) log_memory_usage("Au moment de l'erreur") raise RuntimeError(f"Impossible de charger les modèles: {e}") # --- Instance du Singleton --- models = ModelSingleton() # --- Gestionnaire de "Lifespan" (la bonne méthode pour FastAPI) --- @asynccontextmanager async def lifespan(app: FastAPI): logger.info("Démarrage de l'application... Tentative de chargement des modèles.") models.load_all_models() logger.info("✅ Démarrage réussi : les modèles sont prêts.") yield logger.info("Arrêt de l'application...") # --- Initialisation de l'API --- app = FastAPI(lifespan=lifespan) # --- Modèles Pydantic --- class QuestionRequest(BaseModel): question: str class AnswerResponse(BaseModel): answer: str # --- Points de Terminaison --- @app.get("/") def read_root(): return {"status": "En ligne et modèles prêts."} @app.post("/ask", response_model=AnswerResponse) def ask_wize_rag(request: QuestionRequest): if models.llm is None: raise HTTPException(status_code=503, detail="Service non disponible : les modèles ne sont pas chargés.") user_question = request.question logger.info(f"Requête reçue: '{user_question}'") try: retriever = models.vectorstore.as_retriever(search_type="similarity", search_kwargs={"k": 5}) docs = retriever.invoke(user_question) context = "\n".join([doc.page_content for doc in docs]) is_quiz_request = any(keyword in user_question.lower() for keyword in ["quiz", "interroge-moi", "questionnaire"]) if is_quiz_request: system_message = "Tu es un générateur de quiz expert et précis, utilisant uniquement le Contexte fourni..." else: system_message = "Tu es un coach pédagogique expert. Tu réponds uniquement et exclusivement à partir des informations extraites du Contexte fourni..." prompt = f"""<|begin_of_text|><|start_header_id|>system<|end_header_id|>\n{system_message}<|eot_id|><|start_header_id|>user<|end_header_id|>\nContexte :\n{context}\n\nQuestion : {user_question}<|eot_id|><|start_header_id|>assistant<|end_header_id|>""" logger.info("Génération de la réponse...") response = models.llm(prompt, max_tokens=2048, temperature=0.3, stop=["<|eot_id|>"], echo=False) answer = response['choices'][0]['text'].strip() return AnswerResponse(answer=answer) except Exception as e: logger.error(f"Erreur lors de la génération de la réponse : {e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Erreur interne du serveur: {e}")