AI Job Queue Python: Arq, Celery, and RQ for LLM Inference Background Tasks in 2026

By Fadi Updated Apr 6, 2026 14 min read AI Backend

TL;DR — AI job queue Python: Arq, Celery, RQ for LLM inference. Johal: "Arq asyncio-native task queue for AI workloads. 65% of deployments crash under load. Redis backend, retry with backoff, cron jobs, result storage." Davidmuraya: "ARQ FastAPI seamless, fully async, significantly faster than RQ. BackgroundTasks for lightweight, Arq/Celery for heavy CPU-bound." GitHub fastapi-arq: "ARQ 7x faster than RQ for short jobs, 40x with I/O. Asyncio, no forking." Medium: "AI inference inherently async and long-running. Redis broker, retry with backoff, dead letter queue, monitor GPU utilization." Judoscale: "Celery and RQ most popular. Autoscale based on queue latency." Learn more with FastAPI AI tutorial, async backend, Ollama Docker, and Docker AI API.

Johal frames the problem: "Imagine scaling your AI-driven application in 2026, where generative AI models process millions of inference requests daily, but synchronous bottlenecks cause 65% of deployments to crash under load."

Davidmuraya adds the solution: "ARQ integration with FastAPI is seamless and avoids the complexity of bridging synchronous and asynchronous code that you might encounter with Celery. Because it's fully asynchronous, ARQ is significantly faster than synchronous queues like RQ."

AI Job Queue Architecture

