""" app.py — NPS AI Inference API HuggingFace Space (Docker SDK) — expose un endpoint REST pour l'Automation 2 SFMC Endpoints : GET /health → statut + modèles chargés POST /predict → score IA + description CRM Déploiement : 1. Créer un Space HuggingFace (SDK: Docker) 2. Uploader ce dossier (app.py, modeling_nps_score.py, Dockerfile, requirements.txt) 3. Ajouter le secret HF_TOKEN dans les Settings du Space """ import os import re import logging from contextlib import asynccontextmanager from typing import Optional import torch import numpy as np from fastapi import FastAPI, HTTPException, Security from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from pydantic import BaseModel, Field from transformers import AutoTokenizer, AutoModelForSeq2SeqLM from modeling_nps_score import NPSScoreModel, NPSScoreConfig logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) from huggingface_hub import login import os # ─── Auth HuggingFace ───────────────────────────────────────────────────────── hf_token = os.getenv("HF_TOKEN") if hf_token: login(token=hf_token) # ─── Config ────────────────────────────────────────────────────────────────── SCORE_REPO = os.getenv("SCORE_REPO", "nada-05/nps-score-xlm-roberta") DESC_REPO = os.getenv("DESC_REPO", "nada-05/nps-description-generator-mt5") API_SECRET = os.getenv("API_SECRET", "") # Secret à définir dans le Space HF DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu") # ─── Seuils NPS ────────────────────────────────────────────────────────────── def score_to_segment(score: float) -> str: s = round(score) if s <= 6: return "Detractor" if s <= 8: return "Passive" return "Promoter" def estimate_urgency(ai_score: float, clicked_score: float, comments: str) -> str: avg = (ai_score + clicked_score) / 2 txt = comments.lower() urgent_kw = [ "humiliant", "ignoré", "abîmé", "défectueux", "menti", "retour impossible", "damaged", "refused", "40 minutes", "30 minutes", "waiting" ] has_urgent = any(k in txt for k in urgent_kw) if avg <= 3 or (avg <= 5 and has_urgent): return "high" if avg <= 5: return "medium" if avg <= 6 and has_urgent: return "medium" if avg <= 6: return "low" return "none" # ─── Nettoyage texte ───────────────────────────────────────────────────────── def clean_text(text: str) -> str: if not isinstance(text, str) or not text.strip(): return "" text = re.sub(r"Q\d\s*:", "", text) text = text.replace("|", " ") text = re.sub(r"\b(really|hm|hm!)\b", "", text, flags=re.IGNORECASE) return re.sub(r"\s+", " ", text).strip() def build_input_text(comments: str, improvements: str) -> str: c = clean_text(comments) parts = [c] if c else [] if improvements.strip(): parts.append(f"Points signalés : {improvements.strip()}") return " ".join(parts) if parts else "[aucun commentaire]" def build_desc_prompt(comments: str, improvements: str, ai_score: float, segment: str, urgency: str, language: str) -> str: c = clean_text(comments) lang_hint = f"[lang:{language}] " if language else "" prompt = ( f"generate nps description: {lang_hint}" f"score:{ai_score:.0f}/10 segment:{segment} urgency:{urgency} " f"comments: {c}" ) if improvements.strip(): prompt += f" improvements: {improvements.strip()}" return prompt # ─── État global des modèles ────────────────────────────────────────────────── models = {} @asynccontextmanager async def lifespan(app: FastAPI): """Chargement des modèles au démarrage (une seule fois).""" logger.info(f"Chargement des modèles sur {DEVICE}...") logger.info(f" Score model : {SCORE_REPO}") models["score_tokenizer"] = AutoTokenizer.from_pretrained( SCORE_REPO, trust_remote_code=True ) models["score_model"] = NPSScoreModel.from_pretrained( SCORE_REPO, trust_remote_code=True ).to(DEVICE) models["score_model"].eval() logger.info(f" Description model : {DESC_REPO}") models["desc_tokenizer"] = AutoTokenizer.from_pretrained(DESC_REPO) models["desc_model"] = AutoModelForSeq2SeqLM.from_pretrained(DESC_REPO).to(DEVICE) models["desc_model"].eval() logger.info("✓ Modèles prêts") yield models.clear() # ─── App FastAPI ────────────────────────────────────────────────────────────── app = FastAPI( title="NPS AI Inference API", description="Score sentimental NPS + génération de descriptions CRM", version="1.0.0", lifespan=lifespan, ) security = HTTPBearer(auto_error=False) def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)): """Vérifie le Bearer token si API_SECRET est défini.""" if API_SECRET and (not credentials or credentials.credentials != API_SECRET): raise HTTPException(status_code=401, detail="Token invalide") # ─── Schémas ───────────────────────────────────────────────────────────────── class NPSRequest(BaseModel): comments: str = Field(..., description="AllComments de NPS_Responses") improvements: str = Field("", description="Improvements de NPS_Responses") score_nps: float = Field(5.0, ge=0, le=10, description="Score cliqué par le client") language: str = Field("fr", description="'fr' ou 'en'") class NPSResponse(BaseModel): # Score score_nps_clicked: float segment_clicked: str ai_score: float segment_ai: str # CRM description: str # Texte complet pour DE_Tasks.Description (≤ 4000 chars) ai_reason: str # Version courte pour DE_Tasks.AIReason (≤ 48 chars) ai_reason_full: str # Version longue pour NPS_CloseTheLoop_History.AIReason (≤ 498 chars) urgency_ai: str # high / medium / low / none needs_task: bool # Recommandation IA de créer une tâche # Méta model_version: str = "xlm-roberta-v1" # ─── Endpoints ─────────────────────────────────────────────────────────────── @app.get("/health") def health(): return { "status": "ok", "models_loaded": list(models.keys()), "device": str(DEVICE), "score_repo": SCORE_REPO, "desc_repo": DESC_REPO, } @app.post("/predict", response_model=NPSResponse) def predict(req: NPSRequest, _=Security(verify_token)): if not models: raise HTTPException(status_code=503, detail="Modèles non chargés") # ── 1. Score IA ───────────────────────────────────────────── input_text = build_input_text(req.comments, req.improvements) enc = models["score_tokenizer"]( input_text, return_tensors="pt", max_length=256, truncation=True, padding="max_length", ).to(DEVICE) with torch.no_grad(): out = models["score_model"](**enc) ai_score = round(float(out.logits.item()) * 10, 1) ai_score = max(0.0, min(10.0, ai_score)) segment_ai = score_to_segment(ai_score) seg_clicked = score_to_segment(req.score_nps) urgency_ai = estimate_urgency(ai_score, req.score_nps, req.comments) needs_task = segment_ai == "Detractor" # ── 2. Description (seulement si détracteur IA) ───────────── description = "" if needs_task: prompt = build_desc_prompt( req.comments, req.improvements, ai_score, segment_ai, urgency_ai, req.language ) enc_desc = models["desc_tokenizer"]( prompt, return_tensors="pt", max_length=256, truncation=True ).to(DEVICE) with torch.no_grad(): out_desc = models["desc_model"].generate( **enc_desc, max_new_tokens=200, num_beams=4, no_repeat_ngram_size=3, early_stopping=True, ) description = models["desc_tokenizer"].decode(out_desc[0], skip_special_tokens=True) # ── 3. Versions tronquées pour les DE ──────────────────────── ai_reason = description[:48] if description else "" # DE_Tasks.AIReason (50) ai_reason_full = description[:498] if description else "" # History.AIReason (500) return NPSResponse( score_nps_clicked = req.score_nps, segment_clicked = seg_clicked, ai_score = ai_score, segment_ai = segment_ai, description = description[:4000], ai_reason = ai_reason, ai_reason_full = ai_reason_full, urgency_ai = urgency_ai, needs_task = needs_task, ) # ─── Lancement local ───────────────────────────────────────────────────────── if __name__ == "__main__": import uvicorn uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False)