Queue-based processing for LLM calls

Overview

Of all the reliability patterns, queues are the most transformative for AI services. The reason is the fundamental tension they resolve:

• Users expect a fast response (<3s before they start to doubt)
• LLM calls take 5-30 seconds (typical)
• Platforms like Slack demand acknowledgment in 3 seconds or they resend the event

Without queues, that tension forces you into bad trade-offs: either lax timeouts (bad UX), or empty responses (even worse UX), or systems that flicker under load (worse still). With queues, you decouple reception from processing: your API responds immediately ("received, processing..."), and a background worker does the slow work without timeout pressure.

In this lesson you're going to design the complete flow: how to receive, enqueue, process, and notify the user of the result. It's not just "drop in SQS and you're done" — it's 6-7 design decisions you have to make consciously.

By the end you'll be able to:

• Design the end-to-end flow from reception to notification
• Choose the right queue (Redis, SQS, RabbitMQ, Kafka) based on your case
• Size workers and throughput for your expected load
• Implement user notification (polling, outbound webhook, WebSocket, Slack message API)
• Handle failures: dead letter queues, retries, idempotency

---

The mental model: 5 pieces

User                                               External provider
   │                                                       ▲
   │ 1                                                     │ 4
   ▼                                                       │
┌──────────┐    ┌────────┐    ┌────────┐    ┌────────────┐ │
│   API    │ 2  │  Queue │ 3  │ Worker │ 4  │   LLM      │─┘
│ (fast)   │───▶│        │───▶│ (slow) │───▶│ (5-30s)    │
└──────────┘    └────────┘    └────────┘    └────────────┘
   │                                            │
   │ 5 (immediate ACK)                          │ 6 (result)
   ▼                                            ▼
User     ◀──────────────────────────────────────┘ 7 (notification)

#	Step	Target latency
1	User sends request	—
2	API validates and enqueues	<100ms
5	API responds ACK to the user	<500ms total
3	Worker dequeues	immediate
4	Worker calls LLM	5-30s
6	Worker stores result	<100ms
7	Notifies user	<500ms after step 6

For the user: initial response in <500ms ("processing..."), final result in 5-30s. Tolerable.

---

Choosing the queue

Queue	When to use it	Pros	Cons
Redis Streams / Lists	MVP, dev, low scale	Trivial setup, you already have it for cache	Not durable by default, doesn't handle millions
AWS SQS	AWS-native, up to moderate scale	Managed, durable, dead letter queues	Vendor lock-in, no easy fanout
RabbitMQ	On-prem, fine-grained routing control	Mature, flexible (exchanges, topics)	Operating Erlang/RabbitMQ requires skill
Apache Kafka	High throughput, multi-consumer	Massive throughput, replay, fanout	Overkill for most AI workloads
Google Pub/Sub	GCP-native, fanout	Managed, integrates with the rest of GCP	Vendor lock-in
Cloud Tasks / Lambda + queue	Serverless	You don't operate workers	Cold start, duration limits

For 90% of AI cases: start with Redis Streams (if your team already knows Redis) or SQS (if AWS). Kafka only if you can justify it with throughput >1k msg/s or you need fan-out to multiple independent consumers.

---

Implementation: complete example with FastAPI + Redis

Setup

pip install fastapi redis httpx celery[redis]

The API service

# api.py
import uuid
import time
import json
import redis
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
r = redis.Redis(decode_responses=True)

QUEUE_NAME = "llm-jobs"
RESULTS_HASH = "llm-results"


class ChatRequest(BaseModel):
    prompt: str
    max_tokens: int = 256
    user_id: str
    slack_channel: str | None = None  # for notifying


class ChatAccepted(BaseModel):
    job_id: str
    status: str
    poll_url: str


@app.post("/chat", response_model=ChatAccepted, status_code=202)
def enqueue_chat(req: ChatRequest):
    job_id = str(uuid.uuid4())
    job_payload = {
        "job_id": job_id,
        "prompt": req.prompt,
        "max_tokens": req.max_tokens,
        "user_id": req.user_id,
        "slack_channel": req.slack_channel,
        "enqueued_at": time.time(),
    }
    # Enqueue on a list (FIFO) — alternative: r.xadd for Streams
    r.lpush(QUEUE_NAME, json.dumps(job_payload))
    return ChatAccepted(
        job_id=job_id,
        status="queued",
        poll_url=f"/chat/{job_id}",
    )


@app.get("/chat/{job_id}")
def get_result(job_id: str):
    result = r.hget(RESULTS_HASH, job_id)
    if not result:
        # job not processed yet or doesn't exist
        return {"job_id": job_id, "status": "processing"}
    data = json.loads(result)
    return {"job_id": job_id, **data}

The worker

# worker.py
import json
import time
import redis
from openai import OpenAI

r = redis.Redis(decode_responses=True)
client = OpenAI()

