Long-running LLM inferences don’t belong in synchronous request handlers. Building async llm jobs celery redis lets you offload generation to a worker pool, scale horizontally, and keep API latency predictable. This tutorial walks through a production-shaped implementation using Python, Celery, and Redis as the broker and result backend.
Prerequisites
- Python 3.11+ with
pip - A running Redis instance (local or Docker)
celery,redis, andopenaiPython packages- An API key for an OpenAI-compatible inference gateway
If you don’t have Redis, start it via Docker:
docker run -d --name redis -p 6379:6379 redis:7-alpineVerify it accepts connections:
redis-cli ping
# Expected: PONGInstall dependencies
Create a virtual environment and install the stack.
python -m venv .venv
source .venv/bin/activate
pip install celery redis openaiPin versions in a requirements.txt for reproducibility:
celery==5.4.0
redis==5.0.4
openai==1.40.0Define the Celery app
Keep configuration explicit. Use separate Redis logical databases for the broker and result backend to avoid key collisions.
# celery_app.py
from celery import Celery
app = Celery(
"llm_jobs",
broker="redis://localhost:6379/0",
backend="redis://localhost:6379/1",
task_serializer="json",
result_serializer="json",
accept_content=["json"],
)
app.conf.update(
task_track_started=True,
task_acks_late=True,
worker_prefetch_multiplier=1,
)task_acks_late ensures a task isn’t removed from the queue until the worker finishes or dies, which matters when a single LLM call can take 30 seconds.
Write the LLM task
We route requests through n4n.ai, a single OpenAI-compatible endpoint covering 240+ models with automatic fallback when a provider is degraded. The OpenAI Python client works unchanged with a custom base_url.
# tasks.py
import os
from openai import OpenAI
from celery_app import app
client = OpenAI(
base_url="https://api.n4n.ai/v1",
api_key=os.environ["N4N_API_KEY"],
)
@app.task(bind=True, max_retries=3, default_retry_delay=5, acks_late=True)
def generate_completion(self, prompt: str, model: str = "gpt-4o-mini"):
try:
resp = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
temperature=0.2,
timeout=60,
)
return {
"model": resp.model,
"content": resp.choices[0].message.content,
"usage": resp.usage.model_dump(),
}
except Exception as exc:
# Exponential backoff via Celery retry
raise self.retry(exc=exc, countdown=2 ** self.request.retries)The timeout parameter protects the worker from hanging on a stalled connection. Celery retries use a countdown that doubles each attempt.
Why async llm jobs celery redis wins over threads
Spawned threads in a web process get killed on deploy. Celery tasks are durable in Redis until acknowledged. For LLM workloads where a prompt might trigger a 20-second generation, that durability is non-negotiable.
Enqueue async LLM jobs
From a separate module or REPL, import the task and dispatch it.
# producer.py
from tasks import generate_completion
if __name__ == "__main__":
result = generate_completion.delay(
"Summarize the trade-offs of eventual consistency in 3 bullets."
)
print(f"Dispatched task {result.id}")Run the producer:
python producer.py
# Expected: Dispatched task a1b2c3d4-e5f6-7890-abcd-1234567890abThe task ID is your handle for later lookup.
Run the worker
Start a dedicated worker for the LLM queue.
celery -A celery_app.app worker -Q llm -n llm-worker@%h --loglevel=infoIf you tagged tasks with a queue, specify it:
@app.task(queue="llm", ...)Worker log excerpt on successful processing:
[2024-06-12 10:01:22,341] Received task: tasks.generate_completion[a1b2c3d4...]
[2024-06-12 10:01:24,902] Task tasks.generate_completion[a1b2c3d4...] succeeded in 2.56sRetrieve results
Poll the result backend by task ID, or block with get() in a separate process.
from tasks import generate_completion
result = generate_completion.AsyncResult("a1b2c3d4-e5f6-7890-abcd-1234567890ab")
print(result.state) # PENDING -> STARTED -> SUCCESS
if result.ready():
print(result.get())Expected output after completion:
{
"model": "gpt-4o-mini",
"content": "• Availability over strict consistency...",
"usage": {"prompt_tokens": 12, "completion_tokens": 34, "total_tokens": 46}
}Don’t block web requests on result.get(). Use a separate status endpoint or webhook.
Production notes
Idempotency and dedupe
LLM calls cost tokens. Wrap the task with a deterministic UUID derived from the prompt hash if you need deduplication:
import hashlib, uuid
def task_id_for(prompt: str) -> str:
return "llm-" + hashlib.sha256(prompt.encode()).hexdigest()[:16]
generate_completion.apply_async(
args=[prompt],
task_id=task_id_for(prompt),
queue="llm",
)Celery ignores duplicate task IDs that are still pending.
Webhooks vs polling
For long jobs, push completion to a URL instead of forcing clients to poll. Add a on_success callback:
from celery import current_app
@app.task(bind=True, ...)
def generate_completion(self, prompt, callback_url=None):
# ... call LLM ...
if callback_url:
requests.post(callback_url, json={"task_id": self.request.id, "result": out})
return outThis keeps async llm jobs celery redis fully event-driven.
Rate limits and fallback
Providers throttle. The gateway we use handles automatic fallback, but your worker should still cap concurrency to avoid hammering the API:
app.conf.worker_concurrency = 4Combine with Redis rate limiting if you share the key across services.
Result expiration
Redis backs results forever by default. Set a TTL:
app.conf.result_expires = 3600 # 1 hourOtherwise your Redis memory grows linearly with job count.
Extending the pattern
Once this core is stable, add:
- A FastAPI endpoint that returns
task_idimmediately and serves status. - A dead-letter queue for tasks that exhaust retries.
- Structured logging with the model and token usage for cost tracking.
The async llm jobs celery redis pattern separates concerns cleanly: Redis brokers, Celery orchestrates, and the gateway abstracts model routing. That’s the backbone for any system running LLMs outside a request lifecycle.