Step-by-step: run heavy AI tasks asynchronously with Celery + Redis.
Some AI workloads — document ingestion, batch processing, report generation — take minutes, not milliseconds. Running them inside an API request blocks the response and risks timeouts. Celery lets you offload these tasks to background workers while your API responds immediately. This guide shows how to integrate Celery with FastAPI.
Set up Celery with Redis as the broker
Configure Celery to use Redis as both the message broker and result backend. Create a Celery app instance that your workers and FastAPI app both reference.
# app/worker.py
from celery import Celery
celery_app = Celery(
"ai-tasks",
broker=settings.REDIS_URL,
backend=settings.REDIS_URL,
)
celery_app.conf.update(
task_serializer="json",
result_serializer="json",
accept_content=["json"],
timezone="UTC",
task_track_started=True,
task_acks_late=True, # retry on worker crash
)Create async-safe Celery tasks
Define Celery tasks for heavy operations like document ingestion and batch LLM processing. Use a sync-to-async bridge for tasks that call async functions.
import asyncio
from app.worker import celery_app
@celery_app.task(bind=True, max_retries=3)
def ingest_document_task(self, file_path: str, collection: str):
try:
loop = asyncio.new_event_loop()
loop.run_until_complete(ingest_document(file_path, collection))
except Exception as exc:
self.retry(exc=exc, countdown=60)Trigger background tasks from FastAPI endpoints
Return a task ID immediately while the heavy work runs in a Celery worker. The client can poll for status or receive a webhook when complete.
@router.post("/v1/documents/ingest")
async def start_ingestion(body: IngestRequest, key: APIKey = Depends(verify_api_key)):
task = ingest_document_task.delay(body.file_path, body.collection)
return {"task_id": task.id, "status": "processing"}
@router.get("/v1/tasks/{task_id}")
async def get_task_status(task_id: str):
result = celery_app.AsyncResult(task_id)
return {"task_id": task_id, "status": result.status, "result": result.result if result.ready() else None}Run the Celery worker
Start a Celery worker process alongside your FastAPI app. In production, run workers as a separate service in your Docker Compose or deployment platform.
# Development
celery -A app.worker worker --loglevel=info --concurrency=2
# docker-compose.yml
services:
api:
build: .
command: uvicorn app.main:app --host 0.0.0.0 --port 8000
worker:
build: .
command: celery -A app.worker worker --loglevel=info --concurrency=2
redis:
image: redis:7-alpine
ports:
- "6379:6379"Monitor tasks and handle failures
Add task monitoring with Flower (Celery's web UI) and configure retry logic, dead-letter queues, and failure notifications.
# Install Flower for monitoring
pip install flower
celery -A app.worker flower --port=5555
# Task with failure handling
@celery_app.task(bind=True, max_retries=3, soft_time_limit=300)
def batch_process_task(self, items: list[str]):
try:
for item in items:
process_item(item)
except SoftTimeLimitExceeded:
self.retry(countdown=120)
except Exception as exc:
# Log failure, send notification
logger.error(f"Batch processing failed: {exc}")
self.retry(exc=exc, countdown=60)Skip the setup, start building
Everything in this guide is already implemented and configured in FastAPI AI Kit. Clone the repo and start building your product features immediately.
Ready to ship your AI backend this weekend?
Join developers who skipped weeks of boilerplate and went straight to building.