Batch Processing

When your workflow needs to process many items concurrently — embedding hundreds of document chunks, calling an API for each row in a dataset, or transforming files in bulk — you need to control how many operations run at the same time. promiseAllWithLimit gives you a concurrency limiter that integrates with the Operaide framework.

promiseAllWithLimit

Processes an array of items in parallel while capping the number of concurrent operations.

import { promiseAllWithLimit } from '@operaide/aktor';

const results = await promiseAllWithLimit(
    async (chunk, index) => {
        return await embedChunk(chunk);
    },
    {
        name: 'embeddings',
        items: documentChunks,
        limit: 30,
    }
);

Parameters

Parameter	Type	Default	Description
fn	(item: T, index: number) => Promise<R>	required	Function to execute for each item
name	string	required	Queue identifier
items	T[]	required	Array of items to process
limit	number	required	Maximum concurrent executions
local	boolean	false	Use a local pool instead of the global queue

Results are returned in the original item order, regardless of which operations finish first.

Global vs. Local Mode

Global Queue (default)

All calls with the same name share a single concurrency pool. This is the right choice when you need to respect an external rate limit — for example, an API that allows 30 concurrent requests regardless of how many Reaktors are running.

// Both calls share one pool of 30 slots
const batch1 = promiseAllWithLimit(embedChunk, {
    name: 'embeddings',
    items: chunksFromDoc1,
    limit: 30,
});

const batch2 = promiseAllWithLimit(embedChunk, {
    name: 'embeddings',
    items: chunksFromDoc2,
    limit: 30,
});

await Promise.all([batch1, batch2]);
// Total concurrency across both: max 30

Local Pool

Each call gets its own independent concurrency limit. Use this when the limit is about managing local resource usage rather than a shared external constraint.

const results = await promiseAllWithLimit(
    async (file) => processFile(file),
    {
        name: 'file-processing',
        items: files,
        limit: 10,
        local: true,  // Independent pool for this call
    }
);

Usage in AktorFunctions

promiseAllWithLimit is a plain async function, not an Aktor. Use it inside createAktorFunction to add concurrency control to your Aktor logic:

import { createAktorFunction, promiseAllWithLimit } from '@operaide/aktor';

const aktorEmbedAllChunks = createAktorFunction('aktorEmbedAllChunks',
    async ({ chunks }: { chunks: string[] }) => {
        if (chunks.length === 0) return [];

        const embeddings = await promiseAllWithLimit(
            async (chunk) => {
                const response = await fetch('/api/embed', {
                    method: 'POST',
                    body: JSON.stringify({ text: chunk }),
                });
                return response.json();
            },
            {
                name: 'embedding-api',
                items: chunks,
                limit: 20,
            }
        );

        return embeddings;
    }
);