Implementing Webhook-Driven LLM Pipelines with Async Job Queues: Handling Long-Running AI Tasks Without Blocking Your Application

Introduction

Picture this: a user uploads a 200-page PDF contract to your legal analysis platform and clicks "Analyze." Your application calls an LLM to extract clauses, flag risks, and generate a summary. This takes 45 seconds. Your HTTP server, dutifully waiting for the LLM to finish, holds the connection open. The user's browser eventually times out. A second user uploads their document. Then a third. Within minutes, your server is juggling dozens of open connections, your thread pool is exhausted, and new requests are being dropped entirely.

This is the most common scaling failure mode for LLM-powered applications, and it has nothing to do with the quality of your prompts or your choice of model. It's an architectural problem. Synchronous request-response patterns work beautifully for fast operations, but LLMs — especially when processing large documents, chaining multiple calls, or generating long outputs — are fundamentally long-running tasks. They need an architectural pattern designed for long-running tasks: async job queues with webhook callbacks.

By the end of this lesson, you'll know how to architect and implement this pattern from scratch. We'll build a complete pipeline where API requests immediately return a job ID, a worker process handles the actual LLM work in the background, and a webhook delivers the result to your caller when the job completes. You'll have a production-ready foundation you can drop into real projects.

What you'll learn:

• Why synchronous LLM calls break under load and when async is the right solution
• How to design a job queue system with Redis and Celery for LLM workloads
• How to implement webhook delivery with retry logic and signature verification
• How to build job status tracking so clients can poll or receive push notifications
• How to handle failure modes: LLM timeouts, webhook delivery failures, and dead-letter queues

---

Prerequisites

You should be comfortable calling LLM APIs from Python — if you need a refresher, see Using the OpenAI API with Python. You should understand basic async concepts in Python and have passing familiarity with Flask or FastAPI. Redis and Celery don't need to be things you've used before; we'll cover what you need.

You'll need: Python 3.10+, Redis running locally (or a free Redis Cloud instance), and an OpenAI or Anthropic API key.

---

Why Synchronous LLM Calls Break Production Systems

Before we build the solution, let's be precise about the problem. HTTP has a request-response model with timeouts. Most reverse proxies (nginx, Cloudflare, AWS ALB) have default timeouts between 30 and 60 seconds. Many mobile clients are shorter. When your LLM call takes 90 seconds to process a long document, you're in timeout territory before the response is even ready.

There's also a concurrency problem. Every open HTTP connection in a traditional web framework ties up a worker. With 20 gunicorn workers and 50 concurrent users running LLM jobs, you've saturated your worker pool. New requests queue up and eventually fail with 503 errors. You've built a bottleneck.

Compare this to the async pattern:

SYNCHRONOUS (broken under load):
Client → POST /analyze → [server waits 90s] → response
         ↑ 
         Server holds connection, blocks worker

