Why event-driven

The last topic ended with a rule: ack the webhook fast, do the slow work async. A message queue is how. Instead of running a slow LLM pipeline inline, the handler drops a message on a queue and returns; a separate worker picks it up and processes it. This buys you four things critical for AI workloads:

• Decoupling — the producer (webhook/API) and consumer (AI worker) run independently; either can restart without losing work.
• Buffering bursts — 10,000 tickets arrive at once? They queue up and drain at a sustainable rate instead of overwhelming you.
• Smoothing rate limits — the queue lets workers pull at a pace your model provider's limits allow (Track 7), turning a spike into a steady flow.
• Resilience — if a worker crashes mid-job, the message isn't lost; it's redelivered.

This is the backbone of bulk AI processing and the reliable version of the N8N-style pipelines you may already run.

Producer / consumer

The pattern: a producer enqueues, a worker dequeues and does the slow work. The producer stays fast; the worker does the heavy lifting.

# PRODUCER — the webhook handler from the last topic: enqueue and return immediately.
@app.post("/webhooks/new-ticket")
async def on_new_ticket(request: Request):
    event = await verified_event(request)            # verify signature (Topic 1)
    queue.put({"id": event["id"], "body": event["data"]["body"]})   # drop a message on the queue…
    return {"queued": True}                           # …and ACK fast — no LLM call inline

# WORKER — a separate process pulling messages and doing the slow AI work.
def worker():
    while True:                                       # long-running consumer loop
        msg = queue.get()                             # block until a message is available
        category = my_classifier(msg["body"])         # the slow LLM call happens HERE, off the request path
        route_ticket(msg["id"], category)
        queue.ack(msg)                                # tell the queue this message is done (remove it)

The producer never waits on the model; the worker processes at its own pace, and you can run many workers to scale throughput (the horizontal scaling from Track 7). Real queues: Redis/RQ, RabbitMQ, AWS SQS, Google Cloud Tasks, Azure Queue Storage — same pattern, different backend.

Retries with exponential backoff

Workers call flaky things — LLM APIs (429s, timeouts), tools, downstream services. A transient failure shouldn't lose the job; retry it. But retrying immediately and forever makes things worse (you hammer a struggling service). Exponential backoff waits longer between each attempt:

import time

def with_retry(fn, max_attempts=5):
    for attempt in range(max_attempts):
        try:
            return fn()                               # try the flaky operation (an LLM/tool call)
        except TransientError:
            if attempt == max_attempts - 1:           # out of attempts?
                raise                                 # give up → let it go to the dead-letter queue
            wait = 2 ** attempt                       # 1, 2, 4, 8, 16s — back off exponentially
            time.sleep(wait + random.uniform(0, 1))   # + jitter so many workers don't retry in lockstep
# Only retry TRANSIENT errors (429, timeout, 5xx). A 400 (bad input) won't fix itself — don't retry it.

The nuance: retry transient failures (rate limits, timeouts, 5xx), not permanent ones (a 400, invalid input) — retrying a permanent error just wastes attempts. Add jitter (randomness) so a fleet of workers doesn't retry in a synchronized thundering herd.

Dead-letter queues

Some messages fail every retry — a malformed payload, a permanently-broken downstream, a poison message. If they just retry forever they block the queue and burn money. A dead-letter queue (DLQ) is where a message goes after exhausting its retries: parked aside for inspection instead of endlessly reprocessed:

def worker():
    while True:
        msg = queue.get()
        try:
            with_retry(lambda: process(msg))          # retry transient failures a few times
            queue.ack(msg)
        except Exception as e:
            dead_letter_queue.put({**msg, "error": str(e)})   # exhausted retries → park it in the DLQ
            queue.ack(msg)                            # remove from the main queue so it stops blocking
            alert("message dead-lettered", msg["id"], e)      # notify a human to inspect

The DLQ turns "silently stuck forever" into "visibly failed, awaiting a human" — you monitor it, diagnose the poison messages, fix the cause, and optionally replay them. Every mature queue system has this concept; it's the integration equivalent of the incident runbook from LLMOps (Track 5).

At-least-once delivery ⇒ idempotency

Here's the property that ties this to the last topic: most queues guarantee at-least-once delivery — a message is delivered one or more times (a worker might crash after processing but before acking, so the queue redelivers). This means duplicate processing is normal, not exceptional. The only safe design is the one you already learned: idempotent workers that dedupe on the message id, so processing a redelivered message twice is harmless.

def worker():
    while True:
        msg = queue.get()
        if already_processed(msg["id"]):             # idempotency (Topic 1): skip a redelivered duplicate
            queue.ack(msg); continue
        process(msg)                                  # exactly-once EFFECT, despite at-least-once DELIVERY
        mark_processed(msg["id"])
        queue.ack(msg)

At-least-once delivery + idempotent processing = effectively exactly-once outcomes — the standard reliable pattern. (True exactly-once delivery is largely a myth; you achieve exactly-once effect with idempotency.)

Pitfalls

• Slow work inline. Processing on the request/webhook path causes timeouts — enqueue and return, process in a worker.
• Retrying permanent errors. Backoff on a 400/invalid input wastes attempts — retry only transient failures.
• No jitter. Synchronized retries create a thundering herd — add randomness.
• No dead-letter queue. Poison messages retry forever, block the queue, and burn money — DLQ them and alert.
• Non-idempotent workers. At-least-once delivery guarantees eventual duplicates — dedupe on the message id or corrupt data.

Recap

• Message queues decouple producers from AI workers, buffer bursts, smooth rate limits, and add resilience — the way to honor "ack fast, work async."
• A producer enqueues and returns; workers dequeue and do the slow LLM work, scaled horizontally.
• Retry transient failures with exponential backoff + jitter; don't retry permanent ones; send exhausted messages to a dead-letter queue and alert.
• Queues are at-least-once, so duplicates are normal — idempotent workers turn at-least-once delivery into exactly-once effect.
• Worked examples map to the exercises: producer/consumer (Ex 1), backoff retry (Ex 2), dead-letter queue (Ex 3), an async pipeline design (Ex 4).

import queue, threading
q = queue.Queue()

# PRODUCER
@app.post("/webhooks/new-ticket")
async def on_new_ticket(request):
    event = await verified_event(request)
    q.put({"id": event["id"], "body": event["data"]["body"]})   # enqueue
    return {"queued": True}                                       # ack fast — no LLM inline

# WORKER (run in a separate process/thread)
def worker():
    while True:
        msg = q.get()
        category = my_classifier(msg["body"])   # slow LLM call, off the request path
        route_ticket(msg["id"], category)
        q.task_done()

threading.Thread(target=worker, daemon=True).start()

import time, random

def with_retry(fn, max_attempts=5):
    for attempt in range(max_attempts):
        try:
            return fn()
        except PermanentError:       # e.g. 400 / invalid input — won't fix itself
            raise                    # fail fast, don't waste retries
        except TransientError:       # e.g. 429 / timeout / 5xx
            if attempt == max_attempts - 1:
                raise
            time.sleep(2 ** attempt + random.uniform(0, 1))   # 1,2,4,8s + jitter

dlq = queue.Queue()

def worker():
    while True:
        msg = q.get()
        try:
            with_retry(lambda: process(msg))
        except Exception as e:
            dlq.put({**msg, "error": str(e)})    # park it aside after retries exhausted
            alert("dead-lettered", msg["id"], e)  # notify a human
        finally:
            q.task_done()                         # ALWAYS remove from main queue so it can't block