Async LLM pipelines with queues and workers

An LLM API call takes anywhere from one to thirty seconds depending on the model and output length. If you handle these calls synchronously in a web server, you are blocking a thread for that entire duration. At ten concurrent users that is manageable. At a hundred it becomes a problem. At a thousand you need a fundamentally different architecture.

The answer is the same pattern that solved this problem for email sending, image processing, and any other slow IO operation: decouple the request from the work using a queue and a pool of workers.

Why queues specifically

A queue gives you three properties that matter for LLM workloads:

Decoupling. The request handler acknowledges receipt immediately and returns. The work happens separately. The user gets a fast response ("your analysis is processing") instead of waiting.

Backpressure. When the LLM API is at its rate limit, tasks pile up in the queue rather than failing or timing out. Workers drain the queue as capacity becomes available.

Durability. If a worker crashes mid-task, the message remains in the queue and another worker picks it up. Without a queue, in-flight tasks vanish on restart.

Architecture overview

A minimal async LLM pipeline has four components:

1. API layer — accepts requests, validates input, enqueues tasks, returns a job ID
2. Queue — holds pending tasks and delivers them to workers
3. Workers — consume tasks, call the LLM, store results
4. Result store — holds completed results until the client polls for them

Client → POST /analyse → API layer → Queue → Workers → Result store
Client → GET /result/{id} ← API layer ← Result store

The client submits work, gets a job ID, then polls (or uses a webhook) to retrieve the result when it is ready.

Implementation with BullMQ and Redis

BullMQ is a Node.js queue library backed by Redis. It handles retries, concurrency, rate limiting, and job state tracking out of the box.

// queue.ts
import { Queue, Worker, Job } from "bullmq";
import Anthropic from "@anthropic-ai/sdk";
import { Redis } from "ioredis";

const connection = new Redis({
  host: process.env.REDIS_HOST || "localhost",
  port: 6379,
  maxRetriesPerRequest: null, // Required by BullMQ
});

export interface LLMTask {
  jobId: string;
  prompt: string;
  systemPrompt?: string;
  model?: string;
  maxTokens?: number;
  metadata?: Record<string, unknown>;
}

export interface LLMResult {
  jobId: string;
  output: string;
  inputTokens: number;
  outputTokens: number;
  completedAt: string;
}

// Create the queue
export const llmQueue = new Queue<LLMTask>("llm-tasks", {
  connection,
  defaultJobOptions: {
    attempts: 3,
    backoff: {
      type: "exponential",
      delay: 2000, // Start at 2s, then 4s, then 8s
    },
    removeOnComplete: false, // Keep for result retrieval
    removeOnFail: false,
  },
});

// worker.ts
import { Worker, Job } from "bullmq";
import Anthropic from "@anthropic-ai/sdk";
import { LLMTask, LLMResult } from "./queue";

const client = new Anthropic();

async function processLLMTask(job: Job<LLMTask>): Promise<LLMResult> {
  const { jobId, prompt, systemPrompt, model, maxTokens } = job.data;

  const response = await client.messages.create({
    model: model || "claude-haiku-4-5",
    max_tokens: maxTokens || 1024,
    system: systemPrompt,
    messages: [{ role: "user", content: prompt }],
  });

  return {
    jobId,
    output: response.content[0].type === "text" ? response.content[0].text : "",
    inputTokens: response.usage.input_tokens,
    outputTokens: response.usage.output_tokens,
    completedAt: new Date().toISOString(),
  };
}

export function startWorkers(concurrency = 3) {
  const worker = new Worker<LLMTask, LLMResult>(
    "llm-tasks",
    processLLMTask,
    {
      connection: {
        host: process.env.REDIS_HOST || "localhost",
        port: 6379,
        maxRetriesPerRequest: null,
      },
      concurrency, // Process N jobs simultaneously per worker process
    }
  );

  worker.on("completed", (job, result) => {
    console.log(`[${job.id}] Completed in ${job.processedOn! - job.timestamp}ms`);
  });

  worker.on("failed", (job, err) => {
    console.error(`[${job?.id}] Failed (attempt ${job?.attemptsMade}): ${err.message}`);
  });

  return worker;
}