ASYNC (what we're building):
Client → POST /analyze → immediate response: {"job_id": "abc123"}
                                   ↓
                          Worker picks up job
                          Worker calls LLM (90s)
                          Worker POSTs to client webhook
Client ← POST /webhook ← {"job_id": "abc123", "result": {...}}

The async pattern has an immediate return that's fast and reliable, a worker that can scale independently, and a webhook delivery that pushes results to the caller without requiring them to maintain a connection.

---

System Architecture Overview

Our system has four components:

1. API Server — Receives requests, validates input, creates jobs, returns immediately
2. Redis — Acts as the message broker (job queue) and job state store
3. Worker Process — Picks up jobs from the queue, executes the LLM pipeline, stores results
4. Webhook Dispatcher — Sends HTTP POST callbacks to the caller when jobs complete

Here's the data flow in detail:

┌─────────────┐    POST /jobs     ┌─────────────────┐
│   Client    │ ─────────────── → │   API Server    │
│             │ ← job_id ──────── │   (FastAPI)     │
└─────────────┘                   └────────┬────────┘
      │                                    │ enqueue
      │                                    ↓
      │                            ┌──────────────┐
      │                            │    Redis     │
      │                            │  (queue +    │
      │                            │  job store)  │
      │                            └──────┬───────┘
      │                                   │ dequeue
      │                                   ↓
      │                           ┌───────────────┐
      │                           │    Worker     │
      │                           │   (Celery)    │
      │                           │  calls LLM,  │
      │                           │  stores result│
      │                           └───────┬───────┘
      │                                   │
      │    POST /webhook (result)         │
      └ ← ────────────────────────────────┘

Let's build each piece.

---

Setting Up the Project

Start by creating the project structure and installing dependencies:

mkdir llm-webhook-pipeline && cd llm-webhook-pipeline
python -m venv venv && source venv/bin/activate

pip install fastapi uvicorn celery redis httpx openai python-dotenv pydantic

Project layout:

llm-webhook-pipeline/
├── app/
│   ├── __init__.py
│   ├── main.py          # FastAPI server
│   ├── models.py        # Pydantic schemas
│   ├── tasks.py         # Celery tasks (LLM work)
│   ├── worker.py        # Celery app initialization
│   ├── job_store.py     # Redis job state management
│   └── webhook.py       # Webhook delivery logic
├── .env
└── requirements.txt

Your .env file:

OPENAI_API_KEY=sk-...
REDIS_URL=redis://localhost:6379/0
WEBHOOK_SECRET=your-secret-for-signing-webhook-payloads

---

Building the Job State Store

The job store is the backbone of the system. It tracks every job through its lifecycle: pending → running → completed | failed. We need clients to be able to query job status at any time, and we need workers to be able to write results.

# app/job_store.py
import json
import time
import uuid
from enum import Enum
from typing import Any, Optional
import redis
from dotenv import load_dotenv
import os

load_dotenv()

redis_client = redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379/0"))

class JobStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

JOB_TTL_SECONDS = 86400  # Keep job records for 24 hours

def create_job(
    job_type: str,
    payload: dict,
    webhook_url: str,
    webhook_secret: Optional[str] = None,
) -> str:
    """Create a new job record and return the job_id."""
    job_id = str(uuid.uuid4())
    job_data = {
        "job_id": job_id,
        "job_type": job_type,
        "status": JobStatus.PENDING,
        "payload": payload,
        "webhook_url": webhook_url,
        "webhook_secret": webhook_secret,
        "created_at": time.time(),
        "updated_at": time.time(),
        "result": None,
        "error": None,
        "attempts": 0,
    }
    redis_client.setex(
        f"job:{job_id}",
        JOB_TTL_SECONDS,
        json.dumps(job_data)
    )
    return job_id

def get_job(job_id: str) -> Optional[dict]:
    """Retrieve job data by ID."""
    raw = redis_client.get(f"job:{job_id}")
    if not raw:
        return None
    return json.loads(raw)

def update_job_status(
    job_id: str,
    status: JobStatus,
    result: Any = None,
    error: str = None,
) -> None:
    """Update job status, result, and error in Redis."""
    job = get_job(job_id)
    if not job:
        raise ValueError(f"Job {job_id} not found")
    
    job["status"] = status
    job["updated_at"] = time.time()
    job["attempts"] = job.get("attempts", 0) + 1
    
    if result is not None:
        job["result"] = result
    if error is not None:
        job["error"] = error
    
    redis_client.setex(
        f"job:{job_id}",
        JOB_TTL_SECONDS,
        json.dumps(job)
    )

The TTL on job records is important — Redis is not a durable database, and you don't want stale job records accumulating forever. 24 hours is usually sufficient for clients to retrieve results.

---

Building the API Server

The API server's job is simple: accept requests, validate them, create a job record, enqueue the work, and return the job ID immediately. It should never wait for the LLM.

# app/models.py
from pydantic import BaseModel, HttpUrl
from typing import Optional, Dict, Any

class DocumentAnalysisRequest(BaseModel):
    document_text: str
    analysis_type: str  # e.g., "contract_review", "sentiment", "summarization"
    webhook_url: HttpUrl
    webhook_secret: Optional[str] = None
    metadata: Optional[Dict[str, Any]] = None

class JobCreatedResponse(BaseModel):
    job_id: str
    status: str
    message: str

class JobStatusResponse(BaseModel):
    job_id: str
    status: str
    created_at: float
    updated_at: float
    result: Optional[Any] = None
    error: Optional[str] = None

# app/main.py
from fastapi import FastAPI, HTTPException, Header
from fastapi.responses import JSONResponse
import hmac
import hashlib
import os

from app.models import DocumentAnalysisRequest, JobCreatedResponse, JobStatusResponse
from app.job_store import create_job, get_job, JobStatus
from app.tasks import run_document_analysis

app = FastAPI(title="LLM Webhook Pipeline")

@app.post("/jobs/document-analysis", response_model=JobCreatedResponse)
async def submit_document_analysis(request: DocumentAnalysisRequest):
    """
    Accept a document analysis job. Returns immediately with a job_id.
    Results will be POSTed to webhook_url when processing completes.
    """
    # Validate document isn't empty
    if not request.document_text.strip():
        raise HTTPException(status_code=422, detail="document_text cannot be empty")
    
    if len(request.document_text) > 500_000:  # ~125k tokens, adjust for your use case
        raise HTTPException(status_code=422, detail="Document too large")
    
    # Create the job record in Redis
    job_id = create_job(
        job_type="document_analysis",
        payload={
            "document_text": request.document_text,
            "analysis_type": request.analysis_type,
            "metadata": request.metadata or {},
        },
        webhook_url=str(request.webhook_url),
        webhook_secret=request.webhook_secret,
    )
    
    # Enqueue the Celery task — this is non-blocking
    run_document_analysis.delay(job_id)
    
    return JobCreatedResponse(
        job_id=job_id,
        status=JobStatus.PENDING,
        message=f"Job accepted. Results will be delivered to {request.webhook_url}",
    )

@app.get("/jobs/{job_id}", response_model=JobStatusResponse)
async def get_job_status(job_id: str):
    """
    Poll for job status. Clients can use this as a fallback if webhook delivery fails.
    """
    job = get_job(job_id)
    if not job:
        raise HTTPException(status_code=404, detail=f"Job {job_id} not found")
    
    return JobStatusResponse(
        job_id=job["job_id"],
        status=job["status"],
        created_at=job["created_at"],
        updated_at=job["updated_at"],
        result=job.get("result"),
        error=job.get("error"),
    )

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

Notice the /jobs/{job_id} endpoint. This is your fallback for clients whose webhook receiver was temporarily unavailable. Smart clients should implement polling as a backstop, even when they have a webhook configured.

---

Building the Celery Worker

This is where the actual LLM work happens. The worker is a separate process that reads from the Redis queue and executes tasks.

# app/worker.py
from celery import Celery
import os
from dotenv import load_dotenv

load_dotenv()

celery_app = Celery(
    "llm_pipeline",
    broker=os.getenv("REDIS_URL", "redis://localhost:6379/0"),
    backend=os.getenv("REDIS_URL", "redis://localhost:6379/0"),
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    task_acks_late=True,          # Only ack after task completes, not when received
    worker_prefetch_multiplier=1, # Don't prefetch; LLM tasks are long and variable
    task_track_started=True,
    task_time_limit=600,          # Hard kill after 10 minutes
    task_soft_time_limit=540,     # Soft signal at 9 minutes (triggers SoftTimeLimitExceeded)
)

# app/tasks.py
import time
from celery.exceptions import SoftTimeLimitExceeded
from openai import OpenAI

from app.worker import celery_app
from app.job_store import get_job, update_job_status, JobStatus
from app.webhook import deliver_webhook

client = OpenAI()

ANALYSIS_PROMPTS = {
    "contract_review": """You are a legal analyst. Review the following contract and provide:
1. A list of key obligations for each party
2. Notable risk clauses (limitation of liability, indemnification, termination rights)
3. Any unusual or potentially problematic terms
4. A plain-language summary

Format your response as JSON with keys: obligations, risk_clauses, unusual_terms, summary""",

    "summarization": """Provide a structured summary of the following document.
Include: main_topic, key_points (list), conclusions, and word_count_estimate.
Format as JSON.""",

    "sentiment": """Analyze the sentiment and tone of the following text.
Provide: overall_sentiment (positive/negative/neutral/mixed), 
confidence (0-1), key_phrases, and a brief explanation.
Format as JSON.""",
}

@celery_app.task(
    bind=True,
    name="run_document_analysis",
    max_retries=3,
    default_retry_delay=30,
)
def run_document_analysis(self, job_id: str):
    """
    Main Celery task: fetches job data, runs LLM analysis, stores result,
    and triggers webhook delivery.
    """
    job = get_job(job_id)
    if not job:
        # Job was TTL'd or never existed; nothing to do
        return

    update_job_status(job_id, JobStatus.RUNNING)

    try:
        payload = job["payload"]
        analysis_type = payload["analysis_type"]
        document_text = payload["document_text"]

        system_prompt = ANALYSIS_PROMPTS.get(
            analysis_type,
            "Analyze the following document and provide a structured response as JSON."
        )

        # Call the LLM — this is the slow part
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": document_text},
            ],
            response_format={"type": "json_object"},
            temperature=0.1,
        )

        result = {
            "analysis_type": analysis_type,
            "content": response.choices[0].message.content,
            "model": response.model,
            "usage": {
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
                "total_tokens": response.usage.total_tokens,
            },
            "metadata": payload.get("metadata", {}),
        }

        # Store result and mark complete
        update_job_status(job_id, JobStatus.COMPLETED, result=result)

        # Deliver the webhook — this handles its own retries
        deliver_webhook(
            webhook_url=job["webhook_url"],
            job_id=job_id,
            status=JobStatus.COMPLETED,
            result=result,
            webhook_secret=job.get("webhook_secret"),
        )

    except SoftTimeLimitExceeded:
        error_msg = f"Task exceeded time limit for job {job_id}"
        update_job_status(job_id, JobStatus.FAILED, error=error_msg)
        deliver_webhook(
            webhook_url=job["webhook_url"],
            job_id=job_id,
            status=JobStatus.FAILED,
            result=None,
            error=error_msg,
            webhook_secret=job.get("webhook_secret"),
        )

    except Exception as exc:
        error_msg = str(exc)
        
        # Retry if we haven't hit max_retries
        if self.request.retries < self.max_retries:
            update_job_status(job_id, JobStatus.PENDING)  # Back to pending for retry
            raise self.retry(exc=exc, countdown=30 * (self.request.retries + 1))
        
        # Final failure
        update_job_status(job_id, JobStatus.FAILED, error=error_msg)
        deliver_webhook(
            webhook_url=job["webhook_url"],
            job_id=job_id,
            status=JobStatus.FAILED,
            result=None,
            error=error_msg,
            webhook_secret=job.get("webhook_secret"),
        )

