fastapi_app/app.py

import torch
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, Form
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
from transformers import AutoModelForSequenceClassification, AutoTokenizer

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

model = None
tokenizer = None
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Load model on startup and cleanup on shutdown"""
    global model, tokenizer

    try:
        model_id = "codeby-hp/FinetuneTinybert-SentimentClassification"
        
        logger.info(f"Loading tokenizer from {model_id}...")
        tokenizer = AutoTokenizer.from_pretrained(model_id)

        logger.info(f"Loading model from {model_id}...")
        model = AutoModelForSequenceClassification.from_pretrained(model_id)
        model.to(device)
        model.eval()

        logger.info(f"Model loaded successfully on {device}")
    except Exception as e:
        logger.error(f"Error loading model: {e}")
        raise

    yield

    logger.info("Shutting down...")


app = FastAPI(title="Sentiment Analysis API", lifespan=lifespan)

templates = Jinja2Templates(directory="templates")


@app.get("/", response_class=HTMLResponse)
async def home(request: Request):
    """Render the home page"""
    return templates.TemplateResponse("index.html", {"request": request})


@app.post("/predict")
async def predict(request: Request, text: str = Form(...)):
    """Predict sentiment for the given text"""
    if not text.strip():
        return templates.TemplateResponse(
            "index.html",
            {"request": request, "error": "Please enter some text to analyze"},
        )

    try:
        inputs = tokenizer(
            text, return_tensors="pt", truncation=True, max_length=512, padding=True
        )
        inputs = {k: v.to(device) for k, v in inputs.items()}

        with torch.no_grad():
            outputs = model(**inputs)
            logits = outputs.logits
            probabilities = torch.nn.functional.softmax(logits, dim=-1)
            predicted_class = torch.argmax(probabilities, dim=-1).item()
            confidence = probabilities[0][predicted_class].item()

        sentiment_map = {0: "Negative", 1: "Positive"}
        sentiment = sentiment_map.get(predicted_class, "Unknown")

        return templates.TemplateResponse(
            "index.html",
            {
                "request": request,
                "text": text,
                "sentiment": sentiment,
                "confidence": round(confidence * 100, 2),
            },
        )

    except Exception as e:
        logger.error(f"Prediction error: {e}")
        return templates.TemplateResponse(
            "index.html", {"request": request, "error": f"An error occurred: {str(e)}"}
        )


@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {
        "status": "healthy",
        "model_loaded": model is not None,
        "device": str(device),
    }


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=7860)