Building a Background Job Queue for LLM Applications: Offloading Slow AI Tasks with Celery and Redis

Introduction

Imagine you've built a document analysis tool that takes a 50-page PDF, sends it to an LLM for summarization, extracts entities, and writes the results back to a database. It works beautifully on your laptop. Then you deploy it, your first real user submits a document, and they sit there watching a spinner for 45 seconds while your web server holds its breath. Meanwhile, your server can't handle any other requests, and if the user gets impatient and closes the tab, the whole operation just... dies.

This is the most common scaling wall that LLM application developers hit. The problem isn't the AI — it's that you're doing expensive, slow, unreliable work synchronously inside a request/response cycle that was designed for millisecond operations. LLM calls can take anywhere from 2 seconds to 2 minutes depending on task complexity. That's not a web request — that's a background job.

In this lesson, you'll learn how to architect your LLM application so that slow AI tasks run in the background while your web server stays responsive. By the end, you'll have a working pattern for submitting jobs, checking their status, and retrieving results — the same fundamental architecture used by production AI systems at scale.

What you'll learn:

• Why synchronous LLM calls break under real user load, and what the alternative architecture looks like
• How Redis acts as a message broker to pass jobs between your web server and worker processes
• How to define and register Celery tasks that wrap LLM API calls
• How to submit jobs from a FastAPI endpoint and poll for results
• How to handle retries, timeouts, and failure states gracefully

---

Prerequisites

You should be comfortable writing Python and have a basic understanding of making HTTP requests. Some familiarity with using the OpenAI API with Python is helpful since the examples use OpenAI calls, but the patterns apply to any LLM provider. You'll need Docker installed to run Redis locally without a complex setup.

---

Why Synchronous LLM Calls Break Under Load

Before we build anything, let's understand exactly why the naive approach fails. When a user submits a request to your FastAPI or Flask app, that request occupies a worker thread (or coroutine) until a response is sent back. For a database query, that's 10 milliseconds. For an LLM call that processes a large document, it's 20–60 seconds.

During that time, three bad things can happen:

1. Your server runs out of workers. If you have 4 worker threads and 4 users simultaneously submit documents, the 5th user gets a timeout before you've even touched their data.
2. Network interruptions kill the job. If the user's browser disconnects or the load balancer drops the connection after 30 seconds, your expensive LLM call gets orphaned or cancelled.
3. You have no retry mechanism. If the LLM API returns a rate limit error mid-request, the user just sees a 500 error. All that work is lost.

The solution is a queue-based architecture: your web server accepts the request immediately, puts a description of the work into a queue, returns a job ID to the user, and then a separate worker process picks up that job and does the actual LLM work in the background.

User → FastAPI (accepts job, returns ID immediately)
                        ↓
                   Redis Queue
                        ↓
                Celery Worker (runs LLM call)
                        ↓
                   Redis (stores result)
                        ↓
User polls /status/{job_id} → gets result when ready

This decoupling means your web server is always fast, workers can be scaled independently, and failed jobs can be retried automatically.

---

The Tools: Redis and Celery

Redis is an in-memory data store that doubles as an excellent message broker. Think of it as a shared whiteboard that both your web server and your workers can read and write to. Your web server writes "please do this job" on the whiteboard, and a worker picks it up. After the job finishes, Redis also stores the result so your web server can retrieve it later.

Celery is a Python task queue library that sits on top of Redis (or other brokers like RabbitMQ). It handles all the complexity of distributing work: serializing task arguments, routing tasks to workers, managing retries, and storing results. You write regular Python functions decorated with @celery_app.task, and Celery handles the rest.

Setting Up Redis with Docker

The fastest way to get Redis running locally is Docker. Open a terminal and run:

docker run -d --name redis-llm-queue -p 6379:6379 redis:7-alpine

This starts a Redis server on port 6379. You can verify it's running with:

docker exec -it redis-llm-queue redis-cli ping
# Should return: PONG