Rate limiting with token buckets

The Anthropic API enforces rate limits on both requests per minute and tokens per minute. A naive worker pool will slam into these limits on any substantial workload. BullMQ's rate limiter handles this:

const worker = new Worker<LLMTask, LLMResult>(
  "llm-tasks",
  processLLMTask,
  {
    connection,
    concurrency: 5,
    limiter: {
      max: 50,        // Maximum 50 jobs per duration
      duration: 60000, // Per 60 seconds
    },
  }
);

For token-level rate limiting (more precise), track tokens in Redis and enforce limits before dispatching:

async function checkTokenBudget(
  estimatedTokens: number,
  redis: Redis
): Promise<boolean> {
  const key = `token_budget:${Math.floor(Date.now() / 60000)}`; // Per-minute bucket
  const current = await redis.incrby(key, estimatedTokens);
  await redis.expire(key, 120);

  const TOKENS_PER_MINUTE_LIMIT = 100_000;
  if (current > TOKENS_PER_MINUTE_LIMIT) {
    // Undo the increment and signal to retry later
    await redis.decrby(key, estimatedTokens);
    return false;
  }
  return true;
}

The API layer

The HTTP API enqueues tasks and retrieves results. Keep it thin — all business logic lives in the worker:

// api.ts
import express from "express";
import { v4 as uuidv4 } from "uuid";
import { llmQueue, LLMTask } from "./queue";

const app = express();
app.use(express.json());

// Submit a task
app.post("/analyse", async (req, res) => {
  const { prompt, systemPrompt, model, maxTokens, metadata } = req.body;

  if (!prompt || typeof prompt !== "string") {
    return res.status(400).json({ error: "prompt is required" });
  }

  const jobId = uuidv4();
  const task: LLMTask = { jobId, prompt, systemPrompt, model, maxTokens, metadata };

  const job = await llmQueue.add("analyse", task, { jobId });

  res.json({
    jobId,
    status: "queued",
    pollUrl: `/result/${jobId}`,
  });
});

// Poll for result
app.get("/result/:jobId", async (req, res) => {
  const job = await llmQueue.getJob(req.params.jobId);

  if (!job) {
    return res.status(404).json({ error: "Job not found" });
  }

  const state = await job.getState();

  if (state === "completed") {
    return res.json({
      status: "completed",
      result: await job.returnvalue,
    });
  }

  if (state === "failed") {
    return res.status(500).json({
      status: "failed",
      error: job.failedReason,
    });
  }

  return res.json({
    status: state, // "waiting", "active", "delayed"
    position: await job.getPosition(),
  });
});

app.listen(3000);

Priority queues for urgent tasks

Not all tasks have equal urgency. BullMQ supports job priorities (lower number = higher priority):

// Regular analysis
await llmQueue.add("analyse", task, { priority: 10 });

// User-facing real-time request
await llmQueue.add("analyse", urgentTask, { priority: 1 });

For larger systems, use separate queues per priority tier with separate worker pools, and allocate more workers to high-priority queues.

Dead letter queues and poison pill handling

Jobs that exhaust their retry budget do not disappear — they move to a failed state. But you need a strategy for what happens next, because some failed jobs represent genuine bugs in your task data (a "poison pill") while others are transient API failures worth retrying manually.

The distinction matters because a true poison pill will fail every time and waste API budget each retry. Detect them early by examining the error pattern across attempts:

worker.on("failed", async (job, err) => {
  if (!job) return;

  const isApiError = err.message.includes("overloaded") ||
                     err.message.includes("rate_limit") ||
                     err.message.includes("529");

  const isDataError = err.message.includes("invalid_request") ||
                      err.message.includes("context_length_exceeded");

  if (isDataError) {
    // Poison pill — move to dead letter queue immediately, don't retry
    await deadLetterQueue.add("failed-task", {
      originalJob: job.data,
      error: err.message,
      failedAt: new Date().toISOString(),
      reason: "invalid_input",
    });
    await job.discard(); // Prevent further retries
  }

  // API errors proceed through normal retry backoff
});

