Queues and workflows for slow AI steps

Many AI tasks do not need a person watching tokens appear: extracting fields from an uploaded invoice, embedding a 300-page PDF, transcribing a call, scoring an order for fraud, summarising yesterday's tickets. These steps take seconds to minutes, hit provider rate limits and fail now and then with 429 or 5xx. Running them inside an HTTP request ties up connections, makes users wait and turns every provider hiccup into a user-facing error.

The async pattern: the API validates the request, stores the input (large files go to object storage such as S3), records a job row, publishes a message, and returns 202 Accepted with a job id. Workers consume messages, call the model through the AI gateway, store results and update status. The client learns the result by polling, by a webhook, by a push notification or over an SSE or WebSocket channel.

For a single step, a plain queue (SQS, RabbitMQ, a Kafka topic or a Postgres table with SELECT ... FOR UPDATE SKIP LOCKED) is enough. For multi-step jobs with retries per step, long waits, human review or compensation, a durable workflow engine such as Temporal or AWS Step Functions keeps the progress for you, so a crash resumes at the step that failed instead of starting over. The Messaging and events module covers queues and durable workflows in general; this lesson covers what is different when the slow step is a model.

1. 1Accept and return fast: POST /jobs validates input, uploads the file to S3, inserts jobs(id, tenant_id, type, status='queued', input_ref, idempotency_key UNIQUE) and an outbox row in one transaction, and returns 202 with the job id. A repeated request with the same idempotency key returns the same job.
2. 2Publish reliably: a relay or CDC reads the outbox and publishes to the queue, so a job is never saved without its message (or published without being saved).
3. 3Pass references, not payloads: messages carry ids and S3 keys, not 20 MB documents; queues and workflow histories have size limits (around 1 MB per message by default for SQS and Kafka, and Temporal limits payload and history sizes).
4. 4Size concurrency by the scarce resource: an I/O-bound worker that waits 10 s on a model uses almost no CPU, so CPU autoscaling is the wrong signal. If the provider quota is 2M tokens per minute and each job uses 8,000 tokens, the whole fleet can finish at most 250 jobs per minute; more workers just produce 429s faster. Enforce it with a shared token-bucket limiter in the gateway.
5. 5Autoscale on backlog: scale workers on queue depth or consumer lag (for example with KEDA), capped by the token budget and, for Kafka, by the partition count of the topic.
6. 6Separate priorities: interactive-adjacent jobs (a user waiting on 'processing your document') get their own queue and reserved quota; bulk backfills get a low-priority queue or a provider batch API at about half price.
7. 7Retry safely: retry transient errors (429, 5xx, timeouts) with exponential backoff and jitter, respecting Retry-After; do not retry permanent errors (invalid input, content policy refusal). Set the queue's visibility timeout longer than the worst-case processing time, or extend it while working, so the job is not handed to a second worker mid-call.
8. 8Make side effects idempotent: workers record processed job ids or use upserts keyed by job id, so a duplicate delivery does not create two summaries, two emails or two charges.
9. 9Isolate poison messages: after N attempts move the job to a dead-letter queue with the error attached, mark it failed for the user, and provide a replay tool after the fix.
10. 10Validate AI output before saving: parse JSON against a schema, check confidence; low-confidence results go to a human review queue instead of being accepted.
11. 11Use a durable workflow for multi-step flows: for example extract text, then classify, then extract fields, then wait up to 48 hours for human review, then post to the accounting system. Each step is an activity with its own timeout and retry policy; waiting costs nothing; a crash resumes from the last completed step.
12. 12Notify and observe: update status, notify the client, and track queue age (oldest message), job latency percentiles, retries, DLQ size, tokens and cost per job, with trace context carried in message headers.

from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy

with workflow.unsafe.imports_passed_through():
    from activities import extract_text, extract_fields, post_to_ledger

LLM_RETRY = RetryPolicy(
    initial_interval=timedelta(seconds=2),
    backoff_coefficient=2.0,
    maximum_interval=timedelta(minutes=2),
    maximum_attempts=6,
    non_retryable_error_types=["InvalidDocument"],
)

@workflow.defn
class InvoiceWorkflow:
    def __init__(self) -> None:
        self.review = None  # set by a signal from the reviewer UI

    @workflow.signal
    def reviewed(self, fields: dict) -> None:
        self.review = fields

    @workflow.run
    async def run(self, invoice_id: str, s3_key: str) -> str:
        text_key = await workflow.execute_activity(
            extract_text, s3_key, start_to_close_timeout=timedelta(minutes=5))
        result = await workflow.execute_activity(
            extract_fields, text_key, start_to_close_timeout=timedelta(minutes=2),
            retry_policy=LLM_RETRY)
        fields = result["fields"]
        if result["confidence"] < 0.9:
            await workflow.wait_condition(lambda: self.review is not None,
                                          timeout=timedelta(hours=48))
            fields = self.review
        # invoice_id doubles as the idempotency key in the ledger API
        return await workflow.execute_activity(
            post_to_ledger, args=[invoice_id, fields],
            start_to_close_timeout=timedelta(seconds=30))