Installing Dependencies

Create a fresh virtual environment and install the required packages:

pip install celery[redis] redis fastapi uvicorn openai python-dotenv

---

Project Structure

Before writing code, let's lay out the files we'll create. Good structure separates concerns clearly:

llm_queue_app/
├── .env                  # API keys and config
├── celery_app.py         # Celery application instance
├── tasks.py              # Task definitions (the LLM calls)
├── main.py               # FastAPI web server
└── requirements.txt

This separation is intentional. celery_app.py defines the Celery instance so it can be imported by both tasks.py and main.py without circular imports. Workers import tasks.py to discover what tasks exist. The web server imports tasks.py to call .delay() on those tasks.

---

Configuring the Celery Application

Create celery_app.py:

# celery_app.py
from celery import Celery

REDIS_URL = "redis://localhost:6379/0"

celery_app = Celery(
    "llm_tasks",
    broker=REDIS_URL,
    backend=REDIS_URL,
    include=["tasks"]  # Tells Celery which modules contain task definitions
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    result_expires=3600,          # Results stored for 1 hour
    task_acks_late=True,          # Acknowledge task AFTER completion, not before
    worker_prefetch_multiplier=1, # One task per worker at a time (better for slow tasks)
)

Two configuration options here deserve explanation:

task_acks_late=True means the worker doesn't tell Redis "I received this task" until after it completes. If the worker crashes mid-task, Redis will re-queue the task for another worker. For LLM jobs that might take a minute, this matters.

worker_prefetch_multiplier=1 tells each worker to only grab one task at a time. The default is 4, which means a worker grabs 4 tasks and holds them. For fast tasks this is an optimization, but for slow LLM jobs, it means 3 jobs sit idle waiting when other workers might be free.

---

Defining LLM Tasks

Now create tasks.py. This is where your actual LLM logic lives:

# tasks.py
import os
from openai import OpenAI
from celery_app import celery_app
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


@celery_app.task(
    bind=True,
    max_retries=3,
    default_retry_delay=10,  # seconds between retries
    name="tasks.analyze_document"
)
def analyze_document(self, document_text: str, analysis_type: str) -> dict:
    """
    Analyze a document using an LLM.
    
    Args:
        document_text: The raw text content to analyze
        analysis_type: One of 'summary', 'entities', 'sentiment'
    
    Returns:
        dict with 'analysis_type', 'result', and 'word_count'
    """
    prompts = {
        "summary": "Summarize the following document in 3-5 sentences:\n\n",
        "entities": "Extract all named entities (people, organizations, locations, dates) "
                    "from the following document. Return as a structured list:\n\n",
        "sentiment": "Analyze the overall sentiment and tone of the following document. "
                     "Identify key emotional themes:\n\n"
    }
    
    if analysis_type not in prompts:
        raise ValueError(f"Unknown analysis_type: {analysis_type}")
    
    prompt = prompts[analysis_type] + document_text[:8000]  # Respect token limits
    
    try:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "You are a precise document analyst. "
                                               "Return structured, accurate results."},
                {"role": "user", "content": prompt}
            ],
            temperature=0.2,
        )
        
        return {
            "analysis_type": analysis_type,
            "result": response.choices[0].message.content,
            "word_count": len(document_text.split()),
            "tokens_used": response.usage.total_tokens
        }
        
    except Exception as exc:
        # Celery's retry mechanism: re-raise via self.retry()
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)


