Async job queue pattern for long AI tasks

What goes wrong

A 4-minute agent run inside an HTTP request times out at the load balancer after 60 seconds. The user sees an error and clicks retry, which starts a second run, while the first one is still going in the background. Now two agents are working on the same task, and both may send the same email.

How the pattern works

client ──POST /jobs──> API ──> queue ──> workers ──> model, tools
   │                    │                  │
   │<── 202 + job_id ───┘                  ├── save progress (database)
   │                                       └── emit status events
   └── GET /jobs/{id} or websocket/SSE <── status, result

1. The API validates the request, creates a job with an ID, puts it on the queue, and returns 202 Accepted with the job ID immediately.
2. Workers take jobs from the queue and process them.
3. Workers save progress after each step, so a crashed worker can resume.
4. The client checks status with a status endpoint, or receives progress over WebSockets or server-sent events.
5. When the job finishes, the result is stored and the user is notified.

Parts

• Queue: Redis-based queues (RQ, Celery with Redis), or a cloud queue (SQS, Pub/Sub).
• Workers: scale on queue length, and cap concurrency so you stay within the model provider's rate limits.
• State: persist progress after each step. See agent state and recovery.
• Idempotency: the same request with the same idempotency key returns the existing job instead of creating a second one.
• Progress: status events like "parsing page 120 of 300" keep users informed.
• Timeouts and limits per job: steps, tokens, money and time.

A minimal version

import asyncio
import uuid

jobs: dict[str, dict] = {}
by_key: dict[str, str] = {}
queue: asyncio.Queue = asyncio.Queue()


def submit(payload: dict, idempotency_key: str) -> str:
    if idempotency_key in by_key:
        return by_key[idempotency_key]
    job_id = str(uuid.uuid4())[:8]
    jobs[job_id] = {"status": "queued", "progress": 0, "result": None}
    by_key[idempotency_key] = job_id
    queue.put_nowait((job_id, payload))
    return job_id


async def worker() -> None:
    while True:
        job_id, payload = await queue.get()
        jobs[job_id]["status"] = "running"
        for page in range(1, payload["pages"] + 1):
            await asyncio.sleep(0.01)
            jobs[job_id]["progress"] = page
        jobs[job_id].update(status="done", result=f"summarised {payload['pages']} pages")
        queue.task_done()


async def main() -> None:
    first = submit({"pages": 30}, idempotency_key="report-q3-ravi")
    retry = submit({"pages": 30}, idempotency_key="report-q3-ravi")
    print("same job on retry:", first == retry)
    task = asyncio.create_task(worker())
    await queue.join()
    task.cancel()
    print(first, jobs[first])


asyncio.run(main())

In production, the jobs live in a database and the queue in Redis or a cloud service, so they survive restarts and can be shared by many workers.

Batch APIs

For work that can wait hours, like processing a backlog of documents overnight, many model providers offer batch APIs: you submit a file of requests and get results within a time window, often at a lower price.

Measuring it

• Queue wait time.
• Job success rate.
• Duplicate job rate (should be near zero with idempotency keys).
• Time to resume after a worker crash.

From the floor

In Agent Forge's production design, the back end had a list of workers handling the agent runs, with a cache and a database for state, separate from the API that accepted requests, and the agents' runs could be stopped, replanned or retried with a hint from the user, which only works when the run's state is saved outside the request.

Related

• Agent state and recovery
• Async Python
• Document processing reference architecture
• Latency and streaming

Sources

• Celery documentation
• Batch API, OpenAI documentation
• Message batches, Anthropic documentation