The retry logic uses exponential backoff: 30s, 60s, 90s. This handles transient LLM API errors gracefully. For deeper context on the error types you'll encounter, see Handling LLM Errors, Timeouts, and Rate Limits in Python Applications.

---

Building the Webhook Dispatcher

Webhook delivery sounds simple — just POST the result to a URL. But production webhook delivery needs retry logic, timeout handling, and signature verification so the receiver can trust the payload came from you.

# app/webhook.py
import hashlib
import hmac
import json
import logging
import time
from typing import Any, Optional

import httpx

logger = logging.getLogger(__name__)

WEBHOOK_TIMEOUT_SECONDS = 10
MAX_DELIVERY_ATTEMPTS = 5
RETRY_DELAYS = [0, 5, 30, 120, 300]  # Immediate, then 5s, 30s, 2m, 5m

def sign_payload(payload_bytes: bytes, secret: str) -> str:
    """
    Generate HMAC-SHA256 signature for webhook payload.
    Follows the same pattern as Stripe, GitHub, etc.
    """
    return hmac.new(
        key=secret.encode("utf-8"),
        msg=payload_bytes,
        digestmod=hashlib.sha256,
    ).hexdigest()

def deliver_webhook(
    webhook_url: str,
    job_id: str,
    status: str,
    result: Any = None,
    error: Optional[str] = None,
    webhook_secret: Optional[str] = None,
) -> bool:
    """
    Deliver a webhook payload to webhook_url with retries.
    Returns True if delivery succeeded, False if all attempts failed.
    """
    payload = {
        "job_id": job_id,
        "status": status,
        "timestamp": time.time(),
        "result": result,
        "error": error,
    }
    payload_bytes = json.dumps(payload).encode("utf-8")

    headers = {
        "Content-Type": "application/json",
        "User-Agent": "LLM-Webhook-Pipeline/1.0",
        "X-Job-ID": job_id,
    }

    if webhook_secret:
        signature = sign_payload(payload_bytes, webhook_secret)
        headers["X-Webhook-Signature"] = f"sha256={signature}"

    for attempt, delay in enumerate(RETRY_DELAYS):
        if delay > 0:
            time.sleep(delay)

        try:
            with httpx.Client(timeout=WEBHOOK_TIMEOUT_SECONDS) as http:
                response = http.post(
                    webhook_url,
                    content=payload_bytes,
                    headers=headers,
                )

            if 200 <= response.status_code < 300:
                logger.info(
                    f"Webhook delivered for job {job_id} "
                    f"(attempt {attempt + 1}, status {response.status_code})"
                )
                return True

            logger.warning(
                f"Webhook delivery failed for job {job_id}: "
                f"HTTP {response.status_code} (attempt {attempt + 1})"
            )

        except httpx.TimeoutException:
            logger.warning(
                f"Webhook timeout for job {job_id} (attempt {attempt + 1})"
            )
        except httpx.RequestError as exc:
            logger.warning(
                f"Webhook request error for job {job_id}: {exc} "
                f"(attempt {attempt + 1})"
            )

    logger.error(
        f"Webhook delivery permanently failed for job {job_id} "
        f"after {MAX_DELIVERY_ATTEMPTS} attempts. "
        f"Client can poll GET /jobs/{job_id} to retrieve result."
    )
    return False

