Intermediate: Chatbot API + Image Classifier Service

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import OpenAI
import redis, json

app = FastAPI()
client = OpenAI()
r = redis.Redis(decode_responses=True)

SYSTEM = "You are a helpful support agent for an e-commerce store. Reply in Bangla."

class Msg(BaseModel):
    session_id: str
    text: str

@app.post("/chat")
async def chat(msg: Msg):
    history = json.loads(r.get(f"chat:{msg.session_id}") or "[]")
    history.append({"role": "user", "content": msg.text})

    def stream():
        full = ""
        resp = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role":"system","content":SYSTEM}, *history],
            stream=True,
        )
        for chunk in resp:
            delta = chunk.choices[0].delta.content or ""
            full += delta
            yield f"data: {json.dumps({'delta': delta})}\n\n"
        history.append({"role":"assistant","content":full})
        r.setex(f"chat:{msg.session_id}", 3600, json.dumps(history[-20:]))
        yield "data: [DONE]\n\n"

    return StreamingResponse(stream(), media_type="text/event-stream")

async function sendMessage(text: string) {
  const res = await fetch("/api/chat", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ session_id: sessionId, text }),
  });
  const reader = res.body!.getReader();
  const decoder = new TextDecoder();
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const lines = decoder.decode(value).split("\n\n");
    for (const line of lines) {
      if (!line.startsWith("data: ")) continue;
      const data = line.slice(6);
      if (data === "[DONE]") return;
      const { delta } = JSON.parse(data);
      appendToMessage(delta);
    }
  }
}

from fastapi import FastAPI, UploadFile, File
import torch, io
from torchvision import models, transforms
from PIL import Image

app = FastAPI()
device = "cuda" if torch.cuda.is_available() else "cpu"
model = models.resnet50(weights="IMAGENET1K_V2").to(device).eval()
labels = [l.strip() for l in open("imagenet_classes.txt")]

tfm = transforms.Compose([
    transforms.Resize(256), transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize([0.485,0.456,0.406],[0.229,0.224,0.225]),
])

@app.post("/classify")
async def classify(file: UploadFile = File(...)):
    img = Image.open(io.BytesIO(await file.read())).convert("RGB")
    x = tfm(img).unsqueeze(0).to(device)
    with torch.no_grad():
        probs = torch.softmax(model(x)[0], dim=0)
    top5 = torch.topk(probs, 5)
    return {
        "predictions": [
            {"label": labels[i], "score": float(p)}
            for p, i in zip(top5.values, top5.indices)
        ]
    }

# tasks.py
from celery import Celery
celery = Celery("img", broker="redis://redis:6379/0", backend="redis://redis:6379/1")

@celery.task
def classify_batch(s3_keys: list[str]):
    results = []
    for k in s3_keys:
        img = download_from_s3(k)
        results.append(classify_image(img))
    return results

# producer
job = classify_batch.delay(["uploads/a.jpg", "uploads/b.jpg"])
# status: job.state, result: job.get(timeout=60)

services:
  api:    { build: ./api, ports: ["8000:8000"], depends_on: [redis] }
  worker: { build: ./api, command: celery -A tasks worker, depends_on: [redis] }
  redis:  { image: redis:7-alpine }
  web:    { build: ./web, ports: ["3000:3000"] }