Parallelising LLM calls for batch document processing

The serial bottleneck in document pipelines

Processing a thousand documents through an LLM serially is straightforward to implement and catastrophically slow. A single Claude API call for a 2000-token document takes roughly 3-8 seconds depending on output length. Processing 1000 documents serially therefore takes somewhere between 50 minutes and 2+ hours. Most production workloads cannot tolerate that.

The fix is parallel execution, but naive parallelism creates its own problems: rate-limit errors, runaway costs, memory exhaustion, and partial failures that are hard to recover from. Building a robust parallel pipeline means thinking carefully about concurrency controls, back-pressure, retry semantics, and observability.

This article goes through each of those concerns with concrete TypeScript implementations. The patterns apply to any workload where you process a collection of documents — summarisation, classification, extraction, translation — through one or more LLM calls.

Understanding API rate limits before you parallelize

Before tuning concurrency, understand what limits you're actually operating under. The Anthropic API enforces several independent limits:

• Requests per minute (RPM): How many API calls you can make per minute
• Tokens per minute (TPM): How many total input + output tokens per minute
• Tokens per day (TPD): Daily cap, relevant for lower tiers

These limits vary by tier. If your workload is token-bound (long documents, long outputs), you'll hit TPM before RPM. If it's request-bound (many short documents), you'll hit RPM first. Knowing which bottleneck applies tells you whether to limit concurrency or batch documents together into single prompts.

You can check your current limits in the Anthropic console. For programmatic rate-limit awareness, watch for 429 responses with the retry-after header:

interface RateLimitInfo {
  requestsLimit: number;
  requestsRemaining: number;
  requestsReset: Date;
  tokensLimit: number;
  tokensRemaining: number;
  tokensReset: Date;
}

function parseRateLimitHeaders(headers: Headers): RateLimitInfo | null {
  const requestsLimit = headers.get("anthropic-ratelimit-requests-limit");
  if (!requestsLimit) return null;

  return {
    requestsLimit: parseInt(requestsLimit),
    requestsRemaining: parseInt(
      headers.get("anthropic-ratelimit-requests-remaining") ?? "0"
    ),
    requestsReset: new Date(
      headers.get("anthropic-ratelimit-requests-reset") ?? Date.now()
    ),
    tokensLimit: parseInt(
      headers.get("anthropic-ratelimit-tokens-limit") ?? "0"
    ),
    tokensRemaining: parseInt(
      headers.get("anthropic-ratelimit-tokens-remaining") ?? "0"
    ),
    tokensReset: new Date(
      headers.get("anthropic-ratelimit-tokens-reset") ?? Date.now()
    ),
  };
}

Concurrency limiting with a semaphore

The simplest parallelism pattern is a concurrency-limited Promise pool. You maintain a maximum number of in-flight requests and queue new ones as existing ones complete.

class Semaphore {
  private permits: number;
  private queue: Array<() => void> = [];

  constructor(permits: number) {
    this.permits = permits;
  }

  async acquire(): Promise<void> {
    if (this.permits > 0) {
      this.permits--;
      return;
    }
    await new Promise<void>((resolve) => this.queue.push(resolve));
  }

  release(): void {
    const next = this.queue.shift();
    if (next) {
      next();
    } else {
      this.permits++;
    }
  }

  async run<T>(fn: () => Promise<T>): Promise<T> {
    await this.acquire();
    try {
      return await fn();
    } finally {
      this.release();
    }
  }
}

// Usage
const limiter = new Semaphore(10); // max 10 concurrent requests

async function processDocument(doc: string): Promise<string> {
  return limiter.run(async () => {
    const response = await client.messages.create({
      model: "claude-haiku-4-5",
      max_tokens: 512,
      messages: [{ role: "user", content: `Summarise: ${doc}` }],
    });
    return response.content[0].type === "text" ? response.content[0].text : "";
  });
}

const results = await Promise.all(documents.map(processDocument));

How many concurrent requests should you allow? A reasonable starting point is 10-20 for document workloads on a production Anthropic tier. Monitor your actual rate-limit headers to tune this value. If you're regularly seeing near-zero tokens-remaining before the reset window, lower concurrency or increase batching.

