Distributed Job Queue

advanced15 minworker-queue.ts

Agent runs should be queued, retried and processed by a pool of workers, with metrics for Prometheus.

What it shows

Enqueue serialized agents with JobQueue (priority, delay, retries)
Process them with WorkerPool
Export queue metrics with formatPrometheusMetrics()

The code

worker-queue.ts

import { Cogitator } from '@cogitator-ai/core';
import { JobQueue, WorkerPool, formatPrometheusMetrics, type SerializedAgent } from '@cogitator-ai/worker';

const apiKey = process.env.GOOGLE_API_KEY;
if (!apiKey) throw new Error('Set GOOGLE_API_KEY');

const redisUrl = new URL(process.env.REDIS_URL ?? 'redis://localhost:6379');
const redis = {
  host: redisUrl.hostname,
  port: Number(redisUrl.port || 6379),
  password: redisUrl.password || undefined,
};

const queue = new JobQueue({
  name: 'agent-jobs',
  redis,
  defaultJobOptions: {
    attempts: 3,
    backoff: { type: 'exponential', delay: 1000 },
    removeOnComplete: 50,
    removeOnFail: 100,
  },
});

const finished = new Set<string>();
const pool = new WorkerPool(
  {
    name: 'agent-jobs',
    redis,
    workerCount: 2,
    concurrency: 3,
    cogitator: new Cogitator({ llm: { providers: { google: { apiKey } } } }),
  },
  {
    onJobCompleted: (jobId, result) => {
      console.log(`done ${jobId}:`, result.type === 'agent' ? result.output : result.type);
      finished.add(jobId);
    },
    onJobFailed: (jobId, error) => {
      console.log(`failed ${jobId}: ${error.message}`);
      finished.add(jobId);
    },
  }
);
await pool.start();

const summarizer: SerializedAgent = {
  name: 'summarizer',
  instructions: 'Summarize the given text in one sentence.',
  model: 'google/gemini-3.5-flash-lite',
  provider: 'google',
  temperature: 0.3,
  tools: [],
};

const jobs = await Promise.all([
  queue.addAgentJob(summarizer, 'AI is transforming industries from healthcare to finance.', { priority: 1 }),
  queue.addAgentJob(summarizer, 'Machine learning needs data, tuning and evaluation.', { delay: 1000 }),
]);

console.log(formatPrometheusMetrics(await queue.getMetrics(), { queue: 'agent-jobs' }));

const deadline = Date.now() + 60_000;
while (jobs.some((job) => job.id && !finished.has(job.id)) && Date.now() < deadline) {
  await new Promise((resolve) => setTimeout(resolve, 250));
}

await pool.stop();
await queue.close();

Run it

install · bash

pnpm add @cogitator-ai/core @cogitator-ai/worker ioredis
docker run -d --name redis -p 6379:6379 redis:7-alpine

EnvironmentGOOGLE_API_KEYREDIS_URL=redis://localhost:6379

run · bash

GOOGLE_API_KEY=your-key REDIS_URL=redis://localhost:6379 npx tsx worker-queue.ts

Or run the full example from a clone of the repository (keys in .env):

repository · bash

npx tsx examples/infrastructure/03-worker-queue.ts

Distributed Job Queue

What it shows

The code

Run it

Learn more