first test

.dockerignore

@@ -0,0 +1,23 @@
+__pycache__
+*.pyc
+*.pyo
+*.pyd
+.Python
+*.so
+*.egg
+*.egg-info
+dist
+build
+.git
+.gitignore
+.env
+.venv
+venv/
+ENV/
+env/
+*.log
+.DS_Store
+README.md
+models/
+*.md
+

Dockerfile

@@ -0,0 +1,32 @@
+# Use Python 3.10 slim image for smaller size
+FROM python:3.10-slim
+
+# Set working directory
+WORKDIR /app
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    build-essential \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements first for better caching
+COPY requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY . .
+
+# Expose port (Hugging Face Spaces uses port 7860 by default, but we'll use PORT env var)
+EXPOSE 7860
+
+# Set environment variables
+ENV HOST=0.0.0.0
+ENV PORT=7860
+ENV RELOAD=False
+
+# Run the application
+# Hugging Face Spaces sets PORT environment variable automatically
+CMD uvicorn main:app --host 0.0.0.0 --port ${PORT:-7860}
+

config.py

@@ -0,0 +1,36 @@
+"""Configuration settings for the API"""
+
+import os
+from pathlib import Path
+from dotenv import load_dotenv
+
+# Load environment variables from .env file
+load_dotenv()
+
+# Get project root directory
+API_DIR = Path(__file__).parent
+PROJECT_ROOT = API_DIR.parent
+
+# Hugging Face configuration
+HUGGINGFACE_API_KEY = os.getenv("HUGGINGFACE_API_KEY", "")
+HUGGINGFACE_MODEL_ID = os.getenv("HUGGINGFACE_MODEL_ID", "yassine-mhirsi/debertav3-stance-detection")
+
+# Stance detection model configuration
+# Use Hugging Face model ID instead of local path
+STANCE_MODEL_ID = HUGGINGFACE_MODEL_ID
+
+# API configuration
+API_TITLE = "NLP Project API"
+API_DESCRIPTION = "API for various NLP models including stance detection and more"
+API_VERSION = "1.0.0"
+
+# Server configuration
+HOST = os.getenv("HOST", "0.0.0.0")  # Use 0.0.0.0 for Docker/Spaces
+PORT = int(os.getenv("PORT", "7860"))  # Default 7860 for Hugging Face Spaces
+RELOAD = os.getenv("RELOAD", "False").lower() == "true"  # Set to False in production
+
+# CORS configuration
+CORS_ORIGINS = ["*"]  # In production, specify exact origins
+CORS_CREDENTIALS = True
+CORS_METHODS = ["*"]
+CORS_HEADERS = ["*"]

main.py