Token-aware batching for short documents

If your documents are short (under 500 tokens each), you're paying per-request overhead for every call. A better strategy is to pack multiple documents into a single API call:

interface BatchItem {
  id: string;
  content: string;
}

interface BatchResult {
  id: string;
  output: string;
  error?: string;
}

async function processBatch(items: BatchItem[]): Promise<BatchResult[]> {
  const numberedItems = items
    .map((item, i) => `[${i + 1}] ${item.id}\n${item.content}`)
    .join("\n\n---\n\n");

  const response = await client.messages.create({
    model: "claude-haiku-4-5",
    max_tokens: items.length * 200, // budget per item
    messages: [
      {
        role: "user",
        content: `Classify each of the following texts as POSITIVE, NEGATIVE, or NEUTRAL sentiment. 
For each, respond with just the item number and classification.

${numberedItems}

Format: [N] CLASSIFICATION`,
      },
    ],
  });

  const text =
    response.content[0].type === "text" ? response.content[0].text : "";
  const lines = text.trim().split("\n");

  return items.map((item, i) => {
    const line = lines.find((l) => l.startsWith(`[${i + 1}]`));
    if (!line) return { id: item.id, output: "", error: "Not found in output" };
    const classification = line.replace(`[${i + 1}]`, "").trim();
    return { id: item.id, output: classification };
  });
}

function chunkArray<T>(arr: T[], size: number): T[][] {
  const chunks: T[][] = [];
  for (let i = 0; i < arr.length; i += size) {
    chunks.push(arr.slice(i, i + size));
  }
  return chunks;
}

async function classifyDocuments(items: BatchItem[]): Promise<BatchResult[]> {
  const BATCH_SIZE = 20;
  const limiter = new Semaphore(5);
  const batches = chunkArray(items, BATCH_SIZE);

  const batchResults = await Promise.all(
    batches.map((batch) => limiter.run(() => processBatch(batch)))
  );

  return batchResults.flat();
}

Batching 20 short documents into one request reduces your request count by 20x, dramatically improving throughput against RPM limits. The tradeoff is that a single API error affects 20 documents rather than one, so you need reliable parsing of the batched output.

Retry logic with exponential backoff

Rate-limit errors (429) and transient server errors (529, 500) are inevitable at scale. Your retry logic needs to handle them without thundering-herd effects:

interface RetryOptions {
  maxRetries: number;
  baseDelayMs: number;
  maxDelayMs: number;
  jitterFactor: number;
}

const DEFAULT_RETRY_OPTIONS: RetryOptions = {
  maxRetries: 5,
  baseDelayMs: 1000,
  maxDelayMs: 60000,
  jitterFactor: 0.2,
};

async function withRetry<T>(
  fn: () => Promise<T>,
  options: Partial<RetryOptions> = {}
): Promise<T> {
  const opts = { ...DEFAULT_RETRY_OPTIONS, ...options };

  for (let attempt = 0; attempt <= opts.maxRetries; attempt++) {
    try {
      return await fn();
    } catch (err: any) {
      const isRetryable =
        err?.status === 429 ||
        err?.status === 529 ||
        (err?.status >= 500 && err?.status < 600);

      if (!isRetryable || attempt === opts.maxRetries) {
        throw err;
      }

      // If the API provides a retry-after header, respect it
      const retryAfter = err?.headers?.["retry-after"];
      let delay: number;

      if (retryAfter) {
        delay = parseFloat(retryAfter) * 1000;
      } else {
        const exponential = opts.baseDelayMs * 2 ** attempt;
        const capped = Math.min(exponential, opts.maxDelayMs);
        const jitter = capped * opts.jitterFactor * (Math.random() * 2 - 1);
        delay = capped + jitter;
      }

      console.warn(
        `Attempt ${attempt + 1} failed (${err?.status}). Retrying in ${Math.round(delay)}ms...`
      );
      await new Promise((r) => setTimeout(r, delay));
    }
  }

  throw new Error("Unreachable");
}