@celery_app.task(
    bind=True,
    max_retries=3,
    default_retry_delay=15,
    name="tasks.batch_analyze_document"
)
def batch_analyze_document(self, document_text: str) -> dict:
    """
    Run all three analysis types on a document and return combined results.
    This is a 'slow' task that chains multiple LLM calls.
    """
    results = {}
    analysis_types = ["summary", "entities", "sentiment"]
    
    for analysis_type in analysis_types:
        # Update task state so the caller can see progress
        self.update_state(
            state="PROGRESS",
            meta={"current": analysis_types.index(analysis_type) + 1,
                  "total": len(analysis_types),
                  "step": f"Running {analysis_type} analysis..."}
        )
        
        # Call the OpenAI API for each analysis type
        try:
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": "You are a precise document analyst."},
                    {"role": "user", "content": 
                        f"Perform a {analysis_type} analysis on this document:\n\n"
                        + document_text[:6000]}
                ],
                temperature=0.2,
            )
            results[analysis_type] = response.choices[0].message.content
            
        except Exception as exc:
            raise self.retry(exc=exc, countdown=2 ** self.request.retries)
    
    return {"analyses": results, "word_count": len(document_text.split())}

Notice the bind=True parameter in the decorator. This gives the task access to self, which is the task instance itself. Without self, you can't call self.retry() or self.update_state(). Those two methods are what turn a plain function into a smart, observable background job.

The retry pattern countdown=2 ** self.request.retries implements exponential backoff — wait 1 second after the first failure, 2 after the second, 4 after the third. This is essential when dealing with rate-limited APIs. For more on managing API failures gracefully, see the guide on handling LLM errors, timeouts, and rate limits in Python applications.

---

Building the FastAPI Web Server

Now create main.py — the web layer that accepts user requests and dispatches jobs:

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from celery.result import AsyncResult
from tasks import analyze_document, batch_analyze_document

app = FastAPI(title="LLM Document Analysis API")


class SingleAnalysisRequest(BaseModel):
    document_text: str = Field(..., min_length=50, description="Text to analyze")
    analysis_type: str = Field(..., pattern="^(summary|entities|sentiment)$")


class BatchAnalysisRequest(BaseModel):
    document_text: str = Field(..., min_length=50, description="Text to analyze")


@app.post("/analyze", status_code=202)
def submit_analysis(request: SingleAnalysisRequest):
    """
    Submit a document for single-type analysis.
    Returns immediately with a job_id — does NOT wait for the LLM.
    """
    task = analyze_document.delay(
        document_text=request.document_text,
        analysis_type=request.analysis_type
    )
    
    return {
        "job_id": task.id,
        "status": "queued",
        "message": f"Analysis job submitted. Poll /jobs/{task.id} for results."
    }


@app.post("/analyze/batch", status_code=202)
def submit_batch_analysis(request: BatchAnalysisRequest):
    """Submit a document for all three analysis types simultaneously."""
    task = batch_analyze_document.delay(document_text=request.document_text)
    
    return {
        "job_id": task.id,
        "status": "queued",
        "message": f"Batch analysis submitted. Poll /jobs/{task.id} for results."
    }


@app.get("/jobs/{job_id}")
def get_job_status(job_id: str):
    """
    Check the status of a submitted job.
    State will be: PENDING → PROGRESS → SUCCESS or FAILURE
    """
    result = AsyncResult(job_id)
    
    if result.state == "PENDING":
        return {
            "job_id": job_id,
            "status": "pending",
            "message": "Job is waiting in queue or does not exist."
        }
    
    elif result.state == "PROGRESS":
        # We get this when the task calls self.update_state()
        return {
            "job_id": job_id,
            "status": "in_progress",
            "progress": result.info  # Contains the meta dict we set in the task
        }
    
    elif result.state == "SUCCESS":
        return {
            "job_id": job_id,
            "status": "complete",
            "result": result.result
        }
    
    elif result.state == "FAILURE":
        return {
            "job_id": job_id,
            "status": "failed",
            "error": str(result.info)  # result.info contains the exception
        }
    
    else:
        return {
            "job_id": job_id,
            "status": result.state,
            "info": str(result.info)
        }


@app.get("/health")
def health_check():
    return {"status": "ok"}

The key design decision here is the HTTP 202 status code on the POST endpoints. Status 202 means "Accepted" — the server received your request and will work on it, but the result isn't ready yet. This is semantically more honest than 200 ("OK, here's your result") for async operations.

