Asynchronous Architecture and Queues

Why You Never Block on an LLM Call

An API call to a language model takes between one and thirty seconds, depending on the model and the complexity of the request. Thirty seconds is an eternity for a server process: during that time, the waiting thread does nothing, yet it still occupies a slot. Multiply that by a hundred simultaneous requests, and your application crashes not because it’s computing too much, but because it’s waiting too much.

An asynchronous architecture backed by a queue solves three problems at once. It absorbs traffic spikes, since the queue grows instead of rejecting requests. It centralizes retry logic, because a failed task returns to the queue instead of vanishing with the HTTP request that triggered it. And it decouples your services: the component receiving requests no longer needs to know how long the executing component takes to process them.

The Producer / Consumer Pattern

The basic structure relies on two objects. A task describes what needs to be done and the current processing status; a queue distributes these tasks to a pool of workers that consume them in parallel. Modeling status as an enumeration rather than a free-form string avoids the most common class of bugs in this type of code: two parts of the program writing "terminee" and "terminée".

import asyncio
from dataclasses import dataclass, field
from enum import Enum

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class Task:
    id: str
    prompt: str
    model: str = "gpt-5.6-terra"
    status: TaskStatus = TaskStatus.PENDING
    result: str | None = None
    attempts: int = 0
    max_attempts: int = 3

class ProcessingQueue:
    """Asynchronous queue for LLM calls."""

    def __init__(self, num_workers: int = 5):
        self.queue: asyncio.Queue[Task] = asyncio.Queue()
        self.results: dict[str, Task] = {}
        self.num_workers = num_workers

    async def add(self, task: Task) -> str:
        """Adds a task to the queue."""
        self.results[task.id] = task
        await self.queue.put(task)
        return task.id

    async def worker(self, worker_id: int):
        """Worker that processes tasks from the queue."""
        import openai
        client = openai.AsyncOpenAI()

        while True:
            task = await self.queue.get()
            task.status = TaskStatus.IN_PROGRESS
            task.attempts += 1

            try:
                response = await client.responses.create(
                    model=task.model,
                    input=task.prompt,
                )
                task.result = response.output_text
                task.status = TaskStatus.COMPLETED

            except Exception as e:
                if task.attempts < task.max_attempts:
                    task.status = TaskStatus.PENDING
                    await self.queue.put(task)
                else:
                    task.status = TaskStatus.FAILED
                    task.result = str(e)

            finally:
                self.queue.task_done()

    async def start(self):
        """Starts the workers."""
        workers = [
            asyncio.create_task(self.worker(i))
            for i in range(self.num_workers)
        ]
        return workers

    async def wait_completion(self):
        """Waits until all tasks are processed."""
        await self.queue.join()

Two details in this worker deserve your attention because they appear in every real-world implementation. The attempts counter lives on the task itself, not the worker, allowing any other worker to pick up a requeued task without losing its history. The task_done() call sits inside a finally block to guarantee the queue never hangs indefinitely if an unexpected exception bypasses the except block: without it, a join() that never returns is the symptom, and it is painful to diagnose.

On the caller side, usage is straightforward: start the workers, enqueue tasks, wait, collect results, and stop.

async def process_documents(documents: list[str]):
    queue = ProcessingQueue(num_workers=10)
    workers = await queue.start()

    # Add tasks
    for i, doc in enumerate(documents):
        task = Task(
            id=f"doc-{i}",
            prompt=f"Summarize this document: {doc}",
        )
        await queue.add(task)

    # Wait for completion
    await queue.wait_completion()

    # Collect results
    for task_id, task in queue.results.items():
        print(f"{task_id}: {task.status.value}")

    # Stop workers
    for w in workers:
        w.cancel()

Respecting Quotas Instead of Hitting Them

OpenAI enforces two simultaneous limits: requests per minute and tokens per minute. A pool of ten workers sending everything they can will hit one or the other, receive rate limit errors, retry, and worsen the congestion. It is better to slow down voluntarily upstream than to get rejected downstream.

The limiter below applies a token bucket over a sliding one-minute window. It maintains two histories—timestamps of recent requests and timestamp/token pairs—purges entries that have fallen out of the window, then waits if either limit would be exceeded. The asyncio lock is essential: without it, ten workers would evaluate the same situation simultaneously and all ten would decide there is still room.

import asyncio
import time

class RateLimiter:
    """Rate limiter based on a token bucket."""

    def __init__(self, rpm: int = 500, tpm: int = 200_000):
        self.rpm = rpm
        self.tpm = tpm
        self.recent_requests: list[float] = []
        self.recent_tokens: list[tuple[float, int]] = []
        self.lock = asyncio.Lock()

    async def wait(self, estimated_tokens: int = 1000):
        """Waits if necessary to respect limits."""
        async with self.lock:
            now = time.time()
            window = 60.0

            # Clean up entries outside the window
            self.recent_requests = [
                t for t in self.recent_requests
                if now - t < window
            ]
            self.recent_tokens = [
                (t, n) for t, n in self.recent_tokens
                if now - t < window
            ]

            # Check RPM
            if len(self.recent_requests) >= self.rpm:
                wait_time = self.recent_requests[0] + window - now
                await asyncio.sleep(wait_time)

            # Check TPM
            tokens_used = sum(n for _, n in self.recent_tokens)
            if tokens_used + estimated_tokens > self.tpm:
                wait_time = self.recent_tokens[0][0] + window - now
                await asyncio.sleep(wait_time)

            self.recent_requests.append(time.time())
            self.recent_tokens.append((time.time(), estimated_tokens))

The estimated_tokens parameter deserves a word: you don’t know the actual cost until the response arrives, so you work with an estimate. Make it generous rather than precise. Systematically underestimating amounts to disabling protection exactly when it would be useful.

Returning Control Immediately

For processing that takes several minutes, making an HTTP client wait is no longer viable: proxy, browser, and load balancer timeouts will kick in before you finish. The webhook pattern reverses responsibility. Your endpoint accepts the request, immediately returns a task ID, and notifies the caller when the result exists, without requiring periodic polling.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.post("/api/analyze")
async def start_analysis(
    request: dict,
    background_tasks: BackgroundTasks,
):
    """Starts a background analysis and notifies via webhook."""
    task_id = generate_id()

    background_tasks.add_task(
        run_analysis,
        task_id=task_id,
        document=request["document"],
        webhook_url=request.get("webhook_url"),
    )

    return {"task_id": task_id, "status": "in_progress"}


async def run_analysis(
    task_id: str,
    document: str,
    webhook_url: str | None,
):
    """Runs the analysis and sends the result via webhook."""
    import httpx

    result = await call_llm(document)

    if webhook_url:
        async with httpx.AsyncClient() as http:
            await http.post(webhook_url, json={
                "task_id": task_id,
                "status": "completed",
                "result": result,
            })

Always store the returned ID and associated state somewhere durable. A webhook can get lost, and the caller must retain the ability to fetch the result themselves.

Key Takeaways

• Use asynchronous queues to decouple ingestion from processing
• Implement a rate limiter to respect OpenAI quotas
• Handle retries with exponential backoff in your workers
• For long-running tasks, prefer a webhook pattern over polling