Initial AMANE API deploy — FastAPI + Docker

Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+ENV HF_HOME="/home/user/.cache/huggingface"
+
+COPY --chown=user requirements.txt .
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+COPY --chown=user . .
+
+EXPOSE 7860
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

app.py

@@ -0,0 +1,58 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel
+
+from predictor import load_benchmark, load_teacher, load_student
+
+_models: dict = {}
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    _models["benchmark"] = load_benchmark()
+    _models["teacher"]   = load_teacher()
+    _models["student"]   = load_student()
+    yield
+    _models.clear()
+
+
+app = FastAPI(title="AMANE Toxicity API", version="1.0.0", lifespan=lifespan)
+
+
+class PredictRequest(BaseModel):
+    text: str
+    model: str = "student"
+
+
+class BatchRequest(BaseModel):
+    texts: list[str]
+    model: str = "student"
+
+
+@app.get("/health")
+def health():
+    return {
+        name: {"loaded": p.loaded, "error": p.error}
+        for name, p in _models.items()
+    }
+
+
+@app.post("/predict")
+def predict(req: PredictRequest):
+    p = _models.get(req.model)
+    if p is None:
+        raise HTTPException(status_code=404, detail=f"Unknown model: {req.model}")
+    if not p.loaded:
+        raise HTTPException(status_code=503, detail=f"Model not available: {p.error}")
+    return p.predict(req.text)
+
+
+@app.post("/predict_batch")
+def predict_batch(req: BatchRequest):
+    p = _models.get(req.model)
+    if p is None:
+        raise HTTPException(status_code=404, detail=f"Unknown model: {req.model}")
+    if not p.loaded:
+        raise HTTPException(status_code=503, detail=f"Model not available: {p.error}")
+    return p.predict_batch(req.texts)

predictor.py