---

Running the System

You need three terminal windows to run the full system:

Terminal 1 — Start Redis (if not already running):

docker run -d --name redis-llm-queue -p 6379:6379 redis:7-alpine

Terminal 2 — Start the Celery worker:

celery -A celery_app worker --loglevel=info --concurrency=4

The --concurrency=4 flag starts 4 worker processes. Each can handle one task simultaneously, so you can process 4 documents at once. For LLM applications where the bottleneck is the API, 4–8 concurrent workers is usually sufficient.

Terminal 3 — Start the FastAPI server:

uvicorn main:app --reload --port 8000

Now test it end-to-end. Open another terminal:

# Submit a job
curl -X POST http://localhost:8000/analyze \
  -H "Content-Type: application/json" \
  -d '{
    "document_text": "OpenAI was founded in December 2015 by Sam Altman, Greg Brockman, Ilya Sutskever, Elon Musk, and others. The company created GPT-4, DALL-E, and ChatGPT, which launched in November 2022 and reached 100 million users in two months. The company is headquartered in San Francisco, California.",
    "analysis_type": "entities"
  }'

# You'll get back something like:
# {"job_id": "a3f2c1b4-...", "status": "queued", "message": "..."}

# Poll for status (replace with your actual job_id)
curl http://localhost:8000/jobs/a3f2c1b4-...

The first status call will likely return "pending". Wait 5–10 seconds and call it again — you'll see "complete" with the extracted entities.

---

Hands-On Exercise

Now that the basic system works, extend it with a real-world scenario: a report generation pipeline.

Your task is to build a task called generate_report that:

1. Takes a company_name (string) and a list of financial_metrics (a list of dicts like {"metric": "revenue", "value": "$4.2B", "yoy_change": "+12%"})
2. Sends this data to an LLM asking it to write a 2-paragraph executive summary in a professional tone
3. Returns the generated report text along with the number of metrics analyzed

Here's a skeleton to get you started:

@celery_app.task(bind=True, max_retries=3, name="tasks.generate_report")
def generate_report(self, company_name: str, financial_metrics: list) -> dict:
    # Format the metrics into a readable string
    metrics_text = "\n".join(
        [f"- {m['metric']}: {m['value']} ({m['yoy_change']} YoY)" 
         for m in financial_metrics]
    )
    
    prompt = f"""Write a 2-paragraph executive summary for {company_name} 
    based on the following financial metrics:
    
    {metrics_text}
    
    Use a professional, board-level tone. Focus on key insights and trends."""
    
    # Your implementation here: call the OpenAI API, handle retries,
    # and return a dict with 'report_text' and 'metrics_analyzed'
    pass

Then add a /reports POST endpoint to main.py that accepts company_name and financial_metrics and returns a job ID.

For bonus points: add a /jobs/{job_id}/cancel endpoint that uses AsyncResult(job_id).revoke(terminate=True) to cancel a queued job.

---

Common Mistakes & Troubleshooting

"My tasks are stuck in PENDING forever"

This almost always means your Celery worker isn't running, or it's running but not connected to the same Redis instance. Check that your worker terminal is active and shows "ready" output. Also verify Redis is running: docker ps | grep redis. A common gotcha is running Redis on a non-default database — make sure the Redis URL in celery_app.py matches what your worker is connecting to.

"The worker crashes when it imports my task module"

This usually happens because the task module tries to do something expensive at import time — like making an API call or loading a large model. Keep your task module clean: only imports and function definitions at module level. Move any initialization inside the task function or use Celery's @worker_process_init signal for one-time setup.

"I'm getting 'Task of kind X is not registered' errors"

You need to make sure Celery knows about your tasks. The include=["tasks"] line in celery_app.py tells Celery to import that module when workers start. If you create new task files, add them to that list. You can verify registered tasks by running celery -A celery_app inspect registered while a worker is running.

