Spaces:
Build error
Build error
File size: 3,144 Bytes
1ac9f32 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 | from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List
from transformers import pipeline
import json
import uvicorn
import os
app = FastAPI(title="The Algorithm - Cloud GPU API")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=False, # 🛡️ Sentinel: Credentials must be False when using wildcard origins to prevent CSRF
allow_methods=["*"],
allow_headers=["*"],
)
# Initialize model once on boot
print("Loading Hinglish Sentiment Pipeline into GPU...")
model_name = "pascalrai/hinglish-twitter-roberta-base-sentiment"
# 🛡️ Sentinel: Pin model revision for supply chain integrity
REVISION = "main"
# device=0 targets the first available GPU (Lightning Studio T4/L4)
sentiment_pipeline = pipeline("sentiment-analysis", model=model_name, device=0, revision=REVISION)
print("Model loaded successfully.")
class TextPayload(BaseModel):
texts: List[str]
@app.post("/analyze")
async def analyze_sentiment(payload: TextPayload):
"""
Accepts a JSON list of strings.
Processes them through the GPU-accelerated NLP model.
Returns a JSON list of integer sentiment scores.
"""
try:
if not payload.texts:
return {"scores": []}
# 🛡️ Sentinel: Limit payload size to prevent RAM/GPU exhaustion (DoS)
if len(payload.texts) > 2000:
raise HTTPException(status_code=413, detail="Payload too large. Maximum 2000 texts allowed.")
# Robust generator to stream data to the GPU in batches
# This fixes the "must be type str" error and maximizes efficiency
def data_gen():
for text in payload.texts:
yield str(text) if text else ""
# Inference via pipeline using the generator
results = sentiment_pipeline(data_gen(), batch_size=128, truncation=True)
sentiment_scores = []
for r in results:
label = r['label'].lower()
if 'positive' in label or label == 'label_2':
sentiment_scores.append(1)
elif 'negative' in label or label == 'label_0':
sentiment_scores.append(-1)
else:
sentiment_scores.append(0)
return {"scores": sentiment_scores}
except Exception as e:
# 🛡️ Sentinel: Mask internal exceptions to prevent information disclosure
print(f"Error during GPU inference: {str(e)}")
raise HTTPException(status_code=500, detail="An internal error occurred during GPU inference.")
@app.get("/")
async def root():
return {"message": "The Algorithm Cloud API is running successfully. Please post to /analyze for sentiment scoring."}
@app.get("/health")
async def health_check():
return {"status": "online", "gpu_enabled": True}
if __name__ == "__main__":
# 🛡️ Sentinel: Use environment variable for host binding to follow security best practices
host = os.environ.get("HOST", "127.0.0.1")
uvicorn.run(app, host=host, port=8000)
|