Verifying Webhooks on the Receiver Side

Your clients need to verify that webhook payloads are genuine. Here's a reference implementation they can use:

# Example receiver (your client's endpoint)
from fastapi import FastAPI, Request, HTTPException, Header
import hmac
import hashlib
import json

receiver_app = FastAPI()

WEBHOOK_SECRET = "your-shared-secret"

@receiver_app.post("/webhooks/llm-results")
async def receive_webhook(
    request: Request,
    x_webhook_signature: Optional[str] = Header(None),
):
    body = await request.body()
    
    # Verify signature if secret is configured
    if WEBHOOK_SECRET and x_webhook_signature:
        expected_sig = "sha256=" + hmac.new(
            key=WEBHOOK_SECRET.encode("utf-8"),
            msg=body,
            digestmod=hashlib.sha256,
        ).hexdigest()
        
        # Use hmac.compare_digest to prevent timing attacks
        if not hmac.compare_digest(expected_sig, x_webhook_signature):
            raise HTTPException(status_code=401, detail="Invalid webhook signature")
    
    payload = json.loads(body)
    job_id = payload["job_id"]
    status = payload["status"]
    
    if status == "completed":
        result = payload["result"]
        # Process result: save to database, notify user, etc.
        print(f"Job {job_id} completed. Tokens used: {result['usage']['total_tokens']}")
    
    elif status == "failed":
        error = payload["error"]
        print(f"Job {job_id} failed: {error}")
    
    # Always return 200 quickly — do heavy processing asynchronously
    return {"received": True}

