Add Kpa classification model

models/label.py

@@ -1 +1,98 @@
-# 
+"""Pydantic schemas for key-point matching prediction endpoints"""
+
+from pydantic import BaseModel, Field, ConfigDict
+from typing import List, Optional, Dict
+
+
+class PredictionRequest(BaseModel):
+    """Request model for single key-point/argument prediction"""
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "argument": "Climate change is accelerating due to industrial emissions.",
+                "key_point": "Human industry contributes significantly to global warming."
+            }
+        }
+    )
+
+    argument: str = Field(
+        ..., min_length=5, max_length=1000,
+        description="The argument text to evaluate"
+    )
+    key_point: str = Field(
+        ..., min_length=5, max_length=500,
+        description="The key point used for comparison"
+    )
+
+
+class PredictionResponse(BaseModel):
+    """Response model for single prediction"""
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "prediction": 1,
+                "confidence": 0.874,
+                "label": "MATCH",
+                "probabilities": {
+                    "match": 0.874,
+                    "no_match": 0.126
+                }
+            }
+        }
+    )
+
+    prediction: int = Field(..., description="1 = match, 0 = no match")
+    confidence: float = Field(..., ge=0.0, le=1.0,
+                              description="Confidence score of the prediction")
+    label: str = Field(..., description="MATCH or NO_MATCH")
+    probabilities: Dict[str, float] = Field(
+        ..., description="Dictionary of class probabilities"
+    )
+
+
+class BatchPredictionRequest(BaseModel):
+    """Request model for batch predictions"""
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "pairs": [
+                    {
+                        "argument": "Schools should implement AI tools to support learning.",
+                        "key_point": "AI can improve student engagement."
+                    },
+                    {
+                        "argument": "Governments must reduce plastic usage.",
+                        "key_point": "Plastic waste harms the environment."
+                    }
+                ]
+            }
+        }
+    )
+
+    pairs: List[PredictionRequest] = Field(
+        ..., max_length=100,
+        description="List of argument-keypoint pairs (max 100)"
+    )
+
+
+class BatchPredictionResponse(BaseModel):
+    """Response model for batch key-point predictions"""
+    predictions: List[PredictionResponse]
+    total_processed: int = Field(..., description="Number of processed items")
+
+
+class HealthResponse(BaseModel):
+    """Health check model for the API"""
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "status": "ok",
+                "model_loaded": True,
+                "device": "cuda"
+            }
+        }
+    )
+    
+    status: str = Field(..., description="API health status")
+    model_loaded: bool = Field(..., description="Whether the model is loaded")
+    device: str = Field(..., description="Device used for inference (cpu/cuda)")

routes/label.py

@@ -0,0 +1,123 @@
+"""Keypoint–Argument Matching Endpoints"""
+
+from fastapi import APIRouter, HTTPException
+from datetime import datetime
+import logging
+
+from models import (
+    PredictionRequest,
+    PredictionResponse,
+    BatchPredictionRequest,
+    BatchPredictionResponse
+)
+
+from services import label_model_manage  # ton ModelManager clean
+
+router = APIRouter()
+logger = logging.getLogger(__name__)
+
+
+@router.get("/model-info", tags=["KPA"])
+async def get_model_info():
+    """
+    Return information about the loaded KPA model.
+    """
+    try:
+        model_info = label_model_manage.get_model_info()
+
+        return {
+            "model_name": model_info.get("model_name", "unknown"),
+            "device": model_info.get("device", "cpu"),
+            "max_length": model_info.get("max_length", 256),
+            "num_labels": model_info.get("num_labels", 2),
+            "loaded": model_info.get("loaded", False),
+            "timestamp": datetime.now().isoformat()
+        }
+
+    except Exception as e:
+        logger.error(f"Model info error: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Failed to get model info: {str(e)}")
+
+
+@router.post("/predict", response_model=PredictionResponse, tags=["KPA"])
+async def predict_kpa(request: PredictionRequest):
+    """
+    Predict keypoint–argument matching for a single pair.
+
+    - **argument**: The argument text  
+    - **key_point**: The key point to evaluate  
+
+    Returns the predicted class (apparie / non_apparie) with probabilities.
+    """
+    try:
+        result = label_model_manage.predict(
+            argument=request.argument,
+            key_point=request.key_point
+        )
+
+        response = PredictionResponse(
+            prediction=result["prediction"],
+            confidence=result["confidence"],
+            label=result["label"],
+            probabilities=result["probabilities"]
+        )
+
+        logger.info(
+            f"KPA Prediction: {response.label} "
+            f"(conf={response.confidence:.4f})"
+        )
+
+        return response
+
+    except Exception as e:
+        logger.error(f"KPA prediction error: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Prediction failed: {str(e)}")
+
+
+@router.post("/batch-predict", response_model=BatchPredictionResponse, tags=["KPA"])
+async def batch_predict_kpa(request: BatchPredictionRequest):
+    """
+    Predict keypoint–argument matching for multiple argument/keypoint pairs.
+
+    - **pairs**: List of items to classify
+
+    Returns predictions for all pairs.
+    """
+    try:
+        results = []
+
+        for item in request.pairs:
+            try:
+                result = label_model_manage.predict(
+                    argument=item.argument,
+                    key_point=item.key_point
+                )
+
+                response = PredictionResponse(
+                    prediction=result["prediction"],
+                    confidence=result["confidence"],
+                    label=result["label"],
+                    probabilities=result["probabilities"]
+                )
+
+                results.append(response)
+
+            except Exception:
+                results.append(
+                    PredictionResponse(
+                        prediction=-1,
+                        confidence=0.0,
+                        label="error",
+                        probabilities={"error": 1.0}
+                    )
+                )
+
+        logger.info(f"Batch KPA prediction completed — {len(results)} items processed")
+
+        return BatchPredictionResponse(
+            predictions=results
+        )
+
+    except Exception as e:
+        logger.error(f"Batch KPA prediction error: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Batch prediction failed: {str(e)}")

