Spaces:
Build error
Build error
File size: 4,200 Bytes
1444863 | 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 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | """
NLP Evaluation API - Production Deployment
==========================================
Multi-task NLP model for interview evaluation
"""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import torch
import torch.nn as nn
from transformers import RobertaTokenizer, RobertaModel
import uvicorn
app = FastAPI(title="NLP Evaluation API", version="1.0.0")
# Model configuration
TASKS = {
"sentiment": {"labels": ["negative", "neutral", "positive"], "num_labels": 3},
"emotion": {"labels": ["nervous", "confident", "stressed", "calm", "motivated"], "num_labels": 5},
"communication": {"labels": ["poor", "fair", "good", "excellent"], "num_labels": 4},
"confidence_level": {"labels": ["low", "medium", "high"], "num_labels": 3},
"stress_level": {"labels": ["low", "medium", "high"], "num_labels": 3}
}
class MultiTaskRoBERTa(nn.Module):
def __init__(self):
super().__init__()
self.roberta = RobertaModel.from_pretrained("roberta-base")
hidden_size = 768
self.classifiers = nn.ModuleDict({
task: nn.Sequential(
nn.Dropout(0.1),
nn.Linear(hidden_size, hidden_size//2),
nn.ReLU(),
nn.Dropout(0.1),
nn.Linear(hidden_size//2, info["num_labels"])
)
for task, info in TASKS.items()
})
def forward(self, input_ids, attention_mask):
outputs = self.roberta(input_ids=input_ids, attention_mask=attention_mask)
pooled = outputs.last_hidden_state[:, 0, :]
return {task: classifier(pooled) for task, classifier in self.classifiers.items()}
# Load model
print("🧠 Loading model...")
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = MultiTaskRoBERTa().to(device)
checkpoint = torch.load("best_model_ultimate.pt", map_location=device)
model.load_state_dict(checkpoint["model_state_dict"])
model.eval()
print(f"✅ Model loaded on {device}")
# Load tokenizer
print("🔤 Loading tokenizer...")
tokenizer = RobertaTokenizer.from_pretrained("roberta-base")
print("✅ Tokenizer loaded")
class AnalyzeRequest(BaseModel):
text: str
class AnalyzeResponse(BaseModel):
sentiment: str
emotion: str
communication: str
confidence_level: str
stress_level: str
scores: dict
@app.get("/")
def root():
return {
"message": "NLP Evaluation API",
"version": "1.0.0",
"status": "running",
"model": "RoBERTa-base Multi-Task (126M params)",
"tasks": list(TASKS.keys())
}
@app.get("/health")
def health():
return {"status": "healthy", "model_loaded": True}
@app.post("/analyze", response_model=AnalyzeResponse)
def analyze(request: AnalyzeRequest):
try:
# Tokenize
encoding = tokenizer(
request.text,
max_length=256,
padding="max_length",
truncation=True,
return_tensors="pt"
)
input_ids = encoding["input_ids"].to(device)
attention_mask = encoding["attention_mask"].to(device)
# Predict
with torch.no_grad():
logits = model(input_ids, attention_mask)
# Get predictions and scores
predictions = {}
scores = {}
for task, task_logits in logits.items():
probs = torch.softmax(task_logits, dim=1)
pred_idx = torch.argmax(probs, dim=1).item()
pred_label = TASKS[task]["labels"][pred_idx]
pred_score = probs[0][pred_idx].item()
predictions[task] = pred_label
scores[task] = round(pred_score, 4)
return AnalyzeResponse(
sentiment=predictions["sentiment"],
emotion=predictions["emotion"],
communication=predictions["communication"],
confidence_level=predictions["confidence_level"],
stress_level=predictions["stress_level"],
scores=scores
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=7860)
|