flowchart TD subgraph Client["Client"] Submit["POST /jobs
submit LLM inference
return job_id"] Poll["GET /jobs/{id}
poll status
pending/running/complete/failed"] Result["GET /jobs/{id}/result
get result when complete"] end subgraph FastAPI["FastAPI API"] API["Job Endpoints
POST /jobs
GET /jobs/{id}
GET /jobs/{id}/result
webhook callback"] BG["BackgroundTasks
lightweight tasks
in-process
no external queue
email, logging"] end subgraph Queue["Task Queue"] Arq["Arq
asyncio-native
7-40x faster than RQ
Redis backend
retry with backoff
cron jobs
result storage"] Celery["Celery
mature ecosystem
Redis/RabbitMQ
beat scheduler
sync + async
complex config"] RQ["RQ
simple
Redis backend
synchronous
easy setup
basic retry"] end subgraph Redis["Redis Broker"] JobQueue["Job Queue
LPUSH/BRPOP
priority queues
job serialization"] Results["Result Storage
job status
job result
TTL expiry
failed jobs"] DeadLetter["Dead Letter Queue
permanently failed
max retries exceeded
inspection and debug"] end subgraph Worker["Worker Process"] ArqWorker["Arq Worker
async def tasks
max_jobs limit
job timeout
retry with backoff
cron scheduling"] GPU["GPU Management
max_jobs per GPU
VRAM tracking
model in memory
batch processing"] Retry["Retry Logic
exponential backoff
1s, 2s, 4s, 8s, 16s
max_tries 3-5
circuit breaker"] end subgraph LLM["LLM Backend"] Ollama["Ollama
localhost:11434
async httpx call
stream=true"] end Submit --> API API --> Arq API --> BG Arq --> JobQueue Celery --> JobQueue RQ --> JobQueue JobQueue --> ArqWorker ArqWorker --> GPU ArqWorker --> Retry Retry --> JobQueue GPU --> Ollama ArqWorker --> Results Retry --> DeadLetter Results --> Poll Results --> Result style Queue fill:#4169E1,color:#fff style Redis fill:#39FF14,color:#000 style Worker fill:#2D1B69,color:#fff

Queue Comparison

Queue	Async	Speed	Backend	Best For	Complexity
Arq	Native asyncio	7-40x faster	Redis	Async I/O-bound AI	Low
Celery	Sync + async	Mature	Redis/RabbitMQ	Complex workflows	High
RQ	Synchronous	Simple	Redis	Simple jobs	Low
BackgroundTasks	In-process	Lightweight	None	Email, logging	Minimal

Implementation

from dataclasses import dataclass
from typing import Optional
from enum import Enum

class QueueType(Enum):
    ARQ = "arq"
    CELERY = "celery"
    RQ = "rq"
    BG_TASKS = "bg_tasks"

@dataclass
class AIJobQueueGuide:
    """AI job queue Python implementation guide."""

    def get_arq_setup(self) -> str:
        """Arq async task queue setup."""
        return (
            "# === ARQ SETUP ===\n"
            "# pip install arq redis httpx\n"
            "\n"
            "from arq import create_pool\n"
            "from arq.connections import RedisSettings\n"
            "import httpx, asyncio\n"
            "\n"
            "# === TASK FUNCTIONS ===\n"
            "async def llm_inference(ctx, prompt: str,\n"
            "    model: str = 'llama3.2'):\n"
            "    '''LLM inference task.''' \n"
            "    job_id = ctx['job_id']\n"
            "    \n"
            "    async with httpx.AsyncClient(\n"
            "        timeout=300.0\n"
            "    ) as client:\n"
            "        resp = await client.post(\n"
            "            'http://localhost:11434'\n"
            "            '/api/chat',\n"
            "            json={\n"
            "                'model': model,\n"
            "                'messages': [{\n"
            "                  'role': 'user',\n"
            "                  'content': prompt}]})\n"
            "        result = resp.json()\n"
            "        return {\n"
            "            'response': result[\n"
            "              'message']['content'],\n"
            "            'tokens': result.get(\n"
            "              'eval_count', 0),\n"
            "        }\n"
            "\n"
            "async def embed_text(ctx, text: str):\n"
            "    '''Embedding task.''' \n"
            "    async with httpx.AsyncClient(\n"
            "        timeout=60.0\n"
            "    ) as client:\n"
            "        resp = await client.post(\n"
            "            'http://localhost:11434'\n"
            "            '/api/embeddings',\n"
            "            json={\n"
            "                'model':\n"
            "                  'nomic-embed-text',\n"
            "                'prompt': text})\n"
            "        return {'embedding':\n"
            "          resp.json()['embedding']}\n"
            "\n"
            "# === WORKER CONFIG ===\n"
            "class WorkerSettings:\n"
            "    functions = [\n"
            "        llm_inference, embed_text]\n"
            "    redis_settings = RedisSettings(\n"
            "        host='localhost', port=6379)\n"
            "    max_jobs = 4  # GPU capacity\n"
            "    job_timeout = 300  # 5 min\n"
            "    max_tries = 3  # retries\n"
            "    \n"
            "    async def on_startup(ctx):\n"
            "        print('Worker started')\n"
            "    \n"
            "    async def on_shutdown(ctx):\n"
            "        print('Worker shutdown')\n"
            "\n"
            "# Run worker:\n"
            "# arq WorkerSettings"
        )

    def get_fastapi_integration(self) -> str:
        """FastAPI + Arq integration."""
        return (
            "# === FASTAPI + ARQ ===\n"
            "from fastapi import FastAPI, HTTPException\n"
            "from arq import create_pool\n"
            "from arq.connections import RedisSettings\n"
            "from pydantic import BaseModel\n"
            "\n"
            "app = FastAPI()\n"
            "\n"
            "class JobRequest(BaseModel):\n"
            "    prompt: str\n"
            "    model: str = 'llama3.2'\n"
            "\n"
            "class JobResponse(BaseModel):\n"
            "    job_id: str\n"
            "    status: str = 'pending'\n"
            "\n"
            "@app.on_event('startup')\n"
            "async def startup():\n"
            "    app.state.redis = await (\n"
            "      create_pool(RedisSettings(\n"
            "        host='localhost', port=6379)))\n"
            "\n"
            "@app.on_event('shutdown')\n"
            "async def shutdown():\n"
            "    await app.state.redis.close()\n"
            "\n"
            "@app.post('/jobs',\n"
            "  response_model=JobResponse)\n"
            "async def submit_job(req: JobRequest):\n"
            "    '''Submit LLM inference job.''' \n"
            "    job = await (\n"
            "      app.state.redis.enqueue_job(\n"
            "        'llm_inference',\n"
            "        req.prompt,\n"
            "        req.model))\n"
            "    if not job:\n"
            "        raise HTTPException(\n"
            "          503, 'Queue full')\n"
            "    return JobResponse(\n"
            "        job_id=job.job_id)\n"
            "\n"
            "@app.get('/jobs/{job_id}')\n"
            "async def get_status(job_id: str):\n"
            "    '''Poll job status.''' \n"
            "    job = await (\n"
            "      app.state.redis.job(job_id))\n"
            "    if not job:\n"
            "        raise HTTPException(\n"
            "          404, 'Not found')\n"
            "    return {\n"
            "        'job_id': job_id,\n"
            "        'status': job.status,\n"
            "        'result': job.result()}\n"
            "\n"
            "@app.get('/jobs/{job_id}/result')\n"
            "async def get_result(job_id: str):\n"
            "    '''Get job result.''' \n"
            "    job = await (\n"
            "      app.state.redis.job(job_id))\n"
            "    if not job:\n"
            "        raise HTTPException(\n"
            "          404, 'Not found')\n"
            "    result = await job.result()\n"
            "    return result"
        )

    def get_celery_setup(self) -> str:
        """Celery setup for comparison."""
        return (
            "# === CELERY SETUP ===\n"
            "# pip install celery redis\n"
            "\n"
            "from celery import Celery\n"
            "import httpx\n"
            "\n"
            "app = Celery('ai_tasks',\n"
            "    broker='redis://localhost:6379/0',\n"
            "    backend='redis://localhost:6379/1')\n"
            "\n"
            "@app.task(\n"
            "    bind=True,\n"
            "    max_retries=3,\n"
            "    default_retry_delay=10)\n"
            "def llm_inference(self, prompt, model):\n"
            "    '''LLM inference task.''' \n"
            "    try:\n"
            "        with httpx.Client(\n"
            "            timeout=300.0\n"
            "        ) as client:\n"
            "            resp = client.post(\n"
            "                'http://'\n"
            "                'localhost:11434'\n"
            "                '/api/chat',\n"
            "                json={\n"
            "                    'model': model,\n"
            "                    'messages': [{\n"
            "                      'role': 'user',\n"
            "                      'content': prompt}]})\n"
            "            return resp.json()\n"
            "    except Exception as exc:\n"
            "        raise self.retry(exc=exc)\n"
            "\n"
            "# Run worker:\n"
            "# celery -A tasks worker\n"
            "    --loglevel=info\n"
            "    --concurrency=4\n"
            "\n"
            "# Beat scheduler:\n"
            "# celery -A tasks beat\n"
            "\n"
            "# Submit:\n"
            "# result = llm_inference\n"
            "    .delay(prompt, model)\n"
            "# result.ready()\n"
            "# result.get()"
        )

    def get_retry_strategy(self) -> str:
        """Retry and failure handling."""
        return (
            "# === RETRY STRATEGY ===\n"
            "\n"
            "# Arq retry with backoff\n"
            "async def llm_with_retry(ctx, prompt):\n"
            "    '''LLM with exponential\n"
            "    backoff retry.''' \n"
            "    import asyncio\n"
            "    \n"
            "    max_tries = ctx.get(\n"
            "        'max_tries', 3)\n"
            "    for attempt in range(\n"
            "        max_tries):\n"
            "        try:\n"
            "            async with httpx.AsyncClient(\n"
            "                timeout=300.0\n"
            "            ) as client:\n"
            "                resp = await client.post(\n"
            "                    'http://'\n"
            "                    'localhost:11434'\n"
            "                    '/api/chat',\n"
            "                    json={\n"
            "                        'model':'llama3.2',\n"
            "                        'messages': [{\n"
            "                          'role':'user',\n"
            "                          'content':prompt}]})\n"
            "                return resp.json()\n"
            "        except (\n"
            "          httpx.ReadTimeout,\n"
            "          httpx.ConnectError) as e:\n"
            "            if attempt < max_tries - 1:\n"
            "                # Exponential backoff\n"
            "                delay = 2 ** attempt\n"
            "                await asyncio.sleep(delay)\n"
            "            else:\n"
            "                # Max retries hit\n"
            "                #   dead letter\n"
            "                raise\n"
            "    \n"
            "    raise Exception(\n"
            "      'Max retries exceeded')\n"
            "\n"
            "# Circuit breaker\n"
            "failure_count = 0\n"
            "circuit_open = False\n"
            "\n"
            "async def circuit_breaker():\n"
            "    '''Stop retrying if\n"
            "    backend is down.''' \n"
            "    global failure_count,\n"
            "      circuit_open\n"
            "    if failure_count > 10:\n"
            "        circuit_open = True\n"
            "    if circuit_open:\n"
            "        raise Exception(\n"
            "          'Circuit breaker open')\n"
            "\n"
            "# Dead letter queue\n"
            "async def handle_failure(\n"
            "    ctx, job_id, error):\n"
            "    '''Store failed job.''' \n"
            "    # Store in dead letter\n"
            "    #   queue for inspection\n"
            "    await ctx['redis'].set(\n"
            "        f'dlq:{job_id}',\n"
            "        str(error),\n"
            "        ex=86400)  # 24h TTL"
        )

    def get_gpu_management(self) -> str:
        """GPU resource management."""
        return (
            "# === GPU MANAGEMENT ===\n"
            "\n"
            "class WorkerSettings:\n"
            "    # Limit concurrent jobs\n"
            "    #   per GPU\n"
            "    max_jobs = 4  # RTX 4090\n"
            "    \n"
            "    # Job timeout\n"
            "    job_timeout = 300  # 5 min\n"
            "    \n"
            "    # Max retries\n"
            "    max_tries = 3\n"
            "    \n"
            "    # Priority queues\n"
            "    #   via Redis\n"
            "    \n"
            "    async def on_startup(ctx):\n"
            "        # Track GPU memory\n"
            "        ctx['gpu_memory'] = (\n"
            "          await get_gpu_memory())\n"
            "    \n"
            "    async def on_job_start(ctx):\n"
            "        # Check GPU memory\n"
            "        mem = await (\n"
            "          get_gpu_memory())\n"
            "        if mem < 1000:  # MB\n"
            "            raise Exception(\n"
            "              'Insufficient VRAM')\n"
            "    \n"
            "    async def on_job_end(ctx):\n"
            "        # Release resources\n"
            "        pass\n"
            "\n"
            "async def get_gpu_memory():\n"
            "    '''Get available GPU memory.''' \n"
            "    import subprocess\n"
            "    try:\n"
            "        result = subprocess.run(\n"
            "            ['nvidia-smi',\n"
            "             '--query-gpu='\n"
            "             'memory.free',\n"
            "             '--format=csv'],\n"
            "            capture_output=True,\n"
            "            text=True)\n"
            "        # Parse free memory\n"
            "        lines = result.stdout\n"
            "          .strip().split('\\n')\n"
            "        if len(lines) > 1:\n"
            "            return int(lines[1]\n"
            "              .split()[0])\n"
            "    except Exception:\n"
            "        pass\n"
            "    return float('inf')\n"
            "\n"
            "# Priority queue\n"
            "# Real-time: high priority\n"
            "# Batch: low priority\n"
            "# Arq: separate queues\n"
            "#   per priority"
        )

    def get_production_tips(self) -> dict:
        """Production tips."""
        return {
            "arq_vs_celery": "Arq for async I/O-bound AI workloads with FastAPI (7-40x faster). Celery for complex workflows, large teams, existing infra.",
            "max_jobs": "Limit concurrent jobs per worker to GPU capacity. RTX 4090: 2-4 concurrent 7B streams.",
            "retry": "Exponential backoff: 1s, 2s, 4s, 8s, 16s. max_tries 3-5 for LLM inference. Idempotent jobs for safe retry.",
            "dead_letter": "Dead letter queue for permanently failed jobs. Store for inspection and debugging. TTL 24h.",
            "timeout": "job_timeout=300 for LLM, 60 for embeddings. Fail fast rather than hang.",
            "priority": "Priority queues: real-time high, batch low. Separate Arq queues per priority.",
            "scaling": "Scale workers based on queue depth and GPU utilization. Monitor with Prometheus.",
            "model_memory": "Keep model loaded in memory across jobs. Avoid reloading per job.",
            "batch": "Group small inference requests for efficiency. Batch embeddings.",
            "monitoring": "Monitor: queue depth, worker health, failure rate, retry rate, GPU utilization.",
        }

AI Job Queue Checklist

[ ] AI inference is inherently asynchronous and long-running — needs job queue, not synchronous processing
[ ] 65% of AI deployments crash under load due to synchronous bottlenecks — async job queue is essential
[ ] Arq: asyncio-native task queue, Redis backend, 7x faster than RQ for short jobs, 40x with I/O
[ ] Arq: seamless FastAPI integration, avoids sync/async bridging complexity of Celery
[ ] Arq: retry with exponential backoff, max_tries, job timeout, cron jobs, result storage
[ ] Celery: mature ecosystem, Redis/RabbitMQ backend, beat scheduler, sync+async, complex config
[ ] RQ: simple, Redis backend, synchronous, easy setup, basic retry
[ ] FastAPI BackgroundTasks: lightweight, in-process, no external queue — for email, logging, not heavy AI
[ ] FastAPI BackgroundTasks: great for lightweight non-blocking tasks — heavy CPU-bound to Arq or Celery
[ ] Judoscale: autoscales based on queue latency — works with Celery and RQ
[ ] Job submission: POST /jobs with prompt and model, return job_id immediately
[ ] Job polling: GET /jobs/{id} for status (pending/running/complete/failed)
[ ] Job result: GET /jobs/{id}/result when complete
[ ] Webhook: POST result to callback URL when job done
[ ] WebSocket: push result to client when ready
[ ] Arq worker: separate process, async def tasks, calls Ollama via httpx
[ ] Redis: broker for job queue (LPUSH/BRPOP) and result storage with TTL
[ ] Retry: exponential backoff (1s, 2s, 4s, 8s, 16s), max_tries 3-5 for LLM inference
[ ] Retry: idempotent jobs — safe to retry without side effects
[ ] Dead letter queue: permanently failed jobs (max retries exceeded), store for inspection, TTL 24h
[ ] Circuit breaker: stop retrying if backend is down — failure_count threshold
[ ] Job timeout: 300s for LLM inference, 60s for embeddings — fail fast rather than hang
[ ] GPU management: max_jobs per worker matching GPU capacity (RTX 4090: 2-4 concurrent 7B)
[ ] GPU memory: track VRAM usage, reject jobs if insufficient memory
[ ] Priority queue: real-time high priority, batch low priority — separate Arq queues
[ ] Worker scaling: scale based on queue depth and GPU utilization
[ ] Model loading: keep model in memory across jobs — avoid reloading per job
[ ] Batch processing: group small inference requests for efficiency
[ ] Worker process: one worker per GPU or per model
[ ] Monitoring: queue depth, worker health, failure rate, retry rate, GPU utilization
[ ] Huey, Dramatiq as alternative task queues
[ ] Read FastAPI AI tutorial for API setup
[ ] Read async backend for async patterns
[ ] Read Ollama Docker for LLM deployment
[ ] Read Docker AI API for container deployment
[ ] Test: job submission returns job_id immediately
[ ] Test: worker processes job and stores result
[ ] Test: polling returns correct status transitions
[ ] Test: retry with backoff on transient failures
[ ] Test: dead letter queue catches permanently failed jobs
[ ] Test: max_jobs limits concurrent GPU workloads
[ ] Test: priority queue processes high-priority first
[ ] Test: job timeout fails fast rather than hanging
[ ] Document queue choice, worker config, retry strategy, GPU limits, monitoring metrics

FAQ

What are the main Python task queues for AI workloads in 2026?

Arq (asyncio-native), Celery (mature), RQ (simple), and FastAPI BackgroundTasks (lightweight). Johal: "Arq is the asyncio-native task queue revolutionizing AI workload management. 65% of deployments crash under load due to synchronous bottlenecks." Davidmuraya: "ARQ integration with FastAPI is seamless, avoids complexity of bridging sync and async that you encounter with Celery. ARQ fully asynchronous, significantly faster than synchronous queues like RQ. FastAPI BackgroundTasks great for lightweight, non-blocking tasks like sending welcome email. For heavy CPU-bound jobs, offload to dedicated task queues like Arq or Celery." Judoscale: "Celery and RQ are most popular. Huey, Dramatiq as alternatives. Judoscale autoscales based on queue latency." Queues: (1) Arq: asyncio-native, 7x faster than RQ, Redis backend, FastAPI seamless. (2) Celery: mature, Redis/RabbitMQ, complex, sync+async. (3) RQ: simple, Redis, synchronous. (4) FastAPI BackgroundTasks: lightweight, in-process, no external queue.

How does Arq compare to Celery for AI inference queues?

Arq is asyncio-native and faster for I/O-bound AI workloads; Celery is more mature with broader ecosystem. Davidmuraya: "ARQ integration with FastAPI seamless — avoids complexity of bridging sync and async code with Celery. ARQ fully asynchronous, significantly faster than synchronous queues like RQ." GitHub fastapi-arq: "Asyncio and no forking make ARQ around 7x faster than RQ for short jobs with no I/O. With I/O, that might increase to around 40x faster." Johal: "Arq asyncio-native task queue for AI workload management. Redis backend, cron jobs, retry with backoff, job result storage." Comparison: (1) Arq: asyncio-native, 7-40x faster, Redis, FastAPI seamless, retry/backoff, cron. (2) Celery: mature, Redis/RabbitMQ, broad ecosystem, complex config, beat scheduler, sync+async. (3) Use Arq for async I/O-bound AI workloads with FastAPI. (4) Use Celery for complex workflows, large teams, existing infrastructure.

How do you handle retries and failures in AI job queues?

Use exponential backoff, max retries, dead letter queue, and job timeout. Johal: "Arq: retry with exponential backoff, max_tries configurable, job timeout, result storage. Failed jobs stored for inspection." Medium AI Inference Queue: "AI inference is inherently asynchronous and long-running. Redis as broker, Celery workers, RabbitMQ for complex routing. Retry failed inference with backoff. Dead letter queue for permanently failed jobs. Monitor queue depth, worker health, GPU utilization." Retry handling: (1) Exponential backoff: 1s, 2s, 4s, 8s, 16s. (2) max_tries: 3-5 for LLM inference. (3) Job timeout: 300s for LLM, 60s for embeddings. (4) Dead letter queue: permanently failed jobs. (5) Idempotent jobs: safe to retry. (6) Circuit breaker: stop retrying if backend down. (7) Monitor: queue depth, failure rate, retry rate.

How do you integrate a job queue with FastAPI for LLM inference?

Submit job to queue, return job ID, client polls for result or receives webhook. Davidmuraya: "FastAPI BackgroundTasks for lightweight tasks. ARQ for heavy CPU-bound jobs. Submit to Redis queue, ARQ worker processes, store result." GitHub fastapi-arq: "FastAPI project for managing background task queues using ARQ and Redis." Integration: (1) POST /jobs: submit LLM inference job to Arq queue, return job_id. (2) GET /jobs/{id}: poll job status (pending/running/complete/failed). (3) GET /jobs/{id}/result: get job result when complete. (4) Webhook: POST result to callback URL when done. (5) WebSocket: push result to client when ready. (6) Arq worker: separate process, async def task, calls Ollama. (7) Redis: broker for job queue and result storage. (8) FastAPI: API for job submission and status.

How do you manage GPU resources in AI job queues?

Use worker concurrency limits, GPU memory tracking, and priority queues. Medium AI Inference Queue: "Monitor GPU utilization. Scale workers based on queue depth. Priority queue for real-time vs batch jobs. Redis as broker for job distribution." Johal: "Arq: max_jobs per worker, job timeout, cron jobs for scheduled inference. Worker concurrency configurable." GPU management: (1) max_jobs: limit concurrent jobs per worker to GPU capacity. (2) GPU memory: track VRAM usage, reject jobs if insufficient. (3) Priority queue: real-time high priority, batch low priority. (4) Worker scaling: scale based on queue depth and GPU utilization. (5) Job timeout: 300s for LLM, 60s for embeddings. (6) Worker process: one worker per GPU or per model. (7) Model loading: keep model in memory across jobs. (8) Batch processing: group small inference requests.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →