Celery — Module 18.2

How This Lesson Fits the Module & Volume

Sync POST /v1/chat is fine for short completions. Long jobs—RAG ingest, batch embeddings, A1111/Comfy stills, multi-agent runs—will time out gateways and pin FastAPI workers. Celery (with Redis or RabbitMQ as broker) moves that work off the request thread: enqueue, return 201 + job id, poll or push via WebSockets.

This is the async half of the API contract. Streaming tokens (next-next lecture) is a different pattern: keep-alive on the same request, not a background worker.

Learning Objectives

By the end of this lesson, students should be able to:

Decide sync HTTP vs Celery job vs token streaming.
Define a Celery app, task, broker, and result backend.
Enqueue an OpenAI (or stills) job from FastAPI and poll status.
Scale API Deployments independently from worker Deployments on K8s.
Handle retries, idempotency, and poison tasks without double-billing.
Avoid running GPU infer inside the web replica.

Definition

Celery is a distributed task queue for Python: producers send messages to a broker (Redis/RabbitMQ); workers execute registered tasks; optional result backend stores return values/state. It is not Airflow (DAGs/schedules at data-platform scale) and not asyncio. Think “background jobs for FastAPI.”

Three Ways to Wait on an LLM

Pattern	When	Lecture
Sync POST	< ~15–30s chat	FastAPI
SSE / stream	Token UX, same request open	Streaming
Celery job	Minutes, batches, stills, ingest	This page

FastAPI + Celery Sketch

# worker.py
import os
from celery import Celery
from openai import OpenAI

celery = Celery("jobs", broker=os.environ["REDIS_URL"], backend=os.environ["REDIS_URL"])
oai = OpenAI()

@celery.task(bind=True, max_retries=2, autoretry_for=(Exception,), retry_backoff=True)
def generate_brief(self, prompt: str) -> dict:
    resp = oai.chat.completions.create(
        model="gpt-4.1-mini",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=800,
    )
    return {
        "text": resp.choices[0].message.content or "",
        "input_tokens": resp.usage.prompt_tokens if resp.usage else 0,
        "output_tokens": resp.usage.completion_tokens if resp.usage else 0,
    }

# api.py  (FastAPI)
from fastapi import FastAPI
from pydantic import BaseModel
from worker import generate_brief

app = FastAPI()

class JobIn(BaseModel):
    prompt: str

@app.post("/v1/jobs", status_code=201)
def enqueue(body: JobIn):
    async_res = generate_brief.delay(body.prompt)
    return {"id": async_res.id, "status": "queued"}

@app.get("/v1/jobs/{job_id}")
def status(job_id: str):
    ar = generate_brief.AsyncResult(job_id)
    if ar.state == "PENDING":
        return {"id": job_id, "status": "queued"}
    if ar.state == "SUCCESS":
        return {"id": job_id, "status": "done", "result": ar.result}
    if ar.state == "FAILURE":
        return {"id": job_id, "status": "failed"}
    return {"id": job_id, "status": ar.state.lower()}

# celery -A worker worker --loglevel=INFO
# Separate K8s Deployment: replicas of workers != replicas of uvicorn

Process Topology

Web Deployment

FastAPI / Uvicorn
Auth, validate, enqueue
No 5-minute OpenAI call

Worker Deployment

Celery processes
SDK calls, stills, ingest
Scale on queue depth

Redis

Broker + optional backend
Also cache/rate limits
Not the chat archive

Celery wins

Timeouts live on workers, not the LB
Retries, routing, rate limits per task
Independent scale from HTTP

Watch outs

At-least-once → idempotent tasks
Huge results in Redis will OOM Redis
Do not confuse with token streaming UX

Related Lectures

Lecture	Role
Redis	Broker/backend
WebSockets	Push job completion instead of polling
Streaming	Different pattern: live tokens
ComfyUI	Stills belong on workers, not Uvicorn

Common Misconception

“Celery streaming is how ChatGPT types tokens.” Token UX is SSE/WS on an open connection. Celery is for jobs you can abandon and resume. Second: max_retries without idempotency double-charges OpenAI. Third: storing a 20 MB base64 still in the Redis result backend is how you page Redis. Fourth: one mega-worker replica that also runs Uvicorn is still a monolith.

Knowledge Check

Short Answer: Name Celery’s broker role. Answer: Queue that holds task messages until workers consume them (e.g. Redis/RabbitMQ).
True/False: Long stills/ingest should run inside the FastAPI request thread. Answer: False—enqueue a job.
Multiple Choice: Successful enqueue typically returns: (a) 201 + job id, (b) 204 only, (c) CFG 7. Answer: (a).
Short Answer: Why scale web and workers separately? Answer: HTTP concurrency ≠ job throughput; different bottlenecks and GPU needs.
True/False: Celery replaces SSE token streaming. Answer: False—different UX/pattern.
Multiple Choice: At-least-once delivery implies: (a) idempotent tasks, (b) no Redis, (c) DDPM. Answer: (a).
Short Answer: What does .delay() do? Answer: Sends the task to the broker asynchronously and returns an AsyncResult id.
True/False: Result backends should store huge binary stills by default. Answer: False—store in object storage; Redis keeps ids/status.
Multiple Choice: Next lecture: (a) WebSockets, (b) Naive Bayes, (c) FLUX. Answer: (a).
Short Answer: Name one job type that belongs on Celery vs sync chat. Answer: Any of: batch embeddings, RAG ingest, Comfy/A1111 stills, multi-minute agents.

Key Takeaways

Celery = background jobs; broker + workers + optional results.
201 + poll/push; keep Uvicorn thin.
Retries need idempotency; big blobs ≠ Redis results.
Streaming tokens ≠ Celery.
Next: WebSockets.

Trainer’s Guide

Lab: Enqueue a 20s fake sleep task + a real short OpenAI task. Poll /v1/jobs/{id}. Kill a worker mid-job and discuss retry/idempotency.

Whiteboard: Client → FastAPI → Redis queue → worker → OpenAI. Optional arrow back via WebSocket.

Recap: Celery offloads long AI work. Push updates live next with WebSockets.