Deploy FastAPI app

.env

@@ -0,0 +1 @@
+GEMINI_API_KEY=

Dockerfile

@@ -0,0 +1,20 @@
+FROM python:3.10-slim
+
+WORKDIR /app
+
+COPY . /app
+
+RUN pip install --upgrade pip
+
+# Install CPU-only PyTorch
+RUN pip install --no-cache-dir \
+    torch==2.1.2+cpu \
+    torchvision==0.16.2+cpu \
+    torchaudio==2.1.2+cpu \
+    --index-url https://download.pytorch.org/whl/cpu
+
+RUN pip install --no-cache-dir -r requirements.txt
+
+EXPOSE 7860
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

app.py

@@ -0,0 +1,107 @@
+import torch
+from fastapi import FastAPI
+from transformers import AutoTokenizer, AutoModelForSequenceClassification, AutoConfig
+
+from models import (
+    QueryRequest,
+    QueryResponse,
+    CategoryPrediction,
+    UrgencyPrediction
+)
+from rag import generate_answer
+from memory import get_conversation, add_message
+
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+
+CLASSIFIER_MODEL_ID = "your-org/your-multitask-model"
+
+tag_classes = [
+    "Billing",
+    "Network & Connectivity",
+    "Account Access",
+    "Hardware",
+    "Other"
+]
+
+urgency_encoder = {
+    0: "low",
+    1: "medium",
+    2: "high"
+}
+
+tokenizer = AutoTokenizer.from_pretrained(CLASSIFIER_MODEL_ID)
+config = AutoConfig.from_pretrained(CLASSIFIER_MODEL_ID)
+
+model = AutoModelForSequenceClassification.from_pretrained(
+    CLASSIFIER_MODEL_ID,
+    config=config,
+    trust_remote_code=True
+).to(DEVICE)
+
+model.eval()
+
+app = FastAPI(title="RAG + Conversation Memory API")
+
+# ---------------------
+# CLASSIFIER
+# ---------------------
+def classify_text(text: str, threshold: float = 0.5):
+    inputs = tokenizer(text, return_tensors="pt", truncation=True).to(DEVICE)
+
+    with torch.no_grad():
+        outputs = model(**inputs)
+
+    category_probs = torch.sigmoid(outputs.category_logits)[0].cpu().numpy()
+
+    categories = [
+        CategoryPrediction(
+            category=tag_classes[i],
+            confidence=float(category_probs[i])
+        )
+        for i in range(len(tag_classes))
+        if category_probs[i] >= threshold
+    ]
+
+    urgency_probs = torch.softmax(outputs.urgency_logits, dim=-1)[0].cpu().numpy()
+    urgency_idx = int(torch.argmax(outputs.urgency_logits, dim=-1)[0])
+
+    urgency = UrgencyPrediction(
+        label=urgency_encoder[urgency_idx],
+        confidence=float(urgency_probs[urgency_idx])
+    )
+
+    return categories, urgency
+
+
+def retrieve_documents(query: str):
+    return [
+        "Restarting the router fixes most connectivity issues.",
+        "Check for planned ISP maintenance.",
+        "Verify cables are securely connected."
+    ]
+
+
+@app.post("/query", response_model=QueryResponse)
+def query_endpoint(req: QueryRequest):
+    # ---- Load conversation
+    history = get_conversation(req.user_id)
+
+    # ---- Classification
+    categories, urgency = classify_text(req.query)
+
+    # ---- RAG
+    docs = retrieve_documents(req.query)
+    answer = generate_answer(req.query, docs, history)
+
+    # ---- Update memory
+    add_message(req.user_id, "user", req.query)
+    add_message(req.user_id, "assistant", answer)
+
+    return QueryResponse(
+        user_id=req.user_id,
+        query=req.query,
+        answer=answer,
+        categories=categories,
+        urgency=urgency,
+        conversation=get_conversation(req.user_id)
+    )

memeory.py

@@ -0,0 +1,17 @@
+from collections import defaultdict, deque
+
+MAX_TURNS = 6  # last N messages per user
+
+conversation_store = defaultdict(
+    lambda: deque(maxlen=MAX_TURNS)
+)
+
+
+def get_conversation(user_id: str):
+    return list(conversation_store[user_id])
+
+
+def add_message(user_id: str, role: str, content: str):
+    conversation_store[user_id].append(
+        {"role": role, "content": content}
+    )

models.py

@@ -0,0 +1,31 @@
+from pydantic import BaseModel
+from typing import List
+
+
+class QueryRequest(BaseModel):
+    user_id: str
+    query: str
+
+
+class CategoryPrediction(BaseModel):
+    category: str
+    confidence: float
+
+
+class UrgencyPrediction(BaseModel):
+    label: str
+    confidence: float
+
+
+class Message(BaseModel):
+    role: str
+    content: str
+
+
+class QueryResponse(BaseModel):
+    user_id: str
+    query: str
+    answer: str
+    categories: List[CategoryPrediction]
+    urgency: UrgencyPrediction
+    conversation: List[Message]

rag.py

@@ -0,0 +1,18 @@
+def generate_answer(query: str, retrieved_docs: list[str], history: list[dict]) -> str:
+    history_text = "\n".join(
+        f"{m['role']}: {m['content']}" for m in history
+    )
+
+    context = "\n".join(retrieved_docs[:3])
+
+    return f"""
+Conversation so far:
+{history_text}
+
+Knowledge base:
+{context}
+
+Answer:
+We have received your request regarding "{query}".
+Our support team will assist you shortly.
+""".strip()

requirements.txt

@@ -0,0 +1,12 @@
+fastapi>=0.103,<1
+uvicorn[standard]
+
+torch
+transformers>=4.36,<5
+sentence-transformers>=2.2,<3
+huggingface-hub>=0.20,<1
+accelerate
+
+faiss-cpu
+pandas
+python-dotenv