Set up a separate dead letter queue that your team can inspect and triage:

export const deadLetterQueue = new Queue("llm-dead-letter", {
  connection,
  defaultJobOptions: {
    removeOnComplete: false,
    removeOnFail: false,
  },
});

// Periodic review: re-enqueue fixed tasks or discard them
async function retryDeadLetter(jobId: string) {
  const job = await deadLetterQueue.getJob(jobId);
  if (!job) return;

  // Re-add to main queue after fixing the payload
  await llmQueue.add("analyse", job.data.originalJob);
  await job.remove();
}

Instrument the dead letter queue depth in your dashboards. A growing DLQ is almost always a code bug or a systematic input quality problem — not a transient issue.

Webhook callbacks versus polling

Polling works fine for interactive UIs where the user is watching a progress indicator. For backend-to-backend integrations, polling is wasteful — you are making HTTP requests every few seconds when the job could take thirty seconds or more. Webhooks push the result when it is ready.

Add optional webhook support to the job task schema:

export interface LLMTask {
  jobId: string;
  prompt: string;
  systemPrompt?: string;
  model?: string;
  maxTokens?: number;
  metadata?: Record<string, unknown>;
  webhookUrl?: string;   // If set, POST result here on completion
  webhookSecret?: string; // HMAC secret for request verification
}

In the worker's completion handler, fire the webhook if configured:

import crypto from "crypto";
import fetch from "node-fetch";

worker.on("completed", async (job, result: LLMResult) => {
  const { webhookUrl, webhookSecret } = job.data;
  if (!webhookUrl) return;

  const body = JSON.stringify(result);
  const signature = webhookSecret
    ? crypto.createHmac("sha256", webhookSecret).update(body).digest("hex")
    : undefined;

  const headers: Record<string, string> = {
    "Content-Type": "application/json",
  };
  if (signature) {
    headers["X-Signature-SHA256"] = signature;
  }

  try {
    const res = await fetch(webhookUrl, { method: "POST", headers, body });
    if (!res.ok) {
      console.error(`Webhook delivery failed for ${job.id}: ${res.status}`);
    }
  } catch (err) {
    console.error(`Webhook network error for ${job.id}:`, err);
  }
});

The receiving end verifies the signature before trusting the payload — the same pattern GitHub uses for repository webhooks.

Scaling workers horizontally

A single worker process with concurrency of five handles light workloads. As task volume grows, you need more worker processes, typically across multiple servers. Each worker process connects to the same Redis queue and claims jobs atomically — BullMQ handles the coordination.

The practical scaling knobs are:

A typical production deployment runs three to five worker pods in Kubernetes, each with concurrency of four to eight, behind a HorizontalPodAutoscaler that scales on queue depth via a custom metric:

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: llm-worker-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: llm-worker
  minReplicas: 2
  maxReplicas: 20
  metrics:
    - type: External
      external:
        metric:
          name: bullmq_queue_depth
          selector:
            matchLabels:
              queue: llm-tasks
        target:
          type: AverageValue
          averageValue: "10"  # Scale up when each pod has >10 jobs waiting

The scaler adds worker pods when the queue backs up and removes them when it drains. This handles burst workloads without over-provisioning in steady state.

Monitoring and observability

A queue is a black box without instrumentation. At minimum, track:

• Queue depth (how many jobs are waiting)
• Worker throughput (jobs completed per minute)
• Failure rate (jobs failed / total attempted)
• P50/P95/P99 job duration

BullMQ exposes these metrics through its API; pipe them to your monitoring stack:

async function getQueueMetrics() {
  const [waiting, active, completed, failed] = await Promise.all([
    llmQueue.getWaitingCount(),
    llmQueue.getActiveCount(),
    llmQueue.getCompletedCount(),
    llmQueue.getFailedCount(),
  ]);

  return { waiting, active, completed, failed };
}

Alert when queue depth exceeds a threshold (burst your worker count) and when failure rate spikes (LLM API issue or invalid prompts).