QUEUE_NAME = "llm-jobs"
RESULTS_HASH = "llm-results"
RESULTS_TTL_SECONDS = 3600  # automatic cleanup

def process_job(payload: dict):
    job_id = payload["job_id"]
    try:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": payload["prompt"]}],
            max_tokens=payload["max_tokens"],
            timeout=60,
        )
        result = {
            "status": "completed",
            "text": response.choices[0].message.content,
            "tokens_used": response.usage.total_tokens,
            "completed_at": time.time(),
        }
    except Exception as e:
        result = {
            "status": "failed",
            "error": str(e),
            "completed_at": time.time(),
        }

    # Store result and notify
    r.hset(RESULTS_HASH, job_id, json.dumps(result))
    r.hexpire(RESULTS_HASH, RESULTS_TTL_SECONDS, job_id)  # cleanup

    # If there's a slack_channel, send a message
    if payload.get("slack_channel") and result["status"] == "completed":
        notify_slack(payload["slack_channel"], result["text"])


def notify_slack(channel: str, text: str):
    # implementation with slack_sdk
    pass


def run_worker():
    print("Worker started")
    while True:
        # BRPOP is blocking: waits until there's something, no active polling
        item = r.brpop(QUEUE_NAME, timeout=5)
        if not item:
            continue
        _queue_name, raw_payload = item
        payload = json.loads(raw_payload)
        print(f"Processing job {payload['job_id']}")
        process_job(payload)


if __name__ == "__main__":
    run_worker()

End-to-end flow

# Terminal 1: API
uvicorn api:app --port 8000

# Terminal 2: Worker
python worker.py

# Terminal 3: Client
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"prompt": "Explain REST", "user_id": "u1"}'

# Response (immediate, <500ms):
# {"job_id": "abc-123", "status": "queued", "poll_url": "/chat/abc-123"}

# Poll for result
curl http://localhost:8000/chat/abc-123
# {"job_id": "abc-123", "status": "processing"}  ← while processing
# {"job_id": "abc-123", "status": "completed", "text": "REST is..."}  ← when finished

---

User notification strategies

The worker finished. You have the result. How do you notify the user?

Option 1 — Polling (HTTP GET)

The client does a GET every N seconds until status changes to completed.

Pros: simple, works with any client, doesn't require a persistent connection. Cons: uses your API unnecessarily, latency until the next poll, doesn't scale (10k clients polling every 1s = 10k req/s).

When to use it: few clients, prototypes, MVPs.

Option 2 — WebSocket / SSE (Server-Sent Events)

The client opens a persistent connection. The worker (or a broker) pushes the result to the server; the server sends it over the open connection.

Pros: minimal latency, no polling overhead. Cons: persistent connections are complicated (load balancing, reconnections, idle timeouts).

When to use it: interactive chat, dashboards with live updates.

Option 3 — Outbound webhook

The client registers a URL. When the result is ready, your system POSTs to that URL.

Pros: the client doesn't wait, doesn't poll, doesn't hold a connection. Cons: requires the client to have a public URL, dealing with redelivery if the client doesn't respond.

When to use it: server-to-server integration, callbacks for long jobs.

Option 4 — Platform-native (Slack, Discord)

The worker sends the result to the Slack/Discord channel using the platform's API.

Pros: the user sees the response where they asked for it, no custom client. Cons: you depend on the external API, its rate limits, its auth tokens.