@@ -0,0 +1,427 @@
+import os
+import json
+import joblib
+import numpy as np
+from utils import clean_text, LABEL_COLS
+
+# torch et transformers sont optionnels — uniquement requis pour le Teacher XLM-R
+try:
+    import torch
+    import torch.nn as nn
+    from transformers import AutoTokenizer, AutoModel
+    TORCH_AVAILABLE = True
+except ImportError:
+    TORCH_AVAILABLE = False
+
+
+# ── HuggingFace Hub — fallback si fichiers absents en local ─────────────────
+HF_REPO_ID = 'BHM03/amane-models'
+
+def _resolve(local_path: str, hf_filename: str) -> str:
+    """Retourne le chemin local si le fichier existe, sinon télécharge depuis HF Hub."""
+    if os.path.exists(local_path):
+        return local_path
+    try:
+        from huggingface_hub import hf_hub_download
+        return hf_hub_download(
+            repo_id=HF_REPO_ID,
+            filename=hf_filename,
+            token=os.environ.get('HF_TOKEN'),
+        )
+    except Exception as e:
+        raise FileNotFoundError(
+            f'Fichier introuvable localement ({local_path}) '
+            f'et téléchargement HF Hub échoué : {e}'
+        )
+
+
+# ── Chemins modèles ──────────────────────────────────────────────────────────
+BENCHMARK_MODEL_PATH    = 'models/ovr_balanced.joblib'
+BENCHMARK_TFIDF_PATH    = 'models/tfidf_vectorizer.joblib'
+TEACHER_CKPT_PATH       = 'models/teacher/teacher_best.pt'
+TEACHER_TOKENIZER_PATH  = 'models/teacher/tokenizer'
+TEACHER_THRESH_PATH     = 'models/teacher/thresholds.json'
+TEACHER_STAR_CKPT_PATH  = 'models/teacher_star/teacher_star.pt'
+TEACHER_STAR_THRESH_PATH= 'models/teacher_star/threshold_darija.json'
+STUDENT_CKPT_PATH       = 'models/student/student_best.pt'
+STUDENT_THRESH_PATH     = 'models/student/threshold_student.json'
+
+# ── Hyperparamètres d'inférence Teacher (cohérents avec l'entraînement) ─────
+TEACHER_MAX_LEN      = 256
+TEACHER_STAR_MAX_LEN = 128
+STUDENT_MAX_LEN      = 128
+TEACHER_MODEL_HF     = 'xlm-roberta-base'
+STUDENT_MODEL_HF     = 'distilbert-base-multilingual-cased'
+
+
+# ════════════════════════════════════════════════════════════════════════════
+# BENCHMARK — OneVsRest + LogisticRegression
+# ════════════════════════════════════════════════════════════════════════════
+class BenchmarkPredictor:
+    def __init__(self):
+        self.model    = None
+        self.tfidf    = None
+        self.loaded   = False
+        self.error    = None
+
+    def load(self):
+        try:
+            self.model  = joblib.load(_resolve(BENCHMARK_MODEL_PATH, 'ovr_balanced.joblib'))
+            self.tfidf  = joblib.load(_resolve(BENCHMARK_TFIDF_PATH, 'tfidf_vectorizer.joblib'))
+            self.loaded = True
+        except Exception as e:
+            self.error  = str(e)
+            self.loaded = False
+
+    def predict(self, text: str) -> dict:
+        if not self.loaded:
+            raise RuntimeError(f'Benchmark non chargé : {self.error}')
+        cleaned  = clean_text(text).lower()
+        vec      = self.tfidf.transform([cleaned])
+        probs    = self.model.predict_proba(vec)[0]
+        return {col: float(round(p, 4)) for col, p in zip(LABEL_COLS, probs)}
+
+    def predict_batch(self, texts: list) -> list[dict]:
+        if not self.loaded:
+            raise RuntimeError(f'Benchmark non chargé : {self.error}')
+        cleaned = [clean_text(t).lower() for t in texts]
+        vecs    = self.tfidf.transform(cleaned)
+        probs   = self.model.predict_proba(vecs)
+        return [
+            {col: float(round(p, 4)) for col, p in zip(LABEL_COLS, row)}
+            for row in probs
+        ]
+
+
+# ════════════════════════════════════════════════════════════════════════════
+# XLM-R TEACHER
+# ════════════════════════════════════════════════════════════════════════════
+class TeacherPredictor:
+    def __init__(self):
+        self.model      = None
+        self.tokenizer  = None
+        self.thresholds = None
+        self.loaded     = False
+        self.error      = None
+        self.device     = None
+
+    def load(self):
+        if not TORCH_AVAILABLE:
+            self.error  = (
+                'PyTorch / Transformers non installés. '
+                'Lance : pip install torch transformers sentencepiece'
+            )
+            self.loaded = False
+            return
+
+        try:
+            self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+
+            # 2. Charger le tokenizer — on utilise directement 'xlm-roberta-base'
+            # car le tokenizer ne change pas lors du fine-tuning, et les fichiers
+            # locaux sont incomplets (sentencepiece.bpe.model manquant).
+            self.tokenizer = AutoTokenizer.from_pretrained(TEACHER_MODEL_HF)
+
+            # 3. Reconstruire l'architecture (identique au notebook d'entraînement)
+            class TeacherModel(nn.Module):
+                def __init__(self, model_name: str, num_labels: int = 6, dropout: float = 0.1):
+                    super().__init__()
+                    self.encoder    = AutoModel.from_pretrained(model_name)
+                    hidden_size     = self.encoder.config.hidden_size
+                    self.dropout    = nn.Dropout(dropout)
+                    self.classifier = nn.Linear(hidden_size, num_labels)
+
+                def mean_pool(self, token_embeds, attention_mask):
+                    mask   = attention_mask.unsqueeze(-1).expand(token_embeds.size()).float()
+                    summed = (token_embeds * mask).sum(dim=1)
+                    counts = mask.sum(dim=1).clamp(min=1e-9)
+                    return summed / counts
+
+                def forward(self, input_ids, attention_mask):
+                    outputs = self.encoder(input_ids=input_ids, attention_mask=attention_mask)
+                    pooled  = self.mean_pool(outputs.last_hidden_state, attention_mask)
+                    pooled  = self.dropout(pooled)
+                    return self.classifier(pooled)
+
+            # 4. Charger le checkpoint
+            ckpt  = torch.load(_resolve(TEACHER_CKPT_PATH, 'teacher_best.pt'), map_location=self.device, weights_only=False)
+            model = TeacherModel(TEACHER_MODEL_HF, num_labels=len(LABEL_COLS))
+            model.load_state_dict(ckpt['model_state'])
+            model.to(self.device)
+            model.eval()
+            self.model = model
+
+            # 5. Charger les seuils tunés
+            if os.path.exists(TEACHER_THRESH_PATH):
+                with open(TEACHER_THRESH_PATH) as f:
+                    self.thresholds = json.load(f)
+            else:
+                # Fallback : seuil 0.5 par défaut
+                self.thresholds = {col: 0.5 for col in LABEL_COLS}
+
+            self.loaded = True
+
+        except Exception as e:
+            self.error  = f'{type(e).__name__}: {e}'
+            self.loaded = False
+
+    # ── Inférence : probabilités ───────────────────────────────────────────
+    def predict(self, text: str) -> dict:
+        """Retourne les probabilités sigmoïdes par label."""
+        if not self.loaded:
+            raise RuntimeError(f'Teacher non chargé : {self.error}')
+
+        cleaned  = clean_text(text)
+        encoding = self.tokenizer(
+            cleaned,
+            max_length=TEACHER_MAX_LEN,
+            padding='max_length',
+            truncation=True,
+            return_tensors='pt',
+        )
+        input_ids = encoding['input_ids'].to(self.device)
+        attn_mask = encoding['attention_mask'].to(self.device)
+
+        with torch.no_grad():
+            logits = self.model(input_ids, attn_mask)
+            probs  = torch.sigmoid(logits).cpu().numpy()[0]
+
+        return {col: float(round(p, 4)) for col, p in zip(LABEL_COLS, probs)}
+
+    # ── Inférence : labels binaires (avec seuils tunés) ────────────────────
+    def predict_with_labels(self, text: str) -> dict:
+        """
+        Retourne probabilités + labels binaires selon les seuils tunés.
+        Format : {'probs': {...}, 'labels': {...}, 'thresholds': {...}}
+        """
+        probs  = self.predict(text)
+        labels = {col: int(probs[col] >= self.thresholds[col]) for col in LABEL_COLS}
+        return {
+            'probs'      : probs,
+            'labels'     : labels,
+            'thresholds' : self.thresholds,
+        }
+
+    # ── Inférence : batch ──────────────────────────────────────────────────
+    def predict_batch(self, texts: list) -> list[dict]:
+        """Inférence batch — plus efficace que predict() en boucle."""
+        if not self.loaded:
+            raise RuntimeError(f'Teacher non chargé : {self.error}')
+
+        cleaned = [clean_text(t) for t in texts]
+        encoding = self.tokenizer(
+            cleaned,
+            max_length=TEACHER_MAX_LEN,
+            padding='max_length',
+            truncation=True,
+            return_tensors='pt',
+        )
+        input_ids = encoding['input_ids'].to(self.device)
+        attn_mask = encoding['attention_mask'].to(self.device)
+
+        with torch.no_grad():
+            logits = self.model(input_ids, attn_mask)
+            probs  = torch.sigmoid(logits).cpu().numpy()
+
+        return [
+            {col: float(round(p, 4)) for col, p in zip(LABEL_COLS, row)}
+            for row in probs
+        ]
+
+
+# ════════════════════════════════════════════════════════════════════════════
+# XLM-R TEACHER* — adapté Darija (fine-tuné sur DarLoad, loss masquée)
+# ════════════════════════════════════════════════════════════════════════════
+class TeacherStarPredictor:
+    def __init__(self):
+        self.model          = None
+        self.tokenizer      = None
+        self.threshold      = 0.45   # seuil tunée sur DarLoad
+        self.loaded         = False
+        self.error          = None
+        self.device         = None
+
+    def load(self):
+        if not TORCH_AVAILABLE:
+            self.error  = 'PyTorch / Transformers non installés.'
+            self.loaded = False
+            return
+
+        try:
+            self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+            self.tokenizer = AutoTokenizer.from_pretrained(TEACHER_MODEL_HF)
+
+            class TeacherModel(nn.Module):
+                def __init__(self, model_name, num_labels=6, dropout=0.1):
+                    super().__init__()
+                    self.encoder    = AutoModel.from_pretrained(model_name)
+                    self.dropout    = nn.Dropout(dropout)
+                    self.classifier = nn.Linear(self.encoder.config.hidden_size, num_labels)
+
+                def mean_pool(self, token_embeds, attention_mask):
+                    mask   = attention_mask.unsqueeze(-1).expand(token_embeds.size()).float()
+                    summed = (token_embeds * mask).sum(dim=1)
+                    return summed / mask.sum(dim=1).clamp(min=1e-9)
+
+                def forward(self, input_ids, attention_mask):
+                    out    = self.encoder(input_ids=input_ids, attention_mask=attention_mask)
+                    pooled = self.mean_pool(out.last_hidden_state, attention_mask)
+                    return self.classifier(self.dropout(pooled))
+
+            ckpt  = torch.load(_resolve(TEACHER_STAR_CKPT_PATH, 'teacher_star.pt'), map_location=self.device, weights_only=False)
+            model = TeacherModel(TEACHER_MODEL_HF, num_labels=len(LABEL_COLS))
+            model.load_state_dict(ckpt['model_state'])
+            model.to(self.device)
+            model.eval()
+            self.model = model
+
+            if os.path.exists(TEACHER_STAR_THRESH_PATH):
+                with open(TEACHER_STAR_THRESH_PATH) as f:
+                    self.threshold = json.load(f).get('toxic_darija', 0.45)
+
+            self.loaded = True
+
+        except Exception as e:
+            self.error  = f'{type(e).__name__}: {e}'
+            self.loaded = False
+
+    def predict(self, text: str) -> dict:
+        if not self.loaded:
+            raise RuntimeError(f'Teacher* non chargé : {self.error}')
+        cleaned  = clean_text(text)
+        encoding = self.tokenizer(
+            cleaned, max_length=TEACHER_STAR_MAX_LEN,
+            padding='max_length', truncation=True, return_tensors='pt',
+        )
+        with torch.no_grad():
+            logits = self.model(
+                encoding['input_ids'].to(self.device),
+                encoding['attention_mask'].to(self.device),
+            )
+            probs = torch.sigmoid(logits).cpu().numpy()[0]
+        return {col: float(round(p, 4)) for col, p in zip(LABEL_COLS, probs)}
+
+    def predict_batch(self, texts: list) -> list[dict]:
+        if not self.loaded:
+            raise RuntimeError(f'Teacher* non chargé : {self.error}')
+        cleaned  = [clean_text(t) for t in texts]
+        encoding = self.tokenizer(
+            cleaned, max_length=TEACHER_STAR_MAX_LEN,
+            padding='max_length', truncation=True, return_tensors='pt',
+        )
+        with torch.no_grad():
+            logits = self.model(
+                encoding['input_ids'].to(self.device),
+                encoding['attention_mask'].to(self.device),
+            )
+            probs = torch.sigmoid(logits).cpu().numpy()
+        return [{col: float(round(p, 4)) for col, p in zip(LABEL_COLS, row)} for row in probs]
+
+
+# ════════════════════════════════════════════════════════════════════════════
+# FACTORY — chargement unique via st.cache_resource
+# ════════════════════════════════════════════════════════════════════════════
+def load_benchmark() -> BenchmarkPredictor:
+    p = BenchmarkPredictor()
+    p.load()
+    return p
+
+def load_teacher() -> TeacherPredictor:
+    p = TeacherPredictor()
+    p.load()
+    return p
+
+def load_teacher_star() -> TeacherStarPredictor:
+    p = TeacherStarPredictor()
+    p.load()
+    return p
+
+
+# ════════════════════════════════════════════════════════════════════════════
+# STUDENT — DistilBERT multilingue spécialisé Darija
+# ════════════════════════════════════════════════════════════════════════════
+class StudentPredictor:
+    def __init__(self):
+        self.model     = None
+        self.tokenizer = None
+        self.threshold = 0.6
+        self.loaded    = False
+        self.error     = None
+        self.device    = None
+
+    def load(self):
+        if not TORCH_AVAILABLE:
+            self.error  = 'PyTorch / Transformers non installés.'
+            self.loaded = False
+            return
+
+        try:
+            self.device    = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+            self.tokenizer = AutoTokenizer.from_pretrained(STUDENT_MODEL_HF)
+
+            class StudentModel(nn.Module):
+                def __init__(self, model_name, num_labels=6, dropout=0.1):
+                    super().__init__()
+                    self.encoder    = AutoModel.from_pretrained(model_name)
+                    self.dropout    = nn.Dropout(dropout)
+                    self.classifier = nn.Linear(self.encoder.config.hidden_size, num_labels)
+
+                def forward(self, input_ids, attention_mask):
+                    out    = self.encoder(input_ids=input_ids, attention_mask=attention_mask)
+                    pooled = out.last_hidden_state[:, 0, :]  # token [CLS]
+                    return self.classifier(self.dropout(pooled))
+
+            ckpt  = torch.load(_resolve(STUDENT_CKPT_PATH, 'student_best.pt'), map_location=self.device, weights_only=False)
+            model = StudentModel(STUDENT_MODEL_HF, num_labels=len(LABEL_COLS))
+            model.load_state_dict(ckpt['model_state'])
+            model.to(self.device)
+            model.eval()
+            self.model = model
+
+            if os.path.exists(STUDENT_THRESH_PATH):
+                with open(STUDENT_THRESH_PATH) as f:
+                    self.threshold = json.load(f).get('toxic_darija', 0.35)
+
+            self.loaded = True
+
+        except Exception as e:
+            self.error  = f'{type(e).__name__}: {e}'
+            self.loaded = False
+
+    def predict(self, text: str) -> dict:
+        if not self.loaded:
+            raise RuntimeError(f'Student non chargé : {self.error}')
+        cleaned  = clean_text(text)
+        encoding = self.tokenizer(
+            cleaned, max_length=STUDENT_MAX_LEN,
+            padding='max_length', truncation=True, return_tensors='pt',
+        )
+        with torch.no_grad():
+            logits = self.model(
+                encoding['input_ids'].to(self.device),
+                encoding['attention_mask'].to(self.device),
+            )
+            probs = torch.sigmoid(logits).cpu().numpy()[0]
+        return {col: float(round(p, 4)) for col, p in zip(LABEL_COLS, probs)}
+
+    def predict_batch(self, texts: list) -> list[dict]:
+        if not self.loaded:
+            raise RuntimeError(f'Student non chargé : {self.error}')
+        cleaned  = [clean_text(t) for t in texts]
+        encoding = self.tokenizer(
+            cleaned, max_length=STUDENT_MAX_LEN,
+            padding='max_length', truncation=True, return_tensors='pt',
+        )
+        with torch.no_grad():
+            logits = self.model(
+                encoding['input_ids'].to(self.device),
+                encoding['attention_mask'].to(self.device),
+            )
+            probs = torch.sigmoid(logits).cpu().numpy()
+        return [{col: float(round(p, 4)) for col, p in zip(LABEL_COLS, row)} for row in probs]
+
+
+def load_student() -> StudentPredictor:
+    p = StudentPredictor()
+    p.load()
+    return p

