FastAPI + Celery: Async Processing Patterns for AI Workloads

Any LLM or document-processing call that might take more than a couple seconds doesn't belong in the request/response cycle. The question isn't whether to use background jobs for AI workloads — it's how to design them so retries don't duplicate side effects and clients can actually see progress.

The rule of thumb for what goes in a task

If a step's latency is unpredictable (an LLM call, a large file parse, a call to a slow third-party API) or the total pipeline takes more than ~2 seconds, move it to Celery. Keep the HTTP handler responsible only for validating input, enqueuing the job, and returning a job ID immediately.

@router.post("/v1/documents/analyze")
async def analyze_document(file: UploadFile, key: APIKey = Depends(get_api_key)):
    doc_id = await storage.upload(file)
    job = analyze_document_task.delay(doc_id, key.id)
    return {"job_id": job.id, "status": "processing"}

Designing idempotent tasks

Celery guarantees at-least-once execution, not exactly-once. A task can run twice — after a worker crash mid-task, or a broker redelivery. If your task calls an LLM and then writes a result, a duplicate run means a duplicate LLM charge and possibly a duplicate side effect (like sending a notification twice).

@celery.task(bind=True, max_retries=3)
def analyze_document_task(self, doc_id: str, key_id: str):
    # Idempotency guard: skip if already processed
    existing = get_result_if_exists(doc_id)
    if existing:
        return existing

    try:
        text = storage.read_text(doc_id)
        result = llm.extract(text, output_schema=DocumentAnalysis)
        save_result(doc_id, result.model_dump())  # upsert, not insert
        meter.record(key_id, result.tokens)
        return result
    except Exception as exc:
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

The existing = get_result_if_exists(doc_id) check is the important part — it makes a re-run of the same task a no-op instead of a duplicate charge.

Reporting progress on multi-step tasks

A single job ID with a binary "processing"/"done" status is fine for short tasks. For anything with multiple stages (parse → embed → analyze → notify), clients benefit from seeing which stage is active:

@celery.task(bind=True)
def process_pipeline(self, doc_id: str):
    self.update_state(state="PARSING", meta={"progress": 10})
    text = parse_document(doc_id)

    self.update_state(state="EMBEDDING", meta={"progress": 40})
    chunks = embed_and_store(text, doc_id)

    self.update_state(state="ANALYZING", meta={"progress": 75})
    result = llm.extract(text, output_schema=DocumentAnalysis)

    self.update_state(state="DONE", meta={"progress": 100})
    return result

@router.get("/v1/jobs/{job_id}")
async def get_job_status(job_id: str):
    result = celery.AsyncResult(job_id)
    return {"state": result.state, "meta": result.info}

Chaining tasks vs one large task

For pipelines with genuinely independent steps, Celery's chain() primitive is cleaner than one monolithic task function — each step gets its own retry policy and failure isolation:

from celery import chain

pipeline = chain(
    parse_document_task.s(doc_id),
    embed_chunks_task.s(),
    analyze_content_task.s(key_id=key_id),
)
pipeline.apply_async()

If step two fails, step one's result is preserved and step two can retry independently — you're not re-running the parse step just because the embedding step hit a transient error.

Avoiding the thundering herd on worker restart

If your broker has a large backlog when workers restart (deploy, crash, scaling event), all queued tasks fire at once, potentially overwhelming your LLM provider's rate limits. Set worker_prefetch_multiplier=1 for LLM-heavy queues so workers pull one task at a time instead of grabbing a large batch upfront, and consider a dedicated low-concurrency queue for LLM-calling tasks specifically.

What FastAPI AI Kit includes

The kit ships Celery pre-configured with a Redis broker and result backend, retry-with-backoff as the default task pattern, and the job-status endpoint shown above. The document processing and AI code review use cases in the kit build directly on this pattern — the idempotency and progress-reporting scaffolding is already in place.