The critical line is hmac.compare_digest instead of ==. String equality short-circuits on the first mismatched character, which leaks timing information. compare_digest runs in constant time regardless of where the mismatch occurs.

---

Running the System

You need three processes running simultaneously:

# Terminal 1: Start Redis (if running locally)
redis-server

# Terminal 2: Start the FastAPI server
uvicorn app.main:app --reload --port 8000

# Terminal 3: Start the Celery worker
celery -A app.worker.celery_app worker --loglevel=info --concurrency=4

For --concurrency=4, you're running 4 worker threads. Each thread can be blocked in an LLM call without affecting the others. Scale this based on your I/O budget and API rate limits, not CPU — LLM calls are almost entirely I/O wait.

Test it end-to-end:

curl -X POST http://localhost:8000/jobs/document-analysis \
  -H "Content-Type: application/json" \
  -d '{
    "document_text": "This Software License Agreement is entered into by Acme Corp (Licensor) and Beta LLC (Licensee). Licensee shall pay a monthly fee of $5,000. Licensor may terminate this agreement with 30 days notice for any material breach. Liability is limited to fees paid in the preceding 3 months.",
    "analysis_type": "contract_review",
    "webhook_url": "https://webhook.site/your-unique-id",
    "webhook_secret": "test-secret-123"
  }'

