如何用 BullMQ 在 Node.js 建立 AI 自動化流程並設定佇列、重試及速率限制

Most AI automation demos are one await away from breaking. A webhook comes in, the handler calls a model, the response gets written to the database, and everyone is happy until the provider returns a 429, the model wraps its JSON in a code fence, or the deploy restarts the process halfway through a batch.

This post walks through the pattern I use to move LLM calls out of request handlers and into a queue, using BullMQ on Redis with Node.js and TypeScript. The example is a support-ticket triage pipeline: each ticket goes to a model, comes back as a category, a priority and a summary, and gets saved. The same shape works for lead enrichment, document extraction, content moderation or any other "call a model for every row" job.

Everything below was type-checked against BullMQ 6 and run against a local Redis with a fake model that returns a 429, then a 503, then invalid JSON, then a valid answer.

Why an AI pipeline needs a queue

An LLM call is slower and less predictable than almost anything else your backend does. Four problems show up quickly:

1. Latency. A call can take seconds. Holding an HTTP request open for that long ties up the caller and times out behind proxies.
2. Rate limits. Providers cap requests and tokens per minute. Ten parallel webhooks can trip a limit that one would not.
3. Transient failures. 429s, 5xx responses and dropped connections happen, and most of them succeed on a second try.
4. Bad output. Even with JSON mode, a response can fail your schema. That is a failed attempt, not a reason to crash.

A queue answers all four. The handler adds a job and returns 202 Accepted. Workers pull jobs at a controlled rate, retry with backoff, and park anything hopeless in a dead-letter queue where a human can look at it.

The architecture

webhook ──► triageQueue (Redis) ──► Worker (concurrency 8, 50 req/min)
                                       │
                                       ├─ ok ───────────► saveTriage() upsert
                                       ├─ retryable ────► delayed, custom backoff
                                       └─ hopeless ─────► ticket-triage-dlq

Five small files: a Redis connection, the schemas, the model client, the queue, and the worker.

1. The Redis connection

// connection.ts
import { Redis } from 'ioredis';

// Workers use blocking Redis commands, so ioredis must not give up on them.
export const connection = new Redis(process.env.REDIS_URL ?? 'redis://127.0.0.1:6379', {
  maxRetriesPerRequest: null,
});

maxRetriesPerRequest: null matters. BullMQ workers block on Redis while waiting for jobs, and the ioredis default would abort those commands after a few retries during a Redis blip.

2. Schemas for the input and the model output

// schema.ts
import { z } from 'zod';

export const TicketJob = z.object({
  ticketId: z.string(),
  subject: z.string(),
  body: z.string().max(20_000),
});
export type TicketJob = z.infer<typeof TicketJob>;

// What we accept back from the model. Anything else is a failed attempt.
export const Triage = z.object({
  category: z.enum(['billing', 'bug', 'feature_request', 'account', 'other']),
  priority: z.enum(['low', 'normal', 'high', 'urgent']),
  summary: z.string().min(10).max(400),
});
export type Triage = z.infer<typeof Triage>;

Validating the job payload protects you from old jobs enqueued by an older version of the code. Validating the model output is the real point: an enum for category means downstream code never sees "Billing issue" or "payments?".

3. A model client that reports useful errors

// errors.ts
export class ModelHttpError extends Error {
  constructor(public status: number, message: string, public retryAfterMs?: number) {
    super(message);
    this.name = 'ModelHttpError';
  }
}

// Retry only what can succeed on a second try.
export function isRetryable(err: unknown): boolean {
  if (err instanceof ModelHttpError) {
    return err.status === 408 || err.status === 429 || err.status >= 500;
  }
  return err instanceof Error && /ECONN|ETIMEDOUT|fetch failed|aborted/i.test(err.message);
}

// model.ts
import { ModelHttpError } from './errors.js';

