| import os |
| from fastapi import FastAPI, HTTPException |
| from pydantic import BaseModel |
| from transformers import AutoTokenizer, AutoModelForSequenceClassification |
| import torch |
|
|
| app = FastAPI(title="Routing Service - Space 2") |
|
|
| os.environ["HF_HOME"] = "/data/huggingface-cache" |
| os.environ["TRANSFORMERS_CACHE"] = "/data/huggingface-cache" |
|
|
| |
| |
| |
| class RoutingRequest(BaseModel): |
| text: str |
|
|
| |
| |
| |
| MODEL_NAME = "MoritzLaurer/DeBERTa-v3-base-mnli-fever-anli" |
| tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME) |
| model = AutoModelForSequenceClassification.from_pretrained(MODEL_NAME) |
|
|
| |
| DEPARTMENTS = ["Account", "Software", "Network", "Security", "Hardware", |
| "Infrastructure", "Licensing", "Communication", "RemoteWork", |
| "Training", "Performance"] |
|
|
| |
| |
| |
| @app.post("/route") |
| async def route_ticket(req: RoutingRequest): |
| text = req.text |
| if not text: |
| raise HTTPException(status_code=400, detail="Text cannot be empty") |
| |
| |
| inputs = tokenizer(text, return_tensors="pt", truncation=True) |
| outputs = model(**inputs) |
| logits = outputs.logits[0] |
|
|
| |
| |
| department_idx = torch.argmax(logits).item() % len(DEPARTMENTS) |
| department = DEPARTMENTS[department_idx] |
|
|
| return {"department": department} |
|
|
| |
| |
| |
| @app.get("/health") |
| async def health(): |
| return {"status": "ok"} |
|
|