@@ -0,0 +1,88 @@
+"""Main FastAPI application entry point"""
+
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+import uvicorn
+import logging
+
+from config import (
+    API_TITLE,
+    API_DESCRIPTION,
+    API_VERSION,
+    STANCE_MODEL_ID,
+    HUGGINGFACE_API_KEY,
+    HOST,
+    PORT,
+    RELOAD,
+    CORS_ORIGINS,
+    CORS_CREDENTIALS,
+    CORS_METHODS,
+    CORS_HEADERS,
+)
+from services import stance_model_manager
+from routes import api_router
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Load models on startup and cleanup on shutdown"""
+    # Startup: Load all models
+    logger.info("Loading models on startup...")
+    
+    # Load stance detection model
+    try:
+        logger.info(f"Loading stance model from Hugging Face: {STANCE_MODEL_ID}")
+        stance_model_manager.load_model(STANCE_MODEL_ID, HUGGINGFACE_API_KEY)
+    except Exception as e:
+        logger.error(f"✗ Failed to load stance model: {str(e)}")
+        logger.error("⚠️  Stance detection endpoints will not work!")
+    
+    logger.info("✓ API startup complete")
+    
+    yield  # Application runs here
+    
+    # Shutdown: Cleanup (if needed)
+    # Currently no cleanup needed, but you can add it here if necessary
+
+
+# Create FastAPI application
+app = FastAPI(
+    title=API_TITLE,
+    description=API_DESCRIPTION,
+    version=API_VERSION,
+    docs_url="/docs",
+    redoc_url="/redoc",
+    lifespan=lifespan,
+)
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=CORS_ORIGINS,
+    allow_credentials=CORS_CREDENTIALS,
+    allow_methods=CORS_METHODS,
+    allow_headers=CORS_HEADERS,
+)
+
+# Include API routes
+app.include_router(api_router)
+
+
+if __name__ == "__main__":
+    # Run the API server
+    # Access at: http://localhost:8000
+    # API docs at: http://localhost:8000/docs
+    
+    # Run the API server
+    uvicorn.run(
+        "main:app",
+        host=HOST,
+        port=PORT,
+        reload=RELOAD,
+        log_level="info"
+    )

models/__init__.py

@@ -0,0 +1,24 @@
+"""Pydantic models and schemas for request/response validation"""
+
+# Import stance-related schemas
+from .stance import (
+    StanceRequest,
+    StanceResponse,
+    BatchStanceRequest,
+    BatchStanceResponse,
+)
+
+# Import health-related schemas
+from .health import (
+    HealthResponse,
+)
+
+__all__ = [
+    # Stance schemas
+    "StanceRequest",
+    "StanceResponse",
+    "BatchStanceRequest",
+    "BatchStanceResponse",
+    # Health schemas
+    "HealthResponse",
+]

models/health.py

@@ -0,0 +1,12 @@
+"""Pydantic schemas for health check endpoints"""
+
+from pydantic import BaseModel
+
+
+class HealthResponse(BaseModel):
+    """Health check response"""
+    status: str
+    model_loaded: bool
+    device: str
+    timestamp: str
+

models/stance.py

@@ -0,0 +1,59 @@
+"""Pydantic schemas for stance detection endpoints"""
+
+from pydantic import BaseModel, Field, ConfigDict
+from typing import List
+
+
+class StanceRequest(BaseModel):
+    """Request model for stance prediction"""
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "topic": "Assisted suicide should be a criminal offence",
+                "argument": "People have the right to choose how they end their lives"
+            }
+        }
+    )
+    
+    topic: str = Field(..., min_length=5, max_length=500, 
+                       description="The debate topic or statement")
+    argument: str = Field(..., min_length=5, max_length=1000,
+                         description="The argument text to classify")
+
+
+class StanceResponse(BaseModel):
+    """Response model for stance prediction"""
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "topic": "Assisted suicide should be a criminal offence",
+                "argument": "People have the right to choose how they end their lives",
+                "predicted_stance": "CON",
+                "confidence": 0.9234,
+                "probability_con": 0.9234,
+                "probability_pro": 0.0766,
+                "timestamp": "2024-11-15T10:30:00"
+            }
+        }
+    )
+    
+    topic: str
+    argument: str
+    predicted_stance: str = Field(..., description="PRO or CON")
+    confidence: float = Field(..., ge=0.0, le=1.0)
+    probability_con: float
+    probability_pro: float
+    timestamp: str
+
+
+class BatchStanceRequest(BaseModel):
+    """Request model for batch predictions"""
+    items: List[StanceRequest] = Field(..., max_length=50,
+                                      description="List of topic-argument pairs (max 50)")
+
+
+class BatchStanceResponse(BaseModel):
+    """Response model for batch predictions"""
+    results: List[StanceResponse]
+    total_processed: int
+

requirements.txt

@@ -0,0 +1,8 @@
+fastapi==0.104.1
+uvicorn[standard]==0.24.0
+pydantic==2.5.0
+python-dotenv==1.0.0
+torch>=2.0.0
+transformers>=4.35.0
+accelerate>=0.24.0
+

routes/__init__.py

@@ -0,0 +1,15 @@
+"""API route handlers"""
+
+from fastapi import APIRouter
+from . import root, health, stance
+
+# Create main router
+api_router = APIRouter()
+
+# Include all route modules
+api_router.include_router(root.router)
+api_router.include_router(health.router)
+api_router.include_router(stance.router)
+
+__all__ = ["api_router"]
+

routes/health.py

@@ -0,0 +1,20 @@
+"""Health check endpoint"""
+
+from fastapi import APIRouter
+from datetime import datetime
+from models import HealthResponse
+from services import stance_model_manager
+
+router = APIRouter()
+
+
+@router.get("/health", response_model=HealthResponse, tags=["General"])
+async def health_check():
+    """Health check endpoint"""
+    return HealthResponse(
+        status="healthy" if stance_model_manager.model_loaded else "unhealthy",
+        model_loaded=stance_model_manager.model_loaded,
+        device=str(stance_model_manager.device) if stance_model_manager.device else "unknown",
+        timestamp=datetime.now().isoformat()
+    )
+

routes/root.py

@@ -0,0 +1,25 @@
+"""Root endpoint for API information"""
+
+from fastapi import APIRouter
+
+router = APIRouter()
+
+
+@router.get("/", response_model=dict, tags=["General"])
+async def root():
+    """Root endpoint with API information"""
+    return {
+        "message": "NLP Project API",
+        "version": "1.0.0",
+        "features": {
+            "stance_detection": {
+                "predict": "/predict",
+                "batch_predict": "/batch-predict"
+            }
+        },
+        "endpoints": {
+            "health": "/health",
+            "docs": "/docs"
+        }
+    }
+

routes/stance.py

@@ -0,0 +1,89 @@
+"""Stance detection endpoints"""
+
+from fastapi import APIRouter, HTTPException
+from datetime import datetime
+import logging
+
+from models import (
+    StanceRequest,
+    StanceResponse,
+    BatchStanceRequest,
+    BatchStanceResponse,
+)
+from services import stance_model_manager
+
+router = APIRouter()
+logger = logging.getLogger(__name__)
+
+
+@router.post("/predict", response_model=StanceResponse, tags=["Stance Detection"])
+async def predict_stance(request: StanceRequest):
+    """
+    Predict stance for a single topic-argument pair
+    
+    - **topic**: The debate topic or statement (5-500 chars)
+    - **argument**: The argument to classify (5-1000 chars)
+    
+    Returns predicted stance (PRO/CON) with confidence scores
+    """
+    try:
+        # Make prediction
+        result = stance_model_manager.predict(request.topic, request.argument)
+        
+        # Build response
+        response = StanceResponse(
+            topic=request.topic,
+            argument=request.argument,
+            predicted_stance=result["predicted_stance"],
+            confidence=result["confidence"],
+            probability_con=result["probability_con"],
+            probability_pro=result["probability_pro"],
+            timestamp=datetime.now().isoformat()
+        )
+        
+        logger.info(f"Prediction: {result['predicted_stance']} ({result['confidence']:.4f})")
+        return response
+        
+    except Exception as e:
+        logger.error(f"Prediction error: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Prediction failed: {str(e)}")
+
+
+@router.post("/batch-predict", response_model=BatchStanceResponse, tags=["Stance Detection"])
+async def batch_predict_stance(request: BatchStanceRequest):
+    """
+    Predict stance for multiple topic-argument pairs
+    
+    - **items**: List of topic-argument pairs (max 50)
+    
+    Returns predictions for all items
+    """
+    try:
+        results = []
+        
+        # Process each item
+        for item in request.items:
+            result = stance_model_manager.predict(item.topic, item.argument)
+            
+            response = StanceResponse(
+                topic=item.topic,
+                argument=item.argument,
+                predicted_stance=result["predicted_stance"],
+                confidence=result["confidence"],
+                probability_con=result["probability_con"],
+                probability_pro=result["probability_pro"],
+                timestamp=datetime.now().isoformat()
+            )
+            results.append(response)
+        
+        logger.info(f"Batch prediction completed: {len(results)} items")
+        
+        return BatchStanceResponse(
+            results=results,
+            total_processed=len(results)
+        )
+        
+    except Exception as e:
+        logger.error(f"Batch prediction error: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Batch prediction failed: {str(e)}")
+

services/__init__.py

@@ -0,0 +1,8 @@
+"""Services for business logic and external integrations"""
+
+from .stance_model_manager import StanceModelManager, stance_model_manager
+
+__all__ = [
+    "StanceModelManager",
+    "stance_model_manager",
+]

services/stance_model_manager.py

@@ -0,0 +1,101 @@
+"""Model manager for stance detection model"""
+
+import os
+import torch
+from transformers import AutoTokenizer, AutoModelForSequenceClassification
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class StanceModelManager:
+    """Manages stance detection model loading and predictions"""
+    
+    def __init__(self):
+        self.model = None
+        self.tokenizer = None
+        self.device = None
+        self.model_loaded = False
+    
+    def load_model(self, model_id: str, api_key: str = None):
+        """Load model and tokenizer from Hugging Face"""
+        if self.model_loaded:
+            logger.info("Stance model already loaded")
+            return
+        
+        try:
+            logger.info(f"Loading stance model from Hugging Face: {model_id}")
+            
+            # Determine device
+            self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+            logger.info(f"Using device: {self.device}")
+            
+            # Prepare token for authentication if API key is provided
+            token = api_key if api_key else None
+            
+            # Load tokenizer and model from Hugging Face
+            logger.info("Loading tokenizer...")
+            self.tokenizer = AutoTokenizer.from_pretrained(
+                model_id,
+                token=token,
+                trust_remote_code=True
+            )
+            
+            logger.info("Loading model...")
+            self.model = AutoModelForSequenceClassification.from_pretrained(
+                model_id,
+                token=token,
+                trust_remote_code=True
+            )
+            self.model.to(self.device)
+            self.model.eval()
+            
+            self.model_loaded = True
+            logger.info("✓ Stance model loaded successfully from Hugging Face!")
+            
+        except Exception as e:
+            logger.error(f"Error loading stance model: {str(e)}")
+            raise RuntimeError(f"Failed to load stance model: {str(e)}")
+    
+    def predict(self, topic: str, argument: str) -> dict:
+        """Make a single stance prediction"""
+        if not self.model_loaded:
+            raise RuntimeError("Stance model not loaded")
+        
+        # Format input
+        text = f"Topic: {topic} [SEP] Argument: {argument}"
+        
+        # Tokenize
+        inputs = self.tokenizer(
+            text,
+            return_tensors="pt",
+            truncation=True,
+            max_length=512,
+            padding=True
+        ).to(self.device)
+        
+        # Predict
+        with torch.no_grad():
+            outputs = self.model(**inputs)
+            probabilities = torch.nn.functional.softmax(outputs.logits, dim=-1)
+            predicted_class = torch.argmax(probabilities, dim=-1).item()
+        
+        # Extract probabilities
+        prob_con = probabilities[0][0].item()
+        prob_pro = probabilities[0][1].item()
+        
+        # Determine stance
+        stance = "PRO" if predicted_class == 1 else "CON"
+        confidence = probabilities[0][predicted_class].item()
+        
+        return {
+            "predicted_stance": stance,
+            "confidence": confidence,
+            "probability_con": prob_con,
+            "probability_pro": prob_pro
+        }
+
+
+# Initialize singleton instance
+stance_model_manager = StanceModelManager()
+