You'll get back something like:

{
  "job_id": "7f3a9c12-4d8e-4b2f-a1c6-9e8f2a3b4c5d",
  "status": "pending",
  "message": "Job accepted. Results will be delivered to https://webhook.site/your-unique-id"
}

Use webhook.site for testing — it gives you a public URL that captures incoming webhooks so you can inspect them.

---

Handling Multi-Step LLM Pipelines

Real-world document analysis often requires chaining multiple LLM calls. For example: first extract the document structure, then analyze each section independently, then synthesize a final report. This is where the queue architecture really shines — you can chain tasks.

# app/tasks.py (extended for multi-step pipeline)

@celery_app.task(
    bind=True,
    name="run_pipeline_extract_structure",
    max_retries=3,
)
def run_pipeline_extract_structure(self, job_id: str):
    """Step 1: Extract document structure."""
    job = get_job(job_id)
    update_job_status(job_id, JobStatus.RUNNING)

    document_text = job["payload"]["document_text"]

    response = client.chat.completions.create(
        model="gpt-4o-mini",  # Use a cheaper model for structure extraction
        messages=[
            {
                "role": "system",
                "content": "Extract the sections of this document. Return JSON with a 'sections' array, each with 'title' and 'content'.",
            },
            {"role": "user", "content": document_text},
        ],
        response_format={"type": "json_object"},
    )

    import json
    structure = json.loads(response.choices[0].message.content)

    # Store intermediate result and chain to next step
    update_job_status(
        job_id,
        JobStatus.RUNNING,  # Still running — pipeline continues
        result={"step": "structure_extracted", "structure": structure},
    )

    # Chain to next task
    run_pipeline_analyze_sections.delay(job_id, structure["sections"])


@celery_app.task(
    bind=True,
    name="run_pipeline_analyze_sections",
    max_retries=3,
)
def run_pipeline_analyze_sections(self, job_id: str, sections: list):
    """Step 2: Analyze each section (parallel fan-out)."""
    from celery import group

    job = get_job(job_id)

    # Create a group of parallel tasks — one per section
    section_tasks = group(
        analyze_single_section.s(job_id, i, section)
        for i, section in enumerate(sections[:10])  # Cap at 10 sections
    )

    # Run all section analyses in parallel, then synthesize
    pipeline = section_tasks | synthesize_results.s(job_id)
    pipeline.delay()

---

Hands-On Exercise

Build a Research Paper Summarizer webhook pipeline with three stages:

1. Stage 1 — Extract Metadata: Extract title, authors, abstract, and section headers using gpt-4o-mini
2. Stage 2 — Summarize Sections: For each major section (Introduction, Methods, Results, Discussion), generate a 2-3 sentence summary in parallel
3. Stage 3 — Synthesize: Combine section summaries into a single 300-word accessible summary with key findings and limitations

Requirements:

• The API should accept a raw PDF text string and a webhook URL
• Each stage should update the job status with intermediate progress (e.g., {"stage": "extracting_metadata", "progress": 0.1})
• The final webhook payload should include the full synthesis plus per-section summaries and token usage across all stages
• Implement a dead letter queue: if webhook delivery fails all retries, write the job ID to a Redis list called failed_webhook_deliveries so an operator can inspect and retry manually

Bonus: Add a GET /jobs/{job_id}/retry-webhook endpoint that re-triggers webhook delivery for a completed job. This is invaluable when a client's webhook receiver was temporarily down.

---

Common Mistakes & Troubleshooting

Mistake 1: Doing Real Work in the API Handler

# WRONG: This blocks the API worker
@app.post("/jobs/document-analysis")
async def submit_job(request: DocumentAnalysisRequest):
    result = call_openai(request.document_text)  # ← blocks here for 60s
    return result

# RIGHT: Enqueue immediately, return job_id
@app.post("/jobs/document-analysis")
async def submit_job(request: DocumentAnalysisRequest):
    job_id = create_job(...)
    run_document_analysis.delay(job_id)  # Non-blocking
    return {"job_id": job_id}

Mistake 2: Not Handling Webhook Receiver Timeouts

Your webhook receiver should return 200 immediately and process asynchronously. If you run slow database writes or additional API calls inside the webhook handler before returning, you'll hit your own timeout and cause the webhook dispatcher to retry — resulting in duplicate processing.

# WRONG: Slow work before returning 200
@receiver_app.post("/webhooks/llm-results")
async def receive_webhook(request: Request):
    payload = await request.json()
    save_to_slow_database(payload)        # ← takes 3 seconds
    send_email_notification(payload)      # ← takes 5 seconds
    return {"received": True}             # ← dispatcher already timed out at 10s

# RIGHT: Acknowledge immediately, process in background
@receiver_app.post("/webhooks/llm-results")
async def receive_webhook(request: Request, background_tasks: BackgroundTasks):
    payload = await request.json()
    background_tasks.add_task(process_result, payload)  # Non-blocking
    return {"received": True}   # ← Returns in milliseconds

Mistake 3: Not Deduplicating Webhook Deliveries

Webhook dispatchers can deliver the same event multiple times (network hiccup + successful delivery both happened). Your receiver must be idempotent — processing the same job_id twice should produce the same result as processing it once.

# Use job_id as an idempotency key
async def process_result(payload: dict):
    job_id = payload["job_id"]
    
    # Check if we've already processed this job
    if await db.job_result_exists(job_id):
        logger.info(f"Duplicate webhook for job {job_id}, skipping")
        return
    
    # Process and mark as handled
    await db.save_job_result(job_id, payload["result"])

Mistake 4: Using Default Celery Prefetch Settings

By default, Celery workers prefetch 4x their concurrency count. With 4 workers, that's 16 tasks sitting reserved in memory. If your worker crashes, those 16 tasks are lost. Always set worker_prefetch_multiplier=1 for LLM workloads.

Mistake 5: Forgetting Job Cleanup

Redis isn't free, and job records accumulate. Always set a TTL on job keys. Also consider archiving completed jobs to a persistent database (Postgres, DynamoDB) before they expire if you need long-term audit trails. See Implementing LLM Observability: Tracing, Logging, and Monitoring Requests in Production for patterns on capturing job history for analysis.

Troubleshooting: Worker Not Picking Up Tasks

If you submit a job and nothing happens in the worker:

# Check that Celery can see the broker
celery -A app.worker.celery_app inspect ping

# Check the queue length in Redis
redis-cli llen celery