// Wrap your API calls
async function summariseWithRetry(document: string): Promise<string> {
  return withRetry(() =>
    client.messages.create({
      model: "claude-haiku-4-5",
      max_tokens: 256,
      messages: [{ role: "user", content: `Summarise in 2 sentences: ${document}` }],
    }).then((r) => (r.content[0].type === "text" ? r.content[0].text : ""))
  );
}

The jitter factor is important: if 10 concurrent requests all hit a rate limit at the same moment and retry with the same delay, they'll all hammer the API again simultaneously. Adding random jitter spreads the retries over time.

Progress tracking and partial failure handling

When processing thousands of documents, you need visibility into progress and the ability to resume from where you left off if the process crashes.

interface ProcessingState {
  total: number;
  completed: number;
  failed: number;
  results: Map<string, BatchResult>;
  errors: Map<string, Error>;
}

class DocumentProcessor {
  private state: ProcessingState;
  private limiter: Semaphore;

  constructor(concurrency: number) {
    this.limiter = new Semaphore(concurrency);
    this.state = {
      total: 0,
      completed: 0,
      failed: 0,
      results: new Map(),
      errors: new Map(),
    };
  }

  async processAll(
    items: BatchItem[],
    processor: (item: BatchItem) => Promise<string>,
    onProgress?: (state: ProcessingState) => void
  ): Promise<ProcessingState> {
    this.state.total = items.length;

    const tasks = items.map((item) =>
      this.limiter.run(async () => {
        try {
          const output = await withRetry(() => processor(item));
          this.state.results.set(item.id, { id: item.id, output });
          this.state.completed++;
        } catch (err: any) {
          this.state.errors.set(item.id, err);
          this.state.failed++;
          console.error(`Failed to process ${item.id}: ${err.message}`);
        }

        onProgress?.(this.state);
      })
    );

    await Promise.all(tasks);
    return this.state;
  }

  getFailedItems(originalItems: BatchItem[]): BatchItem[] {
    return originalItems.filter((item) => this.state.errors.has(item.id));
  }
}

// Example usage with progress logging
const processor = new DocumentProcessor(15);

const finalState = await processor.processAll(
  documents,
  async (item) => {
    const response = await client.messages.create({
      model: "claude-haiku-4-5",
      max_tokens: 256,
      messages: [
        { role: "user", content: `Extract key topics from: ${item.content}` },
      ],
    });
    return response.content[0].type === "text" ? response.content[0].text : "";
  },
  (state) => {
    const pct = ((state.completed + state.failed) / state.total * 100).toFixed(1);
    process.stdout.write(`\r${pct}% (${state.completed} ok, ${state.failed} failed)`);
  }
);

// Retry failed items with lower concurrency
if (finalState.failed > 0) {
  const failedItems = processor.getFailedItems(documents);
  console.log(`\nRetrying ${failedItems.length} failed items...`);
  const retryProcessor = new DocumentProcessor(3);
  await retryProcessor.processAll(failedItems, /* same processor */);
}

Checkpointing for long-running jobs

For jobs that run for hours, you need checkpoint-based resumption. If the process crashes at document 850 of 10,000, you want to restart from 850, not from the beginning.

import * as fs from "fs/promises";
import * as path from "path";

class CheckpointedProcessor {
  private checkpointPath: string;
  private completedIds: Set<string>;

  constructor(jobId: string) {
    this.checkpointPath = path.join("/tmp", `checkpoint-${jobId}.json`);
    this.completedIds = new Set();
  }

  async loadCheckpoint(): Promise<void> {
    try {
      const data = await fs.readFile(this.checkpointPath, "utf8");
      const checkpoint = JSON.parse(data);
      this.completedIds = new Set(checkpoint.completedIds);
      console.log(`Resumed from checkpoint: ${this.completedIds.size} already completed`);
    } catch {
      // No checkpoint exists, starting fresh
    }
  }

  async saveCheckpoint(): Promise<void> {
    await fs.writeFile(
      this.checkpointPath,
      JSON.stringify({ completedIds: Array.from(this.completedIds) })
    );
  }

  filterPending(items: BatchItem[]): BatchItem[] {
    return items.filter((item) => !this.completedIds.has(item.id));
  }

  markComplete(id: string): void {
    this.completedIds.add(id);
  }
}