services/label_model_manage.py

@@ -0,0 +1,122 @@
+"""Model manager for keypoint–argument matching model"""
+
+import os
+import torch
+from transformers import AutoTokenizer, AutoModelForSequenceClassification
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class KpaModelManager:
+    """Manages loading and inference for keypoint matching model"""
+
+    def __init__(self):
+        self.model = None
+        self.tokenizer = None
+        self.device = None
+        self.model_loaded = False
+        self.max_length = 256
+
+    def load_model(self, model_path: str = None):
+        """Load fine-tuned model weights and tokenizer"""
+        if self.model_loaded:
+            logger.info("KPA model already loaded")
+            return
+
+        try:
+            # Detect device
+            self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+            logger.info(f"Using device: {self.device}")
+
+            # Resolve model path
+            if model_path is None:
+                base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+                model_path = os.path.join(base_dir, "models", "modele_appariement_rapide.pth")
+
+            logger.info(f"Loading KPA model weights from: {model_path}")
+
+            if not os.path.exists(model_path):
+                raise FileNotFoundError(f"Model file not found: {model_path}")
+
+            # Load tokenizer + architecture
+            model_name = "distilbert-base-uncased"
+            logger.info("Loading tokenizer...")
+            self.tokenizer = AutoTokenizer.from_pretrained(model_name)
+
+            logger.info("Loading base architecture...")
+            self.model = AutoModelForSequenceClassification.from_pretrained(
+                model_name,
+                num_labels=2
+            )
+
+            # Load trained weights
+            checkpoint = torch.load(model_path, map_location=self.device)
+
+            if "model_state_dict" in checkpoint:
+                self.model.load_state_dict(checkpoint["model_state_dict"])
+            else:
+                self.model.load_state_dict(checkpoint)
+
+            self.model.to(self.device)
+            self.model.eval()
+
+            self.model_loaded = True
+            logger.info("✓ KPA model loaded successfully!")
+
+        except Exception as e:
+            logger.error(f"Error loading KPA model: {str(e)}")
+            raise RuntimeError(f"Failed to load KPA model: {str(e)}")
+
+    def predict(self, argument: str, key_point: str) -> dict:
+        """Run a prediction for (argument, key_point)"""
+        if not self.model_loaded:
+            raise RuntimeError("KPA model not loaded")
+
+        try:
+            # Tokenize input
+            encoding = self.tokenizer(
+                argument,
+                key_point,
+                truncation=True,
+                padding="max_length",
+                max_length=self.max_length,
+                return_tensors="pt"
+            ).to(self.device)
+
+            # Forward pass
+            with torch.no_grad():
+                outputs = self.model(**encoding)
+                logits = outputs.logits
+                probabilities = torch.softmax(logits, dim=-1)
+
+                predicted_class = torch.argmax(probabilities, dim=-1).item()
+                confidence = probabilities[0][predicted_class].item()
+
+            return {
+                "prediction": predicted_class,
+                "confidence": confidence,
+                "label": "apparie" if predicted_class == 1 else "non_apparie",
+                "probabilities": {
+                    "non_apparie": probabilities[0][0].item(),
+                    "apparie": probabilities[0][1].item(),
+                },
+            }
+
+        except Exception as e:
+            logger.error(f"Error during prediction: {str(e)}")
+            raise RuntimeError(f"KPA prediction failed: {str(e)}")
+        
+    def get_model_info(self):
+        return {
+            "model_name": self.model_name,
+            "device": str(self.device),
+            "max_length": self.max_length,
+            "num_labels": 2,
+            "loaded": self.model_loaded
+               }
+
+
+
+# Singleton instance
+kpa_model_manager = KpaModelManager()