Prompt caching to reduce latency and cost

When many tasks share the same system prompt — a common case for document analysis pipelines — the Anthropic API's prompt caching feature lets you cache the system prompt across requests. This reduces input token cost by up to 90% for the cached prefix and lowers time-to-first-token for long system prompts.

Enable caching by marking the system prompt with a cache control block:

async function processLLMTaskWithCache(job: Job<LLMTask>): Promise<LLMResult> {
  const { jobId, prompt, systemPrompt, model, maxTokens } = job.data;

  const response = await client.messages.create({
    model: model || "claude-haiku-4-5",
    max_tokens: maxTokens || 1024,
    system: systemPrompt
      ? [
          {
            type: "text",
            text: systemPrompt,
            cache_control: { type: "ephemeral" }, // Cache this prefix
          },
        ]
      : undefined,
    messages: [{ role: "user", content: prompt }],
  });

  return {
    jobId,
    output: response.content[0].type === "text" ? response.content[0].text : "",
    inputTokens: response.usage.input_tokens,
    outputTokens: response.usage.output_tokens,
    completedAt: new Date().toISOString(),
  };
}

Prompt caching is most valuable when your system prompt is long (over 1,024 tokens) and shared across many requests. For a document analysis pipeline that processes thousands of documents with the same extraction instructions, caching can cut your input token bill by 80% or more. The cache is keyed by the exact text of the cached prefix — even a single character difference misses the cache.

Track cache hit rates by logging response.usage.cache_read_input_tokens alongside your other metrics. Low cache hit rates indicate that system prompts are varying unexpectedly between requests, which is often a bug.

Graceful shutdown and job draining

Worker processes need to shut down cleanly to avoid losing in-flight jobs. A worker killed mid-task leaves the job in "active" state in Redis. BullMQ will eventually reclaim it via the stalled jobs mechanism, but there is a delay and the work is duplicated.

Implement a graceful shutdown handler that stops accepting new jobs and waits for active jobs to complete:

async function gracefulShutdown(worker: Worker) {
  console.log("Shutting down — stopping new job intake");

  // Close the worker: stops taking new jobs, waits for active ones to finish
  await worker.close();

  console.log("All active jobs completed. Worker exited cleanly.");
  process.exit(0);
}

// Handle termination signals from the OS or container orchestrator
process.on("SIGTERM", () => gracefulShutdown(worker));
process.on("SIGINT", () => gracefulShutdown(worker));

In Kubernetes, configure a terminationGracePeriodSeconds that is longer than your longest expected job duration. If an LLM call can take up to 60 seconds, set the grace period to at least 90 seconds. This gives SIGTERM time to reach the process and the running job time to complete before Kubernetes sends SIGKILL.

spec:
  template:
    spec:
      terminationGracePeriodSeconds: 120
      containers:
        - name: llm-worker
          lifecycle:
            preStop:
              exec:
                command: ["/bin/sh", "-c", "sleep 5"]

The preStop sleep gives the load balancer time to remove the pod from rotation before the shutdown begins — a common detail that prevents in-flight HTTP requests from failing during rolling deploys.

Putting it together: a production checklist

Before shipping an async LLM pipeline to production, verify these properties are in place:

• Idempotency: Re-processing the same job twice produces the same result and does not double-write to the result store.
• Job TTL: Completed and failed jobs have a cleanup policy so Redis memory does not grow unbounded.
• Result expiry: The result store evicts old results — clients that never poll should not hold results forever.
• Structured errors: Worker failures surface a machine-readable error code, not just a raw exception message, so the API layer can return meaningful status to callers.
• Load test coverage: Run a synthetic burst at 2x expected peak throughput before launch. Queue systems behave differently under saturation than they do at normal load.
• Runbook: Document how to drain the queue, how to re-enqueue failed jobs, and how to roll back a worker deployment without losing in-flight jobs.

The items on this list are not glamorous, but each one has caused a production incident for someone. The async queue pattern is the foundation for any LLM application that goes beyond toy scale. Get it right early — retrofitting it onto a synchronous codebase is painful.