// Any OpenAI-compatible chat completions endpoint works here.
export async function callModel(prompt: string, signal: AbortSignal): Promise<string> {
  const res = await fetch(`${process.env.LLM_BASE_URL}/chat/completions`, {
    method: 'POST',
    signal,
    headers: {
      'content-type': 'application/json',
      authorization: `Bearer ${process.env.LLM_API_KEY}`,
    },
    body: JSON.stringify({
      model: process.env.LLM_MODEL,
      response_format: { type: 'json_object' },
      messages: [
        { role: 'system', content: 'You triage support tickets. Reply with JSON only.' },
        { role: 'user', content: prompt },
      ],
    }),
  });

  if (!res.ok) {
    const retryAfter = Number(res.headers.get('retry-after'));
    throw new ModelHttpError(
      res.status,
      `model returned ${res.status}`,
      Number.isFinite(retryAfter) && retryAfter > 0 ? retryAfter * 1000 : undefined,
    );
  }
  const json = (await res.json()) as { choices: { message: { content: string } }[] };
  return json.choices[0].message.content;
}

The important part is that the error carries the HTTP status and the Retry-After value. The worker needs both to decide whether and when to try again. (Retry-After can also be an HTTP date; this version only handles the seconds form, which is what model APIs usually send.)

4. The queue, with idempotent enqueueing

// queue.ts
import { Queue } from 'bullmq';
import { connection } from './connection.js';
import type { TicketJob } from './schema.js';

export const triageQueue = new Queue<TicketJob>('ticket-triage', {
  connection,
  defaultJobOptions: {
    attempts: 6,
    backoff: { type: 'llm' }, // custom strategy, defined on the worker
    removeOnComplete: { age: 24 * 3600, count: 1000 },
    removeOnFail: { age: 7 * 24 * 3600 },
  },
});

export const deadLetterQueue = new Queue('ticket-triage-dlq', { connection });

export async function enqueueTicket(ticket: TicketJob) {
  // The ticket id doubles as the job id, so a webhook delivered twice
  // does not create two jobs while the first one still exists.
  return triageQueue.add('triage', ticket, { jobId: `ticket-${ticket.ticketId}` });
}

Webhook providers retry deliveries, so duplicates are normal. With a custom jobId, BullMQ ignores an add for an id that already exists. In my test, enqueueing the same ticket twice returned the same job id and the queue held one job.

One caveat: once a job is removed by removeOnComplete, the same id can be added again. That is why the save step below is an upsert as well. Deduplication in the queue cuts wasted model calls; idempotency in the database guarantees correctness.

removeOnComplete and removeOnFail keep Redis from growing forever. Keep failed jobs longer than completed ones, because they are the ones you will want to inspect.

5. The worker: retries, backoff, rate limits

// worker.ts
import { Worker, UnrecoverableError, type Job } from 'bullmq';
import { connection } from './connection.js';
import { ModelHttpError, isRetryable } from './errors.js';
import { callModel } from './model.js';
import { TicketJob, Triage } from './schema.js';
import { deadLetterQueue } from './queue.js';
import { saveTriage } from './store.js';

const BASE_MS = 2_000;
const MAX_MS = 5 * 60_000;

function llmBackoff(attemptsMade: number, _type?: string, err?: Error): number {
  if (err instanceof ModelHttpError && err.retryAfterMs) {
    return Math.min(err.retryAfterMs, MAX_MS); // the provider told us when
  }
  const ceiling = Math.min(MAX_MS, BASE_MS * 2 ** (attemptsMade - 1));
  return Math.round(Math.random() * ceiling); // full jitter
}

async function processTicket(job: Job<TicketJob>): Promise<Triage> {
  const ticket = TicketJob.parse(job.data);

  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), 30_000);
  let raw: string;
  try {
    raw = await callModel(
      `Subject: ${ticket.subject}\n\n${ticket.body}\n\nReturn {"category","priority","summary"}.`,
      controller.signal,
    );
  } catch (err) {
    if (!isRetryable(err)) {
      throw new UnrecoverableError(err instanceof Error ? err.message : String(err));
    }
    throw err;
  } finally {
    clearTimeout(timer);
  }

  let parsed: unknown;
  try {
    parsed = JSON.parse(raw);
  } catch {
    throw new Error('model returned invalid JSON'); // retryable: sampling varies
  }
  const result = Triage.safeParse(parsed);
  if (!result.success) {
    throw new Error(`schema mismatch: ${result.error.issues[0]?.message}`);
  }

  await job.updateProgress(90);
  await saveTriage(ticket.ticketId, result.data); // idempotent upsert
  return result.data;
}

