import gradio as gr import pandas as pd from sentence_transformers import SentenceTransformer import faiss import numpy as np from datasets import load_dataset from huggingface_hub import InferenceClient import os import re from datetime import datetime from markdown_it import MarkdownIt from pypdf import PdfReader # --- CONFIGURATION --- MODEL_ID = "BAAI/bge-m3" DATASET_REPO = "opt-nc/metiers" LLM_MODEL = "Qwen/Qwen2.5-7B-Instruct" APP_URL = "https://huggingface.co/spaces/opt-nc/metiers" # Initialisation du convertisseur Markdown md = MarkdownIt() # Vérification du Token (Logs) HF_TOKEN = os.getenv("HF_TOKEN") if HF_TOKEN: print(f"✅ HF_TOKEN détecté (longueur : {len(HF_TOKEN)})") else: print("⚠️ Aucun HF_TOKEN détecté. L'usage sera limité ou échouera.") # Initialisation du client LLM client = InferenceClient(model=LLM_MODEL, token=HF_TOKEN) print(f"🚀 Chargement des données depuis le Hub HF ({DATASET_REPO})...") try: ds = load_dataset(DATASET_REPO, split="train") df = ds.to_pandas() print(f"✅ {len(df)} fiches chargées avec succès.") except Exception as e: print(f"❌ Erreur lors du chargement du dataset : {e}") df = pd.DataFrame(columns=["id", "title", "text", "famille"]) print(f"🧠 Chargement du modèle d'embedding {MODEL_ID}...") model = SentenceTransformer(MODEL_ID) # Indexation FAISS — IndexFlatIP + normalisation L2 => score cosinus borné [0,1] print("⚡ Indexation des fiches métiers...") if not df.empty: embeddings = model.encode(df['text'].tolist(), show_progress_bar=True, normalize_embeddings=True) embeddings = np.array(embeddings).astype('float32') dimension = embeddings.shape[1] index = faiss.IndexFlatIP(dimension) index.add(embeddings) else: index = None # --------------------------------------------------------------------------- # UTILITAIRES # --------------------------------------------------------------------------- def strip_markdown(text): """Supprime les balises Markdown pour l'affichage en texte brut (PDF).""" text = re.sub(r'\*\*(.*?)\*\*', r'\1', text) text = re.sub(r'\*(.*?)\*', r'\1', text) text = re.sub(r'#{1,6}\s', '', text) text = re.sub(r'`{1,3}', '', text) text = re.sub(r'\n{3,}', '\n\n', text) return text.strip() def extract_text_from_pdf(file_path): try: reader = PdfReader(file_path) text = "" for page in reader.pages: text += page.extract_text() + "\n" return text.strip() except Exception as e: return f"Erreur lors de l'extraction du PDF : {e}" def expand_query(raw_query): """Reformule la query en mots-clés métier via le LLM (query expansion). Retourne (query enrichie, mots-clés) — fallback sur query originale si erreur.""" try: messages = [ { "role": "system", "content": ( "Tu es un expert RH en référentiels métiers (GPEC). " "Transforme la description d'un candidat en mots-clés professionnels " "utilisés dans les fiches de poste et référentiels de compétences. " "Réponds UNIQUEMENT avec les mots-clés séparés par des virgules. " "15 mots maximum. Pas de phrases, pas d'explication." ) }, {"role": "user", "content": raw_query[:500]} ] completion = client.chat_completion(messages, max_tokens=60) keywords = completion.choices[0].message.content.strip() print(f"🔍 Query expansion : {keywords}") # Stratégie B : query originale + mots-clés enrichis return f"{raw_query} {keywords}", keywords except Exception as e: print(f"⚠️ Query expansion échouée, query originale utilisée : {e}") return raw_query, "" # --------------------------------------------------------------------------- # RAG # --------------------------------------------------------------------------- def get_rag_context(query, top_k, score_min): if df.empty or index is None: return [], "" search_k = min(len(df), 50) query_vector = model.encode([query], normalize_embeddings=True).astype('float32') distances, indices = index.search(query_vector, search_k) unique_results = [] seen_metiers = set() for i in range(search_k): idx = indices[0][i] score = distances[0][i] metier = df.iloc[idx] # Filtre score minimum if score < score_min: continue base_id = metier['id'].split('_')[0] if base_id not in seen_metiers: seen_metiers.add(base_id) unique_results.append({"metier": metier, "score": score}) if len(unique_results) >= int(top_k): break context_text = "" for i, res in enumerate(unique_results): m = res["metier"] context_text += f"--- MÉTIER {i+1}: {m['title']} ---\n{m['text']}\n\n" return unique_results, context_text # --------------------------------------------------------------------------- # HTML # --------------------------------------------------------------------------- def build_results_html(results): """Génère le HTML des fiches métiers.""" results_html = """ """ for res in results: metier = res["metier"] score = res["score"] similarity = score * 100 html_content = md.render(str(metier['text'])) results_html += f"""

{metier['title']}

Match: {similarity:.1f}%

ID: {str(metier['id']).split('_')[0]} | Famille: {metier['famille']}

📋 Voir la fiche de compétences complète
{html_content}
""" return results_html # --------------------------------------------------------------------------- # RAPPORT HTML (remplace PDF — imprimable via Ctrl+P) # --------------------------------------------------------------------------- def generate_html_report(combined_query, keywords, results, llm_context, synthesis): """ Appelle le LLM avec un prompt dédié 'rapport RH' puis génère un HTML formaté, imprimable via Ctrl+P ou Enregistrer en PDF depuis le navigateur. """ now = datetime.now().strftime("%d/%m/%Y à %Hh%M") llm_query = combined_query[:2000] # Appel LLM dédié rapport RH (synchrone, sans streaming) print("📝 Génération de la note de synthèse RH...") try: rh_messages = [ { "role": "system", "content": ( "Tu es un consultant RH senior à l'OPT-NC. " "Tu rédiges des notes de synthèse officielles destinées " "à être archivées dans le dossier d'un candidat. " "Ton style est professionnel, structuré et factuel. " "Réponds en HTML avec des balises

,

,