When to use it: your main channel is Slack/Discord (the Capstone's case).

Typical AI combination: WebSocket or SSE for the product's web clients, Slack message for the Slack channel, polling as a universal fallback.

---

Sizing workers and throughput

Simple calculation:

workers_needed = (peak_requests_per_second × average_duration_seconds) / target_utilization

Example: 5 req/s, each 8s, target utilization 70%:

workers = (5 × 8) / 0.7 = 57 workers

That's too many for a single machine. Implications:

• Either horizontal scaling (10 machines × 6 workers each)
• Or reducing the average duration (faster models, lower max_tokens)
• Or batching (one worker processes multiple jobs in parallel)
• Or accepting higher utilization (90% = 44 workers, but queues grow during spikes)

Practical rule: start with workers = peak_rps × duration × 1.5, measure queue length, adjust. If the queue length grows steadily, add workers or reduce the input rate.

---

Idempotency: the problem of repetitions

If your queue redelivers a message (due to a worker timeout, restart, whatever), you don't want to process the LLM twice:

• Double cost
• Potentially inconsistent results
• If it has side effects (sends an email), it's sent twice

Solution: idempotency keys.

def process_job(payload: dict):
    job_id = payload["job_id"]

    # Check if we already processed it
    existing = r.hget(RESULTS_HASH, job_id)
    if existing:
        existing_data = json.loads(existing)
        if existing_data["status"] in ("completed", "failed"):
            return  # already processed, don't reprocess

    # Mark "processing" so another worker doesn't take it
    r.hset(RESULTS_HASH, job_id, json.dumps({"status": "processing"}))

    # ... rest of the processing

For more serious systems, use Redis with SET NX (set if not exists) or distributed locks (Redlock).

---

Dead letter queue (DLQ)

What happens to jobs that fail repeatedly? If you retry infinitely, you fill the main queue with "poison". A DLQ = a separate queue where the jobs that failed N times go.

def process_job_with_dlq(payload: dict):
    attempts = payload.get("attempts", 0)
    MAX_ATTEMPTS = 3

    try:
        # ... process
    except Exception as e:
        attempts += 1
        if attempts >= MAX_ATTEMPTS:
            # Send to DLQ for manual analysis
            r.lpush("llm-jobs-dlq", json.dumps({
                **payload,
                "attempts": attempts,
                "last_error": str(e),
                "moved_to_dlq_at": time.time(),
            }))
        else:
            # Re-enqueue with backoff
            time.sleep(2 ** attempts)
            payload["attempts"] = attempts
            r.lpush(QUEUE_NAME, json.dumps(payload))

There are processes that watch the DLQ (MONITOR llm-jobs-dlq) and alert humans. Those jobs are material for investigation: bug? malicious prompt? degraded provider?

---

Common traps

Trap 1 — "Enqueuing is trivial, the rest doesn't matter." Enqueuing is 10% of the problem. The other 90%: idempotency, DLQ, notification, monitoring queue depth, sizing workers, handling restarts. Don't declare 'done' when you manage to enqueue.

Trap 2 — "The worker processes everything inside a transaction." If the transaction includes the LLM call (8s), your DB has a transaction open for 8 seconds. Bad. Separate them: the LLM call outside the transaction, writing the result inside.

Trap 3 — "The client waits for a synchronous response." You change your API to async but the client keeps waiting. Bad cross-team design. Communicate the change: the client must accept 202 + polling, or use a webhook callback.

Trap 4 — "The worker crashes mid-processing." Your LLM call already started, already spent tokens, but the worker dies before storing the result. Without idempotency, another worker takes the same job and calls the LLM again. Implement a short visibility timeout + idempotency.

Trap 5 — "Monitoring only errors, not queue depth." A queue that grows without limit means you're ingesting faster than you process. If you don't monitor that, you find out when users report "my message is still processing 3 hours later". Alert on queue_length > X and on oldest_message_age > Y.

---

Exercise

Design the queue-based processing flow for this case:

System: a Slack chatbot for internal support. 500 active users, a peak of 30 messages/minute during work hours. Each LLM call takes 8s on average.

Specify:

1. Which queue do you use? Justify it
2. How many workers? Show the calculation
3. How do you notify the user of the result?
4. How do you handle a job that fails 3 times?
5. Which metric do you use to alert if the queue is saturated?

See solution

1. Redis Streams or SQS standard. 30 msg/min is ~0.5 msg/s — very low. You already have Redis if you use cache. SQS if your stack is AWS-native. Kafka is overkill.
2. Peak = 30 msg/min = 0.5 msg/s. Duration 8s. 0.5 × 8 / 0.7 ≈ 6 workers. Round up to 8 workers for margin. A single machine supports them.
3. Slack Web API: the worker, when it finishes, calls chat.postMessage with the result to the channel/user that originated the message. No polling or WebSocket needed — the user sees the response in the same thread.
4. Dead letter queue: after 3 failed attempts, move it to llm-jobs-dlq. Alert on Slack to the #ops channel when a message appears in the DLQ. A human investigates: bug in the bot? malformed prompt? OpenAI outage?
5. Queue depth: alert if LLEN llm-jobs > 50 for more than 2 minutes. It indicates that ingestion exceeds processing. Action: add workers, or investigate whether the LLM provider is degraded.

---

Summary

You learned:

• ✅ The end-to-end flow: fast reception → queue → worker → notification
• ✅ How to choose the queue (Redis to start, SQS for AWS, Kafka only if you can justify it)
• ✅ A working FastAPI + Redis implementation
• ✅ Four notification strategies: polling, WebSocket/SSE, webhook, Slack/Discord
• ✅ Sizing workers with the formula (rps × duration) / utilization
• ✅ Idempotency with keys + Redis SET NX
• ✅ Dead letter queues for jobs that fail repeatedly

Checkpoint: if your API responds 202 in <500ms and your worker processes the LLM in the background, you have the pattern working.

---

Next lesson

04 — Technical and budget rate limiting. Now that your system processes async, comes the question: how many requests do you accept? Without a limit, a client with a bug can flood your queue and drain your budget. You're going to design rate limiting that protects both dimensions.

---

Resources

1. Redis Streams documentation — modern alternative to Lists.
2. AWS SQS — Best practices — for the AWS case.
3. Celery — Python distributed tasks — abstraction over queue + workers.
4. Dramatiq — simpler alternative to Celery.
5. Designing Data-Intensive Applications (Kleppmann) — chapter on message brokers.