Initial deployment

Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.12-slim
+
+WORKDIR /code
+
+COPY requirements.txt .
+
+RUN pip install --no-cache-dir --timeout 300 -r requirements.txt
+
+COPY app.py .
+
+EXPOSE 7860
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

app.py

@@ -0,0 +1,213 @@
+import os
+import sys
+import warnings
+warnings.filterwarnings("ignore")
+os.environ["TF_ENABLE_ONEDNN_OPTS"] = "0"
+
+from pathlib import Path
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+from contextlib import asynccontextmanager
+from typing import Optional
+import torch
+import pickle
+import re
+import redis
+from huggingface_hub import hf_hub_download
+from transformers import AutoTokenizer, AutoModelForSequenceClassification
+
+# ── Constants ────────────────────────────────────────────
+CONFIDENCE_THRESHOLD = 10.0
+OOS_INTENT           = "oos"
+OOS_RESPONSE         = "I'm not sure I understand that. Could you rephrase or ask me something banking related?"
+MAX_LEN              = 64
+DEVICE               = torch.device("cpu")
+MODEL_REPO           = "RexTheSlasher/intent-classifier-bert"
+
+# ── Redis Cloud ──────────────────────────────────────────
+r = redis.Redis(
+    host=os.getenv('REDIS_HOST'),
+    port=int(os.getenv('REDIS_PORT', 6379)),
+    password=os.getenv('REDIS_PASSWORD'),
+    decode_responses=True
+)
+
+# ── Load model from Hugging Face Hub ────────────────────
+print("Loading model from Hugging Face...")
+tokenizer     = AutoTokenizer.from_pretrained(MODEL_REPO)
+model         = AutoModelForSequenceClassification.from_pretrained(MODEL_REPO)
+model         = model.to(DEVICE)
+model.eval()
+
+# ── Load label encoder from Hugging Face Hub ────────────
+encoder_path  = hf_hub_download(repo_id=MODEL_REPO, filename="label_encoder.pkl")
+with open(encoder_path, "rb") as f:
+    label_encoder = pickle.load(f)
+
+print(f"Model loaded! Classes: {len(label_encoder.classes_)}")
+
+# ── Text cleaning ────────────────────────────────────────
+def clean_text(text: str) -> str:
+    text = text.lower()
+    text = re.sub(r"[^a-z0-9\s]", "", text)
+    text = re.sub(r"\s+", " ", text).strip()
+    return text
+
+# ── Predict ──────────────────────────────────────────────
+def predict(text: str) -> dict:
+    cleaned  = clean_text(text)
+    encoding = tokenizer(
+        cleaned,
+        max_length=MAX_LEN,
+        padding="max_length",
+        truncation=True,
+        return_tensors="pt"
+    )
+    input_ids      = encoding["input_ids"].to(DEVICE)
+    attention_mask = encoding["attention_mask"].to(DEVICE)
+
+    with torch.no_grad():
+        outputs = model(input_ids=input_ids, attention_mask=attention_mask)
+
+    probabilities               = torch.softmax(outputs.logits, dim=1)
+    confidence, predicted_class = torch.max(probabilities, dim=1)
+    predicted_label             = label_encoder.inverse_transform([predicted_class.item()])[0]
+    confidence_score            = round(confidence.item() * 100, 2)
+
+    top3_probs, top3_indices = torch.topk(probabilities, 3, dim=1)
+    top3 = [
+        {
+            "intent":     label_encoder.inverse_transform([idx.item()])[0],
+            "confidence": round(prob.item() * 100, 2)
+        }
+        for prob, idx in zip(top3_probs[0], top3_indices[0])
+    ]
+
+    is_low_confidence = confidence_score < CONFIDENCE_THRESHOLD
+    if predicted_label == OOS_INTENT or is_low_confidence:
+        predicted_label = "out_of_scope"
+        is_oos = True
+    else:
+        is_oos = False
+
+    return {
+        "input_text":        text,
+        "cleaned_text":      cleaned,
+        "intent":            predicted_label,
+        "confidence":        confidence_score,
+        "top3":              top3,
+        "is_low_confidence": is_low_confidence,
+        "is_oos":            is_oos,
+        "response":          OOS_RESPONSE if is_oos else f"I detected your intent as {predicted_label}"
+    }
+
+# ── FastAPI App ──────────────────────────────────────────
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    print("✅ API started!")
+    yield
+
+app = FastAPI(
+    title="Intent Classification Chatbot API",
+    version="2.0.0",
+    lifespan=lifespan
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"]
+)
+
+# ── Models ───────────────────────────────────────────────
+class PredictRequest(BaseModel):
+    text:       str
+    session_id: Optional[str] = None
+
+class IntentResult(BaseModel):
+    intent:     str
+    confidence: float
+
+class PredictResponse(BaseModel):
+    session_id:        str
+    input_text:        str
+    cleaned_text:      str
+    intent:            str
+    confidence:        float
+    top3:              list[IntentResult]
+    last_intent:       Optional[str] = None
+    is_low_confidence: bool = False
+    is_oos:            bool = False
+    response:          str = ""
+
+# ── Routes ───────────────────────────────────────────────
+@app.get("/")
+def root():
+    return {"message": "Intent Chatbot API", "status": "running", "intents": len(label_encoder.classes_)}
+
+@app.get("/health")
+def health():
+    return {"status": "healthy", "model": "bert-base-uncased", "intents": len(label_encoder.classes_)}
+
+@app.post("/predict", response_model=PredictResponse)
+def predict_intent(request: PredictRequest):
+    text = request.text.strip()
+    if not text:
+        raise HTTPException(status_code=400, detail="Text cannot be empty")
+    if len(text) < 2:
+        raise HTTPException(status_code=400, detail="Text too short")
+    if len(text) > 200:
+        raise HTTPException(status_code=400, detail="Text too long")
+
+    import uuid, json
+    from datetime import datetime
+
+    session_id = request.session_id or str(uuid.uuid4())
+
+    # Get last intent
+    existing = r.get(f"session:{session_id}")
+    messages = json.loads(existing) if existing else []
+    last_intent = None
+    for msg in reversed(messages):
+        if msg.get("intent"):
+            last_intent = msg["intent"]
+            break
+
+    result = predict(text)
+
+    # Save to Redis
+    messages.append({"role": "user", "text": text, "timestamp": datetime.now().isoformat()})
+    messages.append({"role": "bot", "text": result["intent"], "intent": result["intent"], "confidence": result["confidence"], "timestamp": datetime.now().isoformat()})
+    r.setex(f"session:{session_id}", 3600, json.dumps(messages))
+
+    return PredictResponse(
+        session_id=session_id,
+        input_text=result["input_text"],
+        cleaned_text=result["cleaned_text"],
+        intent=result["intent"],
+        confidence=result["confidence"],
+        top3=[IntentResult(**r_) for r_ in result["top3"]],
+        last_intent=last_intent,
+        is_low_confidence=result["is_low_confidence"],
+        is_oos=result["is_oos"],
+        response=result["response"]
+    )
+
+@app.get("/history/{session_id}")
+def get_history(session_id: str):
+    existing = r.get(f"session:{session_id}")
+    if not existing:
+        raise HTTPException(status_code=404, detail="Session not found")
+    return {"session_id": session_id, "history": json.loads(existing)}
+
+@app.delete("/history/{session_id}")
+def delete_session(session_id: str):
+    r.delete(f"session:{session_id}")
+    return {"message": "Session cleared"}
+
+@app.get("/intents")
+def get_intents():
+    return {"total": len(label_encoder.classes_), "intents": list(label_encoder.classes_)}
+

requirements.txt

@@ -0,0 +1,10 @@
+torch
+transformers
+fastapi
+uvicorn
+pydantic
+scikit-learn
+numpy
+redis
+huggingface_hub
+python-dotenv