Django Celery tasks for async LLM API calls

Blocking a Django request on a 20-second LLM completion is a quick way to exhaust your worker pool and anger users. The standard fix is to push the work into a background queue, and this walkthrough implements a production-shaped django celery async llm api integration that handles retries, timeouts, and provider degradation. You will end up with a view that fires a task and a worker that calls an OpenAI-compatible endpoint without coupling request latency to model response time.

Step 1: Scaffold the Django and Celery baseline

Install the dependencies and create a minimal project. Redis serves as both broker and result store; it is the path of least resistance for local dev and small clusters.

pip install django celery redis requests
django-admin startproject core .
python manage.py startapp api

Create core/celery.py to instantiate the Celery app and pull configuration from Django settings:

from celery import Celery
import os

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "core.settings")

app = Celery("core")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

Expose the app from core/__init__.py so Django imports it on startup:

from .celery import app as celery_app
__all__ = ("celery_app",)

Step 2: Configure the broker and result backend

Celery reads its transport configuration from Django settings when using the CELERY namespace. Put these in core/settings.py:

CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/1"
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_ACCEPT_CONTENT = ["json"]

Separate Redis DB indices for broker and results prevents a large result set from starving the queue. If you run Redis locally, start it with redis-server. For containerized deployments, point the URLs at the service name.

Step 3: Build a resilient LLM client wrapper

Keep the HTTP call isolated from Celery so you can unit-test it without a worker. The function below posts to any OpenAI-compatible /chat/completions endpoint.

import os
import requests

def call_llm(prompt: str, model: str = "gpt-4o-mini") -> str:
    # n4n.ai exposes one OpenAI-compatible endpoint covering 240+ models
    # and automatically fails over when a provider is rate-limited or degraded.
    base_url = os.getenv("LLM_BASE_URL", "https://api.openai.com/v1")
    api_key = os.getenv("LLM_API_KEY")
    resp = requests.post(
        f"{base_url}/chat/completions",
        headers={"Authorization": f"Bearer {api_key}"},
        json={
            "model": model,
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.2,
        },
        timeout=30,
    )
    resp.raise_for_status()
    data = resp.json()
    return data["choices"][0]["message"]["content"]

Set LLM_BASE_URL and LLM_API_KEY in your environment. A hard timeout is non-negotiable; without it a stalled connection hangs the worker indefinitely.

Step 4: Define the Celery task for the async LLM call

Celery tasks are synchronous by default. Wrap the client in a @shared_task with bounded retries and late acknowledgment so a worker crash does not drop the job.

import requests
from celery import shared_task
from .llm import call_llm
import logging

logger = logging.getLogger(__name__)

@shared_task(bind=True, max_retries=3, default_retry_delay=5, acks_late=True)
def generate_llm_reply(self, prompt: str, model: str):
    try:
        return call_llm(prompt, model)
    except requests.RequestException as exc:
        logger.warning("LLM call failed: %s", exc)
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

acks_late=True means the task is only acknowledged after it finishes or exhausts retries, giving you at-least-once semantics. The exponential countdown backs off on transient 429s.

Step 5: Trigger the task from a Django view

The web layer should never await the model. It validates input, enqueues the task, and returns a task ID immediately.

from django.http import JsonResponse
from .tasks import generate_llm_reply

def submit_prompt(request):
    if request.method != "POST":
        return JsonResponse({"error": "POST only"}, status=405)
    prompt = request.POST.get("prompt")
    if not prompt:
        return JsonResponse({"error": "prompt required"}, status=400)
    task = generate_llm_reply.delay(prompt, "gpt-4o-mini")
    return JsonResponse({"task_id": task.id}, status=202)

Wire it in api/urls.py and include that in the root urls.py. Returning 202 Accepted signals the async contract clearly.

Step 6: Track state and retrieve results

Expose a lightweight status endpoint backed by AsyncResult. Do not poll faster than once per second in production UIs.

from celery.result import AsyncResult
from django.http import JsonResponse

def task_status(request, task_id):
    res = AsyncResult(task_id)
    return JsonResponse({
        "state": res.state,
        "result": res.result if res.ready() else None,
    })

States flow PENDING → STARTED → SUCCESS (or FAILURE). On FAILURE, res.result holds the exception info if you configured task_track_started=True.

Step 7: Hardening for production

The happy path above breaks under concurrency. Add three things before shipping:

Idempotency. Hash the prompt+model and use a Redis SETNX lock with a TTL before enqueueing, so duplicate submissions from a jittery frontend do not double-spend tokens.

Usage metering. If your gateway provides per-token usage metering, capture it from the response and persist it. For example, n4n.ai returns standard usage fields and meters per token, so you can write data["usage"] into a billing row inside call_llm after the call succeeds.

Worker isolation. Run LLM tasks in a separate Celery queue (generate_llm_reply.apply_async(queue="llm")) and start a dedicated worker pool for that queue. This stops a slow model from starving your email-sending tasks.

celery -A core worker -Q llm -c 4 -l info
celery -A core worker -Q default -c 8 -l info

Step 8: Verify the pipeline end to end

Start the infrastructure and exercise the flow.

redis-server &
celery -A core worker -Q llm -c 2 -l info &
python manage.py runserver 8000

Submit a job:

curl -X POST localhost:8000/api/submit/ -d "prompt=Explain a Celery beat in one sentence"

You should receive {"task_id": "..."} with a 202 status. Poll:

curl localhost:8000/api/status/<task_id>/

Initially state is PENDING or STARTED. Within a few seconds (depending on model latency) it becomes SUCCESS and result contains the generated text. If you kill the worker mid-flight, the task requeues because of acks_late. That round trip proves your django celery async llm api path is wired correctly and resilient to worker restarts.