"Results disappear before I can retrieve them"

The result_expires=3600 setting means results are deleted from Redis after 1 hour. For applications where users might not check results immediately, increase this. For high-volume applications generating many results, decrease it to avoid filling Redis memory.

"My LLM calls are timing out in the worker"

Celery has its own soft and hard time limits for tasks. If you're doing multi-step LLM analysis that might take 3+ minutes, set explicit limits:

@celery_app.task(
    bind=True,
    soft_time_limit=180,   # Raises SoftTimeLimitExceeded after 3 min (catchable)
    time_limit=240,        # Hard kill after 4 min (not catchable)
)

This is related to a broader challenge in deploying LLM applications: production systems need explicit timeouts at every layer.

"I need to track costs across jobs"

The tokens_used field we return in task results is a good starting point. For more systematic cost tracking, consider writing task completion data (tokens used, model, timestamp) to a database inside your task function. The article on cost optimization: token counting, caching, and model selection covers the token accounting side in depth.

---

Going Further: Patterns Worth Knowing

Chaining Tasks

Celery supports task chaining with chain() for multi-step pipelines. For example, you could chain a document parsing task to a summarization task to a database-write task:

from celery import chain
from tasks import extract_text, analyze_document, save_result

workflow = chain(
    extract_text.s(document_path),
    analyze_document.s(analysis_type="summary"),  # receives output of previous task
    save_result.s(user_id=user_id)
)
job = workflow.apply_async()

This is the foundation of more complex agentic loop design patterns where multi-step LLM pipelines need to handle intermediate failures and state.

Caching Repeated Work

If multiple users submit the same document for analysis, you're paying for redundant LLM calls. You can cache task results by hashing the input before dispatching:

import hashlib
import json

def get_cache_key(document_text: str, analysis_type: str) -> str:
    content = json.dumps({"text": document_text, "type": analysis_type}, sort_keys=True)
    return hashlib.sha256(content.encode()).hexdigest()

Then check Redis for a cached result before dispatching a new task. For more sophisticated caching strategies including semantic deduplication, see implementing LLM response caching with Redis.

Monitoring with Flower

Celery ships with a web-based monitoring tool called Flower. Install and run it with:

pip install flower
celery -A celery_app flower --port=5555

Then open http://localhost:5555 in your browser. You'll see real-time metrics: active tasks, task history, worker status, and failure rates. For production, you'll also want to integrate structured logging into your tasks — the patterns in implementing LLM observability apply directly here.

---

Summary & Next Steps

You've built a complete background job queue for LLM applications. The core pattern is straightforward: accept work immediately, queue it, do it in the background, and let clients poll for results. This architecture is what separates toy demos from production AI systems.

Here's what you built and why each piece matters:

Component	Role	Why it matters
Redis	Message broker + result backend	Durable, fast, shareable state between processes
Celery worker	Executes LLM tasks	Isolated, scalable, independent of web server
FastAPI endpoints	Accept work and serve status	Always responsive, never blocked
task_acks_late=True	Reliability	Jobs survive worker crashes
Exponential backoff retry	Fault tolerance	Handles rate limits and transient failures

To deepen this further, consider these directions:

• Priority queues: Route quick single-analysis jobs to a fast queue and slow batch jobs to a separate queue with dedicated workers
• Result streaming: For long-running tasks, consider pushing intermediate results through WebSockets rather than polling — the patterns in building streaming AI interfaces complement the queue approach well
• More complex pipelines: Once you're comfortable with single tasks, explore Celery's chord and group primitives for fan-out/fan-in patterns where you parallelize multiple LLM calls and aggregate the results

The queue pattern also scales naturally to multi-tenant applications. When you need to enforce per-customer rate limits or usage quotas, the job queue becomes the natural control point — something explored in depth in building a multi-tenant LLM platform.

The rule of thumb: any LLM operation that takes more than 2 seconds belongs in a background job. That's most of them.