requirements.txt

@@ -0,0 +1,12 @@
+fastapi>=0.110.0
+uvicorn>=0.27.0
+pydantic>=2.0.0
+
+torch>=2.0.0
+transformers>=4.38.0
+sentencepiece>=0.1.99
+huggingface_hub>=0.22.0
+
+scikit-learn==1.6.1
+joblib>=1.3.0
+numpy>=1.24.0

utils.py

@@ -0,0 +1,38 @@
+import re
+
+LABEL_COLS = ['toxic', 'severe_toxic', 'obscene', 'threat', 'insult', 'identity_hate']
+
+LABEL_META = {
+    'toxic'         : {'fr': 'Toxique',         'color': '#D85A30'},
+    'severe_toxic'  : {'fr': 'Très toxique',     'color': '#993C1D'},
+    'obscene'       : {'fr': 'Obscène',          'color': '#EF9F27'},
+    'threat'        : {'fr': 'Menace',           'color': '#A32D2D'},
+    'insult'        : {'fr': 'Insulte',          'color': '#534AB7'},
+    'identity_hate' : {'fr': 'Haine identitaire','color': '#0F6E56'},
+}
+
+def clean_text(text: str) -> str:
+    if not isinstance(text, str):
+        return ''
+    text = re.sub(r'[\r\n\t]+', ' ', text)
+    text = re.sub(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', '[USER]', text)
+    text = re.sub(r'http[s]?://\S+', '[URL]', text)
+    text = re.sub(r'==+\s*.+?\s*==+', '', text)
+    text = re.sub(r'\{\{.*?\}\}', '', text)
+    text = re.sub(r'([!?.])\1{2,}', r'\1\1', text)
+    text = re.sub(r' {2,}', ' ', text)
+    return text.strip()
+
+def toxicity_level(probs: dict) -> tuple[str, str]:
+    """Retourne (niveau, couleur) selon les probabilités détectées."""
+    max_prob = max(probs.values())
+    if probs.get('threat', 0) > 0.5 or probs.get('severe_toxic', 0) > 0.5:
+        return 'Très élevé', '#A32D2D'
+    elif max_prob > 0.7:
+        return 'Élevé', '#D85A30'
+    elif max_prob > 0.4:
+        return 'Modéré', '#EF9F27'
+    elif max_prob > 0.2:
+        return 'Faible', '#854F0B'
+    else:
+        return 'Aucun', '#0F6E56'