async function processWithCheckpointing(
  jobId: string,
  items: BatchItem[]
): Promise<void> {
  const checkpoint = new CheckpointedProcessor(jobId);
  await checkpoint.loadCheckpoint();

  const pending = checkpoint.filterPending(items);
  console.log(`Processing ${pending.length} remaining items (${items.length - pending.length} already done)`);

  const limiter = new Semaphore(10);
  let saveCounter = 0;

  await Promise.all(
    pending.map((item) =>
      limiter.run(async () => {
        try {
          await withRetry(async () => {
            const response = await client.messages.create({
              model: "claude-haiku-4-5",
              max_tokens: 256,
              messages: [{ role: "user", content: item.content }],
            });
            // persist result to your database here
            return response;
          });

          checkpoint.markComplete(item.id);

          // Save checkpoint every 50 completions
          if (++saveCounter % 50 === 0) {
            await checkpoint.saveCheckpoint();
          }
        } catch (err) {
          console.error(`Permanently failed: ${item.id}`);
        }
      })
    )
  );

  await checkpoint.saveCheckpoint();
}

Cost-aware scheduling: off-peak processing for non-urgent jobs

If your batch job isn't time-sensitive, schedule it during periods when your rate-limit quotas have reset and API latency is typically lower. For jobs that need to complete within a day, distributing load evenly over 24 hours rather than hammering the API for one hour is both cheaper (fewer retries) and more reliable.

For very large batches (millions of documents), the Anthropic Batch API provides a more efficient pathway: you submit a JSONL file of requests and retrieve results asynchronously within 24 hours at reduced pricing. This is ideal for offline enrichment workloads where latency doesn't matter:

import Anthropic from "@anthropic-ai/sdk";
import * as fs from "fs/promises";

async function submitBatchJob(items: BatchItem[]): Promise<string> {
  const client = new Anthropic();

  const requests = items.map((item) => ({
    custom_id: item.id,
    params: {
      model: "claude-haiku-4-5",
      max_tokens: 256,
      messages: [
        {
          role: "user" as const,
          content: `Classify sentiment: ${item.content}`,
        },
      ],
    },
  }));

  const batch = await client.beta.messages.batches.create({
    requests,
  });

  console.log(`Batch submitted: ${batch.id}`);
  return batch.id;
}

async function pollBatchUntilComplete(batchId: string): Promise<void> {
  const client = new Anthropic();

  while (true) {
    const batch = await client.beta.messages.batches.retrieve(batchId);

    if (batch.processing_status === "ended") {
      console.log(
        `Batch complete: ${batch.request_counts.succeeded} succeeded, ${batch.request_counts.errored} errored`
      );
      return;
    }

    console.log(
      `Batch status: ${batch.processing_status} - ${batch.request_counts.processing} in progress`
    );
    await new Promise((r) => setTimeout(r, 30000)); // poll every 30s
  }
}

The Batch API removes rate-limit concerns entirely — you submit everything at once and Anthropic handles the scheduling. The tradeoff is latency: results may take up to 24 hours. For nightly enrichment jobs, data preprocessing pipelines, or content moderation queues, that's a worthwhile tradeoff for 50% reduced pricing.

Measuring throughput: the metrics that matter

Once your pipeline is running in parallel, track these metrics to understand performance and identify bottlenecks:

• Effective throughput: documents per minute actually completed
• API utilisation: tokens consumed vs. your TPM limit (want to stay at 70-80%)
• Error rate: percentage of requests that hit rate limits or permanent failures
• P99 latency: the long tail, often caused by long documents or overloaded API endpoints
• Retry overhead: percentage of successful completions that required at least one retry

A well-tuned pipeline should operate at 75-85% of your rate limit capacity, have an error rate under 2% (all retried successfully), and complete batches in predictable time windows. If your throughput is well below your rate limit, you're under-utilizing available concurrency. If your error rate is high, you're overrunning your limits or documents are taking too long.

Building in these concerns from the start — concurrency control, retry logic, checkpointing, and observability — means the jump from a prototype that processes 10 documents to a production pipeline processing 100,000 is a configuration change, not an architectural rewrite.