# Check for active workers
celery -A app.worker.celery_app inspect active

# See what's in the failed tasks
celery -A app.worker.celery_app inspect reserved

If redis-cli llen celery shows tasks accumulating but workers aren't processing them, check that the task name in @celery_app.task(name="...") exactly matches the name you're calling in .delay().

Troubleshooting: Webhook Delivery Failing with 401

If the receiver is returning 401, the signature isn't matching. Most common causes:

1. The webhook_secret in the job creation request doesn't match what the receiver is using to verify
2. The payload was modified in transit (unlikely but check for encoding issues)
3. You're signing the serialized string but the receiver is re-serializing with different key ordering

Fix option 3 by using sort_keys=True in json.dumps() on both sides to guarantee consistent serialization.

---

Scaling Beyond a Single Worker

For production, you'll want multiple worker instances and monitoring:

# docker-compose.yml (production-oriented excerpt)
version: "3.8"

services:
  api:
    build: .
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4
    environment:
      - REDIS_URL=redis://redis:6379/0
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    ports:
      - "8000:8000"

  worker:
    build: .
    command: celery -A app.worker.celery_app worker --loglevel=info --concurrency=8
    environment:
      - REDIS_URL=redis://redis:6379/0
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    deploy:
      replicas: 3  # 3 worker instances × 8 concurrency = 24 parallel LLM calls

  flower:
    build: .
    command: celery -A app.worker.celery_app flower --port=5555
    ports:
      - "5555:5555"  # Celery monitoring UI

  redis:
    image: redis:7-alpine
    command: redis-server --maxmemory 2gb --maxmemory-policy allkeys-lru

Flower is Celery's built-in monitoring dashboard. It shows active tasks, task history, worker status, and failure rates. Run it in production — it'll save you hours of debugging.

For deeper observability into what's happening inside each LLM call, consider the tracing patterns in Implementing LLM Observability: Tracing, Logging, and Monitoring Requests in Production.

---

When to Use This Pattern vs. Alternatives

Scenario	Best Approach
LLM response < 10s, real-time UX needed	Streaming responses
LLM response 10–30s, acceptable to wait	Async endpoint with client polling
LLM response > 30s or unpredictable duration	Webhook + job queue (this article)
Batch processing thousands of documents overnight	Webhook + job queue + priority queues
Complex multi-agent pipelines with human-in-the-loop	Webhook + job queue + agentic loop patterns

The webhook pattern adds operational complexity — you're now running Redis, Celery workers, and managing webhook retry state. It's worth it when tasks are genuinely long-running, but don't reach for it by default.

One excellent use case for this architecture is building a production document Q&A system where document ingestion (chunking, embedding, indexing) happens asynchronously after upload, and users receive a notification when their documents are ready to query.

---

Summary & Next Steps

You've built a complete webhook-driven LLM pipeline: an API that accepts jobs and returns immediately, a Redis-backed job store that tracks every state transition, a Celery worker that executes LLM calls with retry logic and timeout handling, and a webhook dispatcher with HMAC signature verification and exponential backoff retries.

The key architectural insights:

• Never block your API workers on LLM calls — enqueue and return a job ID immediately
• State lives in Redis — workers and the API server are stateless; the job record is the source of truth
• Webhooks need redundancy — always expose a polling endpoint as a fallback; clients can't always guarantee their receiver is up
• Idempotency is non-negotiable — webhook receivers will see duplicate deliveries; design for it from day one
• Tune your Celery settings for long I/O-bound tasks: task_acks_late=True, worker_prefetch_multiplier=1

Where to go next:

• Add structured output guarantees to your LLM calls so the worker can always parse the result reliably
• Explore cost optimization strategies — async job queues pair naturally with batching and caching because jobs don't need to be processed immediately
• Add observability to trace exactly how long each stage of your pipeline takes so you can identify bottlenecks
• If you're serving multiple customers, look at building a multi-tenant LLM platform — job queues with per-tenant priority lanes are a natural extension of this architecture