export const worker = new Worker<TicketJob, Triage>('ticket-triage', processTicket, {
  connection,
  concurrency: 8,
  limiter: { max: 50, duration: 60_000 }, // stay under the provider's RPM
  settings: { backoffStrategy: llmBackoff },
});

A few decisions worth calling out.

Classify every error. A 400 or 401 will fail the same way six times, so the worker wraps it in UnrecoverableError, and BullMQ skips the remaining attempts. A 429, a 5xx or a network reset is thrown as a normal error and retried.

Respect Retry-After, otherwise use full jitter. When the provider says "wait one second", the custom strategy waits one second. Otherwise it picks a random delay between zero and an exponentially growing ceiling, capped at five minutes. The randomness matters. Without it, every job that hit the same 429 retries at the same instant and hits it again. BullMQ calls a custom backoffStrategy when the job's backoff type is not one of the built-in fixed or exponential types, which is why the queue sets type: 'llm'. Returning -1 from the strategy stops retrying.

Treat bad output as a retryable failure. Invalid JSON or a schema mismatch is thrown as a plain error. A second sample often comes back clean. If it keeps failing, the job runs out of attempts and lands in the dead-letter queue instead of writing garbage to the database.

Rate-limit at the worker, not in your code. limiter: { max: 50, duration: 60_000 } caps processing at 50 jobs per minute across every worker on the same queue, because the limit is tracked in Redis. Set it a little below your provider's limit. concurrency: 8 controls how many calls one process has in flight.

Always set a timeout. A hung request holds a concurrency slot forever. The AbortController turns it into an aborted error after 30 seconds, which isRetryable treats as transient.

6. A dead-letter queue for jobs that never succeed

worker.on('failed', async (job, err) => {
  if (!job) return;
  const outOfAttempts = job.attemptsMade >= (job.opts.attempts ?? 1);
  if (outOfAttempts || err.name === 'UnrecoverableError') {
    await deadLetterQueue.add('dead', {
      jobId: job.id,
      data: job.data,
      reason: err.message,
      attemptsMade: job.attemptsMade,
    });
  }
});

The worker's failed event fires on every failed attempt, including ones that will be retried, so the handler checks whether this was the last one. Only final failures go to the DLQ. From there you can alert, show them in an admin page, or re-enqueue them after fixing a prompt.

What the test run showed

I ran the pipeline against a local Redis with fetch replaced by a scripted fake:

• Ticket 1 got a 429 with Retry-After: 1, then a 503, then a response that was not JSON, then a valid answer. It succeeded on the fourth attempt, about nine seconds after it was enqueued, and was saved once.
• Ticket 2 got a 400. It failed after a single call, with no retries, and landed in the dead-letter queue with the reason model returned 400.
• Enqueueing ticket 1 twice produced one job.

That is the behavior you want in production: transient errors absorbed, permanent errors surfaced quickly, and no duplicate work.

Production checklist

• Run workers as a separate process from your web server so a slow batch never affects request latency.
• Call await worker.close() on SIGTERM so in-flight jobs finish before a deploy kills the process.
• Put Retry-After handling and error classification in one place, like isRetryable, and test it.
• Log job.id, attemptsMade and the error on every failure. Track DLQ size as a metric.
• Make every side effect idempotent: database upserts keyed by a business id, and an idempotency key on any external API that supports one.
• Version your prompts, and store the prompt version with each result, so you can re-run a slice of jobs after a prompt change.

Wrapping up

The model call is the easy part of AI automation. The work is everything around it: deciding what to retry, how long to wait, how fast to go, and where failures end up. A queue such as BullMQ gives you those controls for the cost of a Redis instance and a few hundred lines of TypeScript.

If you are building agents rather than batch pipelines, my posts on MCP vs API and building an MCP server in Next.js cover the other side of the same problem